@Test public void testCommit() throws ProducerException, InterruptedException, IOException { SlsRecordResolver<Row> serializationSchema = Mockito.mock(SlsRecordResolver.class); SlsOutputFormat<Row> outputFormat = new SlsOutputFormat( "", "", "", "test_project", "test_store", serializationSchema); LogProducer producer = Mockito.mock(LogProducer.class); LogProducerProvider producerProvider = Mockito.mock(LogProducerProvider.class); Mockito.when(producerProvider.getClient()).thenReturn(producer); SettableFuture future = SettableFuture.create(); Mockito.when( producer.send( Mockito.eq("test_project"), Mockito.eq("test_store"), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyList())).thenReturn(future); ExecutorService executor = Executors.newSingleThreadExecutor(); Whitebox.setInternalState(outputFormat, "logProducerProvider", producerProvider); Whitebox.setInternalState( outputFormat, "sendFutureCallback", outputFormat.new SendFutureCallback()); Whitebox.setInternalState(outputFormat, "executor", executor); Whitebox.setInternalState(outputFormat, "latencyGauge", Mockito.mock(MetricUtils.LatencyGauge.class)); Whitebox.setInternalState(outputFormat, "outTps", Mockito.mock(Meter.class)); Row record = new Row(3); record.setField(0, 100); record.setField(1, 1000); record.setField(2, "test"); outputFormat.writeRecord(record); AtomicLong numSent = (AtomicLong) Whitebox.getInternalState(outputFormat, "numSent"); AtomicLong numCommitted = (AtomicLong) Whitebox.getInternalState(outputFormat, "numCommitted"); assertEquals(1, numSent.get()); assertEquals(0, numCommitted.get()); future.set(new Result(true, null, 0)); executor.awaitTermination(1, TimeUnit.SECONDS); assertEquals(1, numSent.get()); assertEquals(1, numCommitted.get()); }
@Override public void writeRecord(T row) throws IOException { if (null != row) { long start = System.currentTimeMillis(); List<LogItem> tmpLogGroup = new ArrayList<>(1); tmpLogGroup.add(serializationSchema.getLogItem(row)); String partitionKey = serializationSchema.getPartitionKey(row); String topic = serializationSchema.getTopic(row); String source = serializationSchema.getSource(row); try { ListenableFuture<Result> future = logProducerProvider.getClient().send( this.projectName, this.logstore, topic, source, partitionKey, tmpLogGroup); Futures.addCallback(future, sendFutureCallback, executor); numSent.incrementAndGet(); } catch (InterruptedException | ProducerException e) { callBackException = new RuntimeException(e); } if (null != callBackException) { LOG.warn("Fail in write to SLS", callBackException); if (failOnError) { throw callBackException; } callBackException = null; } long end = System.currentTimeMillis(); latencyGauge.report(end - start, 1); outTps.markEvent(); } }
SlsOutputFormat extends RichOutputFormat<T> implements Syncable,
		HasRetryTimeout { @Override public void writeRecord(T row) throws IOException { if (null != row) { long start = System.currentTimeMillis(); List<LogItem> tmpLogGroup = new ArrayList<>(1); tmpLogGroup.add(serializationSchema.getLogItem(row)); String partitionKey = serializationSchema.getPartitionKey(row); String topic = serializationSchema.getTopic(row); String source = serializationSchema.getSource(row); try { ListenableFuture<Result> future = logProducerProvider.getClient().send( this.projectName, this.logstore, topic, source, partitionKey, tmpLogGroup); Futures.addCallback(future, sendFutureCallback, executor); numSent.incrementAndGet(); } catch (InterruptedException | ProducerException e) { callBackException = new RuntimeException(e); } if (null != callBackException) { LOG.warn("Fail in write to SLS", callBackException); if (failOnError) { throw callBackException; } callBackException = null; } long end = System.currentTimeMillis(); latencyGauge.report(end - start, 1); outTps.markEvent(); } } }
