KafkaConnectorTask extends AbstractKafkaBasedConnectorTask { public static Consumer<?, ?> createConsumer(KafkaConsumerFactory<?, ?> consumerFactory, Properties consumerProps, String groupId, KafkaConnectionString connectionString) { Properties props = getKafkaConsumerProperties(consumerProps, groupId, connectionString); return consumerFactory.createConsumer(props); } KafkaConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      GroupIdConstructor groupIdConstructor); static Consumer<?, ?> createConsumer(KafkaConsumerFactory<?, ?> consumerFactory, Properties consumerProps,
      String groupId, KafkaConnectionString connectionString); static List<BrooklinMetricInfo> getMetricInfos(String connectorName);  }
@Test public void testCommittingOffsetRegularly() throws Exception { String topic = "pizza1"; createTopic(_zkUtils, topic); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); Datastream datastream = getDatastream(_broker, topic); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); task.setEventProducer(datastreamProducer); CountDownLatch remainingCommitSyncCalls = new CountDownLatch(3); KafkaConsumerFactory<byte[], byte[]> kafkaConsumerFactory = new KafkaConsumerFactoryImpl() { @Override public Consumer<byte[], byte[]> createConsumer(Properties properties) { Consumer<byte[], byte[]> result = spy(super.createConsumer(properties)); doAnswer(invocation -> { remainingCommitSyncCalls.countDown(); return null; }) .when(result).commitSync(any(Duration.class)); return result; } }; KafkaConnectorTask connectorTask = createKafkaConnectorTask(task, new KafkaBasedConnectorConfigBuilder() .setConsumerFactory(kafkaConsumerFactory).build()); Assert.assertTrue(remainingCommitSyncCalls.await(10, TimeUnit.SECONDS), "Kafka consumer commitSync was not invoked as often as expected"); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }