@Test public void testStartConsumingFromNewTopic() { String yummyTopic = "YummyPizza"; String saltyTopic = "SaltyPizza"; createTopic(_zkUtils, yummyTopic); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); Properties override = new Properties(); override.put(KafkaBasedConnectorConfig.DOMAIN_KAFKA_CONSUMER + "." + ConsumerConfig.METADATA_MAX_AGE_CONFIG, "100"); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("MirrorMakerConnector", getDefaultConfig(Optional.of(override)), "testCluster"); connector.start(null); connector.onAssignmentChange(Collections.singletonList(task)); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 1, _kafkaCluster); if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 1, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("did not transfer the msgs within timeout. transferred " + datastreamProducer.getEvents().size()); } DatastreamProducerRecord record = datastreamProducer.getEvents().get(0); Assert.assertEquals(record.getDestination().get(), yummyTopic, "Expected record from YummyPizza topic but got record from " + record.getDestination().get()); createTopic(_zkUtils, saltyTopic); KafkaMirrorMakerConnectorTestUtils.produceEvents(saltyTopic, 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()); } record = datastreamProducer.getEvents().get(1); Assert.assertEquals(record.getDestination().get(), saltyTopic, "Expected record from SaltyPizza topic but got record from " + record.getDestination().get()); connector.stop(); }
@Override public void stop() { super.stop(); _shutdown = true; _partitionDiscoveryThreadMap.values().forEach(PartitionDiscoveryThread::shutdown); }
KafkaMirrorMakerConnector extends AbstractKafkaConnector { @Override public void stop() { super.stop(); _shutdown = true; _partitionDiscoveryThreadMap.values().forEach(PartitionDiscoveryThread::shutdown); } }
KafkaMirrorMakerConnector extends AbstractKafkaConnector { @Override public void stop() { super.stop(); _shutdown = true; _partitionDiscoveryThreadMap.values().forEach(PartitionDiscoveryThread::shutdown); } KafkaMirrorMakerConnector(String connectorName, Properties config, String clusterName); }
KafkaMirrorMakerConnector extends AbstractKafkaConnector { @Override public void stop() { super.stop(); _shutdown = true; _partitionDiscoveryThreadMap.values().forEach(PartitionDiscoveryThread::shutdown); } KafkaMirrorMakerConnector(String connectorName, Properties config, String clusterName); @Override void initializeDatastream(Datastream stream, List<Datastream> allDatastreams); @Override String getDestinationName(Datastream stream); @Override List<BrooklinMetricInfo> getMetricInfos(); @Override void postDatastreamInitialize(Datastream datastream, List<Datastream> allDatastreams); @Override void onPartitionChange(java.util.function.Consumer<DatastreamGroup> callback); @Override boolean isPartitionManagementSupported(); @Override void stop(); @Override Map<String, Optional<DatastreamGroupPartitionsMetadata>> getDatastreamPartitions(); @Override void handleDatastream(List<DatastreamGroup> datastreamGroups); }
KafkaMirrorMakerConnector extends AbstractKafkaConnector { @Override public void stop() { super.stop(); _shutdown = true; _partitionDiscoveryThreadMap.values().forEach(PartitionDiscoveryThread::shutdown); } KafkaMirrorMakerConnector(String connectorName, Properties config, String clusterName); @Override void initializeDatastream(Datastream stream, List<Datastream> allDatastreams); @Override String getDestinationName(Datastream stream); @Override List<BrooklinMetricInfo> getMetricInfos(); @Override void postDatastreamInitialize(Datastream datastream, List<Datastream> allDatastreams); @Override void onPartitionChange(java.util.function.Consumer<DatastreamGroup> callback); @Override boolean isPartitionManagementSupported(); @Override void stop(); @Override Map<String, Optional<DatastreamGroupPartitionsMetadata>> getDatastreamPartitions(); @Override void handleDatastream(List<DatastreamGroup> datastreamGroups);  }