SlsOutputFormat extends RichOutputFormat<T> implements Syncable,
		HasRetryTimeout { @Override public void writeRecord(T row) throws IOException { if (null != row) { long start = System.currentTimeMillis(); List<LogItem> tmpLogGroup = new ArrayList<>(1); tmpLogGroup.add(serializationSchema.getLogItem(row)); String partitionKey = serializationSchema.getPartitionKey(row); String topic = serializationSchema.getTopic(row); String source = serializationSchema.getSource(row); try { ListenableFuture<Result> future = logProducerProvider.getClient().send( this.projectName, this.logstore, topic, source, partitionKey, tmpLogGroup); Futures.addCallback(future, sendFutureCallback, executor); numSent.incrementAndGet(); } catch (InterruptedException | ProducerException e) { callBackException = new RuntimeException(e); } if (null != callBackException) { LOG.warn("Fail in write to SLS", callBackException); if (failOnError) { throw callBackException; } callBackException = null; } long end = System.currentTimeMillis(); latencyGauge.report(end - start, 1); outTps.markEvent(); } } SlsOutputFormat(
			String endPoint, String accessKeyId, String accessKey,
			String projectName, String logstoreName,
			SlsRecordResolver<T> serializationSchema);  SlsOutputFormat(
			String endPoint, Configuration properties,
			String projectName, String logstoreName,
			SlsRecordResolver<T> serializationSchema); }
SlsOutputFormat extends RichOutputFormat<T> implements Syncable,
		HasRetryTimeout { @Override public void writeRecord(T row) throws IOException { if (null != row) { long start = System.currentTimeMillis(); List<LogItem> tmpLogGroup = new ArrayList<>(1); tmpLogGroup.add(serializationSchema.getLogItem(row)); String partitionKey = serializationSchema.getPartitionKey(row); String topic = serializationSchema.getTopic(row); String source = serializationSchema.getSource(row); try { ListenableFuture<Result> future = logProducerProvider.getClient().send( this.projectName, this.logstore, topic, source, partitionKey, tmpLogGroup); Futures.addCallback(future, sendFutureCallback, executor); numSent.incrementAndGet(); } catch (InterruptedException | ProducerException e) { callBackException = new RuntimeException(e); } if (null != callBackException) { LOG.warn("Fail in write to SLS", callBackException); if (failOnError) { throw callBackException; } callBackException = null; } long end = System.currentTimeMillis(); latencyGauge.report(end - start, 1); outTps.markEvent(); } } SlsOutputFormat(
			String endPoint, String accessKeyId, String accessKey,
			String projectName, String logstoreName,
			SlsRecordResolver<T> serializationSchema);  SlsOutputFormat(
			String endPoint, Configuration properties,
			String projectName, String logstoreName,
			SlsRecordResolver<T> serializationSchema); SlsOutputFormat setFailOnError(boolean failOnError); SlsOutputFormat setTimeZone(String timeZone); SlsOutputFormat setFlushInterval(int flushInterval); @Override void open(int taskNumber, int numTasks); @Override void writeRecord(T row); @Override void sync(); @Override void close(); void setMaxRetryTime(int maxRetryTime); @Override void configure(Configuration config); @Override long getRetryTimeout(); @Override String toString(); }
SlsOutputFormat extends RichOutputFormat<T> implements Syncable,
		HasRetryTimeout { @Override public void writeRecord(T row) throws IOException { if (null != row) { long start = System.currentTimeMillis(); List<LogItem> tmpLogGroup = new ArrayList<>(1); tmpLogGroup.add(serializationSchema.getLogItem(row)); String partitionKey = serializationSchema.getPartitionKey(row); String topic = serializationSchema.getTopic(row); String source = serializationSchema.getSource(row); try { ListenableFuture<Result> future = logProducerProvider.getClient().send( this.projectName, this.logstore, topic, source, partitionKey, tmpLogGroup); Futures.addCallback(future, sendFutureCallback, executor); numSent.incrementAndGet(); } catch (InterruptedException | ProducerException e) { callBackException = new RuntimeException(e); } if (null != callBackException) { LOG.warn("Fail in write to SLS", callBackException); if (failOnError) { throw callBackException; } callBackException = null; } long end = System.currentTimeMillis(); latencyGauge.report(end - start, 1); outTps.markEvent(); } } SlsOutputFormat(
			String endPoint, String accessKeyId, String accessKey,
			String projectName, String logstoreName,
			SlsRecordResolver<T> serializationSchema);  SlsOutputFormat(
			String endPoint, Configuration properties,
			String projectName, String logstoreName,
			SlsRecordResolver<T> serializationSchema); SlsOutputFormat setFailOnError(boolean failOnError); SlsOutputFormat setTimeZone(String timeZone); SlsOutputFormat setFlushInterval(int flushInterval); @Override void open(int taskNumber, int numTasks); @Override void writeRecord(T row); @Override void sync(); @Override void close(); void setMaxRetryTime(int maxRetryTime); @Override void configure(Configuration config); @Override long getRetryTimeout(); @Override String toString();  }