@Test public void testConsumeFromMultipleTopicsWithDestinationTopicPrefixMetadata() throws Exception { String yummyTopic = "YummyPizza"; String saltyTopic = "SaltyPizza"; String saladTopic = "HealthySalad"; String destinationTopicPrefixOverride = "newPrefix"; createTopic(_zkUtils, saladTopic); createTopic(_zkUtils, yummyTopic); createTopic(_zkUtils, saltyTopic); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); datastream.getMetadata().put(DatastreamMetadataConstants.DESTINATION_TOPIC_PREFIX, destinationTopicPrefixOverride); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); KafkaMirrorMakerConnectorTask connectorTask = KafkaMirrorMakerConnectorTestUtils.createKafkaMirrorMakerConnectorTask(task); KafkaMirrorMakerConnectorTestUtils.runKafkaMirrorMakerConnectorTask(connectorTask); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 1, _kafkaCluster); KafkaMirrorMakerConnectorTestUtils.produceEvents(saltyTopic, 1, _kafkaCluster); KafkaMirrorMakerConnectorTestUtils.produceEvents(saladTopic, 1, _kafkaCluster); if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 2, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("did not transfer the msgs within timeout. transferred " + datastreamProducer.getEvents().size()); } List<DatastreamProducerRecord> records = datastreamProducer.getEvents(); for (DatastreamProducerRecord record : records) { String destinationTopic = record.getDestination().get(); Assert.assertTrue(destinationTopic.endsWith("Pizza"), "Unexpected event consumed from Datastream and sent to topic: " + destinationTopic); Assert.assertTrue(destinationTopic.startsWith(destinationTopicPrefixOverride), "Destination topic prefix enabled, topic should start with prefix. Topic: " + destinationTopic); } validateTaskConsumerAssignment(connectorTask, Sets.newHashSet(new TopicPartition(yummyTopic, 0), new TopicPartition(saltyTopic, 0))); connectorTask.stop(); Assert.assertTrue(connectorTask.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; }