KafkaMirrorMakerConnector extends AbstractKafkaConnector { @Override public void initializeDatastream(Datastream stream, List<Datastream> allDatastreams) throws DatastreamValidationException { if (DatastreamUtils.isReuseAllowed(stream)) { throw new DatastreamValidationException( String.format("Destination reuse is not allowed for connector %s. Datastream: %s", stream.getConnectorName(), stream)); } if (DatastreamUtils.isUserManagedDestination(stream)) { throw new DatastreamValidationException( String.format("BYOT is not allowed for connector %s. Datastream: %s", stream.getConnectorName(), stream)); } if (!DatastreamUtils.isConnectorManagedDestination(stream)) { stream.getMetadata() .put(DatastreamMetadataConstants.IS_CONNECTOR_MANAGED_DESTINATION_KEY, Boolean.TRUE.toString()); } KafkaConnectionString connectionString = KafkaConnectionString.valueOf(stream.getSource().getConnectionString()); try { Pattern pattern = Pattern.compile(connectionString.getTopicName()); LOG.info("Successfully compiled topic name pattern {}", pattern); } catch (PatternSyntaxException e) { throw new DatastreamValidationException( String.format("Regular expression in Datastream source connection string (%s) is ill-formatted.", stream.getSource().getConnectionString()), e); } } 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 testInitializeDatastream() throws Exception { String sourceRegex = "\\w+Event"; StringMap metadata = new StringMap(); metadata.put(DatastreamMetadataConstants.REUSE_EXISTING_DESTINATION_KEY, Boolean.FALSE.toString()); Datastream ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastream", _broker, sourceRegex, metadata); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("testInitializeDatastream", getDefaultConfig(Optional.empty()), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); sourceRegex = "SpecificTopic"; ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastream2", _broker, sourceRegex, metadata); connector.initializeDatastream(ds, Collections.emptyList()); sourceRegex = "(\\w+Event)|^(Topic)"; ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastream3", _broker, sourceRegex, metadata); connector.initializeDatastream(ds, Collections.emptyList()); sourceRegex = "^(?!__)\\w+"; ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastream4", _broker, sourceRegex, metadata); connector.initializeDatastream(ds, Collections.emptyList()); Assert.assertTrue(DatastreamUtils.isConnectorManagedDestination(ds)); }
@Test(expectedExceptions = DatastreamValidationException.class) public void testInitializeDatastreamWithDestinationReuse() throws DatastreamValidationException { String sourceRegex = "\\w+Event"; StringMap metadata = new StringMap(); metadata.put(DatastreamMetadataConstants.REUSE_EXISTING_DESTINATION_KEY, Boolean.TRUE.toString()); Datastream ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastreamWithDestinationReuse", _broker, sourceRegex, metadata); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("testInitializeDatastreamWithDestinationReuse", getDefaultConfig(Optional.empty()), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); }
@Test(expectedExceptions = DatastreamValidationException.class) public void testInitializeDatastreamWithBYOT() throws DatastreamValidationException { String sourceRegex = "\\w+Event"; StringMap metadata = new StringMap(); metadata.put(DatastreamMetadataConstants.IS_USER_MANAGED_DESTINATION_KEY, Boolean.TRUE.toString()); Datastream ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastreamWithBYOT", _broker, sourceRegex, metadata); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("testInitializeDatastreamWithBYOT", getDefaultConfig(Optional.empty()), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); }
@Test(expectedExceptions = DatastreamValidationException.class) public void testInitializeDatastreamWithBadSource() throws DatastreamValidationException { String sourceRegex = "*Event*"; StringMap metadata = new StringMap(); metadata.put(DatastreamMetadataConstants.REUSE_EXISTING_DESTINATION_KEY, Boolean.FALSE.toString()); Datastream ds = KafkaMirrorMakerConnectorTestUtils.createDatastream("testInitializeDatastreamWithBadSource", _broker, sourceRegex, metadata); KafkaMirrorMakerConnector connector = new KafkaMirrorMakerConnector("testInitializeDatastreamWithBadSource", getDefaultConfig(Optional.empty()), "testCluster"); connector.initializeDatastream(ds, Collections.emptyList()); }