@Test public void testReduceDatastreamStates() { KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("MirrorMakerConnector", getDefaultConfig(Optional.empty()), "testCluster"); String datastreamName = "testProcessDatastreamStates"; String yummyTopic = "YummyPizza"; String saltyTopic = "SaltyPizza"; String instance1 = "instance1"; String instance2 = "instance2"; Set<TopicPartition> assignedPartitions1 = new HashSet<>(); assignedPartitions1.add(new TopicPartition(yummyTopic, 0)); assignedPartitions1.add(new TopicPartition(yummyTopic, 4)); assignedPartitions1.add(new TopicPartition(yummyTopic, 10)); assignedPartitions1.add(new TopicPartition(yummyTopic, 11)); assignedPartitions1.add(new TopicPartition(yummyTopic, 23)); assignedPartitions1.add(new TopicPartition(saltyTopic, 2)); assignedPartitions1.add(new TopicPartition(saltyTopic, 5)); assignedPartitions1.add(new TopicPartition(saltyTopic, 77)); Map<TopicPartition, PausedSourcePartitionMetadata> autoPausedPartitions1 = new HashMap<>(); Map<String, Set<String>> manualPausedPartitions1 = new HashMap<>(); autoPausedPartitions1.put(new TopicPartition(yummyTopic, 0), new PausedSourcePartitionMetadata(() -> false, PausedSourcePartitionMetadata.Reason.SEND_ERROR)); autoPausedPartitions1.put(new TopicPartition(yummyTopic, 10), new PausedSourcePartitionMetadata(() -> false, PausedSourcePartitionMetadata.Reason.SEND_ERROR)); manualPausedPartitions1.put(saltyTopic, Sets.newHashSet("2", "5", "77")); manualPausedPartitions1.put(yummyTopic, Sets.newHashSet("4", "11", "23")); Map<FlushlessEventProducerHandler.SourcePartition, Long> inflightMsgCounts1 = new HashMap<>(); inflightMsgCounts1.put(new FlushlessEventProducerHandler.SourcePartition(yummyTopic, 4), 6L); KafkaDatastreamStatesResponse process1Response = new KafkaDatastreamStatesResponse(datastreamName, autoPausedPartitions1, manualPausedPartitions1, assignedPartitions1, inflightMsgCounts1); Set<TopicPartition> assignedPartitions2 = new HashSet<>(); assignedPartitions2.add(new TopicPartition(yummyTopic, 19)); assignedPartitions2.add(new TopicPartition(saltyTopic, 1)); assignedPartitions2.add(new TopicPartition(saltyTopic, 6)); assignedPartitions2.add(new TopicPartition(saltyTopic, 9)); assignedPartitions2.add(new TopicPartition(saltyTopic, 17)); assignedPartitions2.add(new TopicPartition(saltyTopic, 19)); assignedPartitions2.add(new TopicPartition(saltyTopic, 25)); Map<TopicPartition, PausedSourcePartitionMetadata> autoPausedPartitions2 = new HashMap<>(); Map<String, Set<String>> manualPausedPartitions2 = new HashMap<>(); autoPausedPartitions2.put(new TopicPartition(saltyTopic, 6), new PausedSourcePartitionMetadata(() -> false, PausedSourcePartitionMetadata.Reason.SEND_ERROR)); autoPausedPartitions2.put(new TopicPartition(saltyTopic, 17), new PausedSourcePartitionMetadata(() -> false, PausedSourcePartitionMetadata.Reason.SEND_ERROR)); manualPausedPartitions2.put(saltyTopic, Sets.newHashSet("1", "9", "25")); manualPausedPartitions2.put(yummyTopic, Sets.newHashSet("19")); Map<FlushlessEventProducerHandler.SourcePartition, Long> inflightMsgCounts2 = new HashMap<>(); inflightMsgCounts2.put(new FlushlessEventProducerHandler.SourcePartition(saltyTopic, 9), 20L); KafkaDatastreamStatesResponse process2Response = new KafkaDatastreamStatesResponse(datastreamName, autoPausedPartitions2, manualPausedPartitions2, assignedPartitions2, inflightMsgCounts2); Map<String, String> responseMap = new HashMap<>(); responseMap.put(instance1, KafkaDatastreamStatesResponse.toJson(process1Response)); responseMap.put(instance2, KafkaDatastreamStatesResponse.toJson(process2Response)); String resultJson = connector.reduce(DATASTREAM_STATE_QUERY, responseMap); Map<String, String> reducedResults = JsonUtils.fromJson(resultJson, new TypeReference<HashMap<String, String>>() { }); Assert.assertEquals(reducedResults.keySet().size(), 2, "Should have received states from 2 instances"); KafkaDatastreamStatesResponse response1 = KafkaDatastreamStatesResponse.fromJson(responseMap.get(instance1)); Assert.assertEquals(response1.getAssignedTopicPartitions(), assignedPartitions1, "Assigned partitions mismatch for instance1"); Assert.assertEquals(response1.getInFlightMessageCounts(), inflightMsgCounts1, "In-flight message counts mismatch for instance1"); Assert.assertEquals(response1.getManualPausedPartitions(), manualPausedPartitions1, "Manually paused partitions mismatch for instance1"); Assert.assertEquals(response1.getAutoPausedPartitions(), autoPausedPartitions1, "Auto-paused partitions mismatch for instance1"); Assert.assertEquals(response1.getDatastream(), datastreamName, "Datastream name mismatch for instance1"); KafkaDatastreamStatesResponse response2 = KafkaDatastreamStatesResponse.fromJson(responseMap.get(instance2)); Assert.assertEquals(response2.getAssignedTopicPartitions(), assignedPartitions2, "Assigned partitions mismatch for instance2"); Assert.assertEquals(response2.getInFlightMessageCounts(), inflightMsgCounts2, "In-flight message counts mismatch for instance2"); Assert.assertEquals(response2.getManualPausedPartitions(), manualPausedPartitions2, "Manually paused partitions mismatch for instance2"); Assert.assertEquals(response2.getAutoPausedPartitions(), autoPausedPartitions2, "Auto-paused partitions mismatch for instance2"); Assert.assertEquals(response2.getDatastream(), datastreamName, "Datastream name mismatch for instance"); 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);  }