Add kafka source connection string validation in datastream update pa…

…th (#825) Add validation in datastream update path to reject request for incorrect source regex connection string.
linkedin · May 13, 2021 · f1111e5 · f1111e5
1 parent d411dbc
commit f1111e5
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 10 deletions.
diff --git a/.../java/com/linkedin/datastream/connectors/kafka/mirrormaker/KafkaMirrorMakerConnector.java b/.../java/com/linkedin/datastream/connectors/kafka/mirrormaker/KafkaMirrorMakerConnector.java
@@ -135,16 +135,7 @@ public void initializeDatastream(Datastream stream, List<Datastream> allDatastre
           .put(DatastreamMetadataConstants.IS_CONNECTOR_MANAGED_DESTINATION_KEY, Boolean.TRUE.toString());
     }
 
-    // verify that the source regular expression can be compiled
-    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);
-    }
+    validateSourceConnectionString(stream);
   }
 
   @Override
@@ -250,6 +241,29 @@ public void handleDatastream(List<DatastreamGroup> datastreamGroups) {
     LOG.info("handleDatastream: new datastream groups: {}", _partitionDiscoveryThreadMap.keySet());
   }
 
+  @Override
+  public void validateUpdateDatastreams(List<Datastream> datastreams, List<Datastream> allDatastreams)
+      throws DatastreamValidationException {
+    super.validateUpdateDatastreams(datastreams, allDatastreams);
+    //validate connection string
+    for (Datastream datastream : datastreams) {
+      validateSourceConnectionString(datastream);
+    }
+  }
+
+  private void validateSourceConnectionString(Datastream stream) throws DatastreamValidationException {
+    // verify that the source regular expression can be compiled
+    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);
+    }
+  }
+
   /**
    *  PartitionDiscoveryThread listens to Kafka partitions periodically using the _consumer.listTopic()
    *  to fetch the latest subscribed partitions for a given datastreamGroup

diff --git a/...a/com/linkedin/datastream/connectors/kafka/mirrormaker/TestKafkaMirrorMakerConnector.java b/...a/com/linkedin/datastream/connectors/kafka/mirrormaker/TestKafkaMirrorMakerConnector.java
@@ -254,6 +254,20 @@ public void testValidateDatastreamUpdatePausedPartitions() throws Exception {
     coordinator.stop();
   }
 
+  @Test
+  public void testValidateDatastreamUpdateWithBadSource() 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");
+    Assert.assertThrows(DatastreamValidationException.class, () -> connector.validateUpdateDatastreams(Collections.singletonList(ds), Collections.emptyList()));
+  }
+
   @Test
   public void testMirrorMakerConnectorBasics() {
     String yummyTopic = "YummyPizza";