@Test public void testAutoOffsetResetConfigOverride() throws Exception { String yummyTopic = "YummyPizza"; createTopic(_zkUtils, yummyTopic); Datastream datastreamLatest = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastreamLatest.getMetadata().put(KafkaDatastreamMetadataConstants.CONSUMER_OFFSET_RESET_STRATEGY, AbstractKafkaBasedConnectorTask.CONSUMER_AUTO_OFFSET_RESET_CONFIG_LATEST); datastreamLatest.getMetadata().put(DatastreamMetadataConstants.GROUP_ID, "groupIdForDatastreamLatest"); Datastream datastreamEarliest = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastreamEarliest.getMetadata().put(KafkaDatastreamMetadataConstants.CONSUMER_OFFSET_RESET_STRATEGY, AbstractKafkaBasedConnectorTask.CONSUMER_AUTO_OFFSET_RESET_CONFIG_EARLIEST); datastreamEarliest.getMetadata().put(DatastreamMetadataConstants.GROUP_ID, "groupIdForDatastreamEarliest"); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 5, _kafkaCluster); DatastreamTaskImpl taskLatest = new DatastreamTaskImpl(Collections.singletonList(datastreamLatest)); MockDatastreamEventProducer datastreamProducerLatest = new MockDatastreamEventProducer(); taskLatest.setEventProducer(datastreamProducerLatest); DatastreamTaskImpl taskEarliest = new DatastreamTaskImpl(Collections.singletonList(datastreamEarliest)); MockDatastreamEventProducer datastreamProducerEarliest = new MockDatastreamEventProducer(); taskEarliest.setEventProducer(datastreamProducerEarliest); KafkaMirrorMakerConnectorTask connectorTaskLatest = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(taskLatest); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTaskLatest); KafkaMirrorMakerConnectorTask connectorTaskEarliest = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(taskEarliest); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTaskEarliest); Assert.assertTrue(PollUtils.poll( () -> datastreamProducerEarliest.getEvents().size() == 5 && datastreamProducerLatest.getEvents().size() == 0, POLL_PERIOD_MS, POLL_TIMEOUT_MS), "Datastream configured with offset reset latest should have gotten " + "0 events, and datastream configured with earliest should have gotten 5 events"); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 5, _kafkaCluster); Assert.assertTrue(PollUtils.poll( () -> datastreamProducerEarliest.getEvents().size() == 10 && datastreamProducerLatest.getEvents().size() == 5, POLL_PERIOD_MS, POLL_TIMEOUT_MS), "Datastream configured with offset reset latest should have gotten " + "5 events, and datastream configured with earliest should have gotten 10 events"); connectorTaskLatest.stop(); Assert.assertTrue(connectorTaskLatest.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); connectorTaskEarliest.stop(); Assert.assertTrue(connectorTaskEarliest.awaitStop(CONNECTOR_AWAIT_STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS), "did not shut down on time"); }
@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; }