@Override
public void start() {
    super.start();

    ProjectConfig projectConfig = buildProjectConfig();

    producerConfig.getProjectConfigs().put(projectConfig);
    producerConfig.setBatchCountThreshold(batchCountThreshold);
    producerConfig.setBatchSizeThresholdInBytes(batchSizeThresholdInBytes);
    producerConfig.setIoThreadCount(ioThreadCount);
    producerConfig.setRetries(retries);
    producerConfig.setBaseRetryBackoffMs(baseRetryBackoffMs);
    producerConfig.setLingerMs(lingerMs);
    producerConfig.setMaxBlockMs(maxBlockMs);
    producerConfig.setMaxRetryBackoffMs(maxRetryBackoffMs);

    producer = new LogProducer(producerConfig);
}
 

private Producer createProducer(ConfigWrapper configWrapper) {
    ProducerConfig producerConfig = new ProducerConfig();
    producerConfig.setLingerMs(configWrapper.getInt(FLUSH_INTERVAL_MS,
            ProducerConfig.DEFAULT_LINGER_MS));
    producerConfig.setRetries(configWrapper.getInt(MAX_RETRIES,
            ProducerConfig.DEFAULT_RETRIES));
    producerConfig.setBaseRetryBackoffMs(
            configWrapper.getLong(BASE_RETRY_BACK_OFF_TIME_MS,
                    ProducerConfig.DEFAULT_BASE_RETRY_BACKOFF_MS));
    producerConfig.setMaxRetryBackoffMs(
            configWrapper.getLong(MAX_RETRY_BACK_OFF_TIME_MS, ProducerConfig.DEFAULT_MAX_RETRY_BACKOFF_MS));
    producerConfig.setMaxBlockMs(
            configWrapper.getLong(MAX_BLOCK_TIME_MS, ProducerConfig.DEFAULT_MAX_BLOCK_MS));
    producerConfig.setIoThreadCount(
            configWrapper.getInt(IO_THREAD_NUM, ProducerConfig.DEFAULT_IO_THREAD_COUNT));
    Producer producer = new LogProducer(producerConfig);
    ProjectConfig config = new ProjectConfig(project,
            configWrapper.getString(ConfigConstants.LOG_ENDPOINT),
            configWrapper.getString(ConfigConstants.LOG_ACCESSSKEYID),
            configWrapper.getString(ConfigConstants.LOG_ACCESSKEY));
    producer.putProjectConfig(config);
    return producer;
}
 

@Override
protected LogProducer produceNormalClient(String accessId, String accessKey) {
	ProjectConfigs projectConfigs = new ProjectConfigs();
	ProjectConfig projectConfig = new ProjectConfig(this.projectName, this.endPoint, accessId, accessKey);
	projectConfigs.put(projectConfig);
	producerConfig = new ProducerConfig(projectConfigs);
	producerConfig.setLingerMs(flushInterval);
	producerConfig.setRetries(maxRetryTimes);
	LogProducer producer = new LogProducer(producerConfig);
	return producer;
}
 

@Override
protected LogProducer produceStsClient(String accessId, String accessKey, String securityToken) {
	ProjectConfigs projectConfigs = new ProjectConfigs();
	ProjectConfig projectConfig = new ProjectConfig(this.projectName, this.endPoint, accessId, accessKey, securityToken);
	projectConfigs.put(projectConfig);
	producerConfig = new ProducerConfig(projectConfigs);
	producerConfig.setLingerMs(flushInterval);
	producerConfig.setRetries(maxRetryTimes);
	LogProducer producer = new LogProducer(producerConfig);
	return producer;
}
 

@Test
public void testClose() throws ProducerException, InterruptedException {
	LogProducerProvider producerProvider = new LogProducerProvider(
			"test-project", "test-endpoint", "test-ak", "test-secret", 1, 100);

	LogProducer producer = Mockito.mock(LogProducer.class);
	Whitebox.setInternalState(producerProvider, "client", producer);

	Mockito.doThrow(new InterruptedException()).doNothing().when(producer).close();
	producerProvider.closeClient();

	Mockito.verify(producer, Mockito.times(2)).close();
}
 

@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();
	// Use any() instead of anyString() because in Mockito 2.x, anyString() does not match null any more,
	// which may cause the test to fail.
	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());

	// trigger call back.
	future.set(new Result(true, null, 0));
	// wait call back finished.
	executor.awaitTermination(1, TimeUnit.SECONDS);
	assertEquals(1, numSent.get());
	assertEquals(1, numCommitted.get());
}
 

public Producer createProducer() {
    projectConfig = buildProjectConfig();
    Producer producer = new LogProducer(producerConfig);
    producer.putProjectConfig(projectConfig);
    return producer;
}
 

public Producer createProducer() {
  projectConfig = buildProjectConfig();
  Producer producer = new LogProducer(producerConfig);
  producer.putProjectConfig(projectConfig);
  return producer;
}