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