@Test public void testGetStateForMultipleTasks() throws Exception { List<String> topics = Arrays.asList("YummyPizza", "SaltyPizza", "HealthySalad"); topics.forEach(t -> createTopic(_zkUtils, t)); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); List<DatastreamTask> tasks = ImmutableList.of(new DatastreamTaskImpl(Collections.singletonList(datastream)), new DatastreamTaskImpl(Collections.singletonList(datastream)), new DatastreamTaskImpl(Collections.singletonList(datastream))); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); tasks.forEach(t -> ((DatastreamTaskImpl) t).setEventProducer(datastreamProducer)); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("MirrorMakerConnector", getDefaultConfig(Optional.empty()), "testCluster"); connector.start(null); connector.onAssignmentChange(tasks); boolean hasExpectedTopicPartitions = PollUtils.poll(() -> { String jsonStr = connector.process("/datastream_state?datastream=pizzaStream"); KafkaDatastreamStatesResponse response = KafkaDatastreamStatesResponse.fromJson(jsonStr); return response.getAssignedTopicPartitions().equals(ImmutableSet.of(new TopicPartition("SaltyPizza", 0), new TopicPartition("YummyPizza", 0))); }, POLL_PERIOD_MS, POLL_TIMEOUT_MS); Assert.assertTrue(hasExpectedTopicPartitions); 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);  }