@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(); }
@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);  }