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);  }
@Test public void testMirrorMakerConnectorBasics() { String yummyTopic = "YummyPizza"; String saltyTopic = "SaltyPizza"; String saladTopic = "HealthySalad"; createTopic(_zkUtils, saladTopic); createTopic(_zkUtils, yummyTopic); createTopic(_zkUtils, saltyTopic); Datastream datastream = KafkaMirrorMakerConnectorTestUtils.createDatastream("pizzaStream", _broker, "\\w+Pizza"); DatastreamTaskImpl task = new DatastreamTaskImpl(Collections.singletonList(datastream)); MockDatastreamEventProducer datastreamProducer = new MockDatastreamEventProducer(); task.setEventProducer(datastreamProducer); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("MirrorMakerConnector", getDefaultConfig(Optional.empty()), "testCluster"); connector.start(null); connector.onAssignmentChange(Collections.singletonList(task)); 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); } Map<String, Set<String>> pausedPartitions = new HashMap<>(); pausedPartitions.put(saltyTopic, Sets.newHashSet("0")); datastream.getMetadata() .put(DatastreamMetadataConstants.PAUSED_SOURCE_PARTITIONS_KEY, JsonUtils.toJson(pausedPartitions)); connector.onAssignmentChange(Collections.singletonList(task)); if (!PollUtils.poll(() -> { Map<String, Set<String>> manuallyPaused = KafkaDatastreamStatesResponse.fromJson(connector.process(DATASTREAM_STATE_QUERY + datastream.getName())) .getManualPausedPartitions(); return manuallyPaused.containsKey(saltyTopic) && manuallyPaused.get(saltyTopic).contains("0"); }, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("manualPausedPartitions was not properly retrieved from process() after pause"); } pausedPartitions.remove(saltyTopic); datastream.getMetadata() .put(DatastreamMetadataConstants.PAUSED_SOURCE_PARTITIONS_KEY, JsonUtils.toJson(pausedPartitions)); connector.onAssignmentChange(Collections.singletonList(task)); if (!PollUtils.poll(() -> { Map<String, Set<String>> manuallyPaused = KafkaDatastreamStatesResponse.fromJson(connector.process(DATASTREAM_STATE_QUERY + datastream.getName())) .getManualPausedPartitions(); return manuallyPaused.isEmpty(); }, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("manualPausedPartitions was not properly retrieved from process() after resume"); } datastreamProducer.setSendFailCondition((record) -> true); KafkaMirrorMakerConnectorTestUtils.produceEvents(yummyTopic, 1, _kafkaCluster); if (!PollUtils.poll(() -> { Map<TopicPartition, PausedSourcePartitionMetadata> autoPausedPartitions = KafkaDatastreamStatesResponse.fromJson(connector.process(DATASTREAM_STATE_QUERY + datastream.getName())) .getAutoPausedPartitions(); TopicPartition expected = new TopicPartition(yummyTopic, 0); return autoPausedPartitions.containsKey(expected) && autoPausedPartitions.get(expected).getReason() == PausedSourcePartitionMetadata.Reason.SEND_ERROR; }, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("autoPausedPartitions was not properly retrieved from process() after auto-pause"); } datastreamProducer.setSendFailCondition((record) -> false); if (!PollUtils.poll(() -> { Map<TopicPartition, PausedSourcePartitionMetadata> autoPausedPartitions = KafkaDatastreamStatesResponse.fromJson(connector.process(DATASTREAM_STATE_QUERY + datastream.getName())) .getAutoPausedPartitions(); return autoPausedPartitions.isEmpty(); }, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("autoPausedPartitions was not properly retrieved from process() after auto-resume"); } connector.stop(); }
@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(); }
@Test public void testConsumptionUnaffectedByDeleteTopic() { String yummyTopic = "YummyPizza"; String saltyTopic = "SaltyPizza"; createTopic(_zkUtils, yummyTopic); createTopic(_zkUtils, saltyTopic); 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); 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()); } datastreamProducer.getEvents().clear(); deleteTopic(_zkUtils, yummyTopic); KafkaMirrorMakerConnectorTestUtils.produceEvents(saltyTopic, 1, _kafkaCluster); if (!PollUtils.poll(() -> datastreamProducer.getEvents().size() == 1, POLL_PERIOD_MS, POLL_TIMEOUT_MS)) { Assert.fail("did not get the event from SaltyPizza topic"); } DatastreamProducerRecord record = datastreamProducer.getEvents().get(0); Assert.assertEquals(record.getDestination().get(), saltyTopic, "Expected record from SaltyPizza topic but got record from " + record.getDestination().get()); connector.stop(); }
@Test public void testProcessDatastreamStatesRequestInvalid() { KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("MirrorMakerConnector", getDefaultConfig(Optional.empty()), "testCluster"); connector.start(null); Assert.assertNull(connector.process("")); Assert.assertNull(connector.process("/not_datastream_state?datastream=name")); Assert.assertNull(connector.process("/datastream_state?notdatastream=name")); connector.stop(); }
@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(); }
@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(); }