@Test public void testRegularCommitWithFlushlessProducer() throws Exception { MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); 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(anyMapOf(TopicPartition.class, OffsetAndMetadata.class), any(Duration.class)); return result; } }; KafkaBasedConnectorConfig connectorConfig = new KafkaBasedConnectorConfigBuilder() .setConsumerFactory(kafkaConsumerFactory) .setCommitIntervalMillis(200) .build(); KafkaMirrorMakerConnectorTask connectorTask = new KafkaMirrorMakerConnectorTask( connectorConfig, task, "", true, new KafkaGroupIdConstructor(false, "test")); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); Assert.assertTrue(remainingCommitSyncCalls.await(10, TimeUnit.SECONDS), "Kafka consumer commitSync was not invoked as often as expected"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 0); connectorTask.stop(); Assert.assertTrue(connectorTask.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); Assert.assertEquals(datastreamProducer.getNumFlushes(), 1); }
@Override public void stop() { super.stop(); _topicManager.stop(); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); @Override void onPartitionsAssigned(Collection<TopicPartition> partitions); @Override void run(); @Override void stop(); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); static List<BrooklinMetricInfo> getMetricInfos(String connectorName); @Override KafkaDatastreamStatesResponse getKafkaDatastreamStatesResponse(); @VisibleForTesting TopicManager getTopicManager(); }
KafkaMirrorMakerConnectorTask extends AbstractKafkaBasedConnectorTask { @Override public void stop() { super.stop(); _topicManager.stop(); } KafkaMirrorMakerConnectorTask(KafkaBasedConnectorConfig config, DatastreamTask task, String connectorName,
      boolean isFlushlessModeEnabled, GroupIdConstructor groupIdConstructor); @Override void onPartitionsAssigned(Collection<TopicPartition> partitions); @Override void run(); @Override void stop(); @Override void onPartitionsRevoked(Collection<TopicPartition> partitions); static List<BrooklinMetricInfo> getMetricInfos(String connectorName); @Override KafkaDatastreamStatesResponse getKafkaDatastreamStatesResponse(); @VisibleForTesting TopicManager getTopicManager(); static final String TOPIC_MANAGER_FACTORY; static final String DEFAULT_TOPIC_MANAGER_FACTORY; static final String DESTINATION_TOPIC_PREFIX; static final String DOMAIN_TOPIC_MANAGER; static final String TOPIC_MANAGER_METRICS_PREFIX; }