Don't allow regex in the source topic when a specific table is provided as destination

Author: sdomalap

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/mirrormaker/KafkaMirrorMakerConnector.java

@@ -131,12 +131,31 @@ public void initializeDatastream(Datastream stream, List<Datastream> allDatastre
         if (placeholderIndex != -1 && placeholderIndex == destinationConnectionString.lastIndexOf(MM_TOPIC_PLACEHOLDER)) {
           LOG.info("Allowing user managed datastream destination with connector {} for datastream {}", stream.getConnectorName(), stream.getName());
         } else {
-          // TODO: Implement validation on the source to confirm that it is not a regex/wildcard source
+
+          // validation on the source to confirm that it is not a regex/wildcard source
+          // handling for DOT character is enough as Kafka allows only alphanumeric characters, periods (.), underscores (_), and hyphens (-) in its topic names
+          KafkaConnectionString connectionString = KafkaConnectionString.valueOf(stream.getSource().getConnectionString());
+          String topicName = connectionString.getTopicName();
+          int indexOfDOT = topicName.indexOf(".");
+          while (indexOfDOT >= 0) {
+            if (indexOfDOT == 0 || topicName.charAt(indexOfDOT - 1) != '\\') {
+              LOG.error("DOT character is not escaped. Regex in the source is not supported with explicit user managed datastream destination.");
+              throw new DatastreamValidationException(
+                      String.format("Datastream source is invalid for connector %s. Datastream: %s. {}" +
+                                      "Regex in the source is not supported with explicit user managed datastream destination.",
+                              stream.getConnectorName(),
+                              stream
+                      )
+              );
+            }
+            indexOfDOT = topicName.indexOf(".", indexOfDOT + 1);
+          }
+
           LOG.warn("Allowing explicit user managed datastream destination with connector {} for datastream {}", stream.getConnectorName(), stream.getName());
-//          LOG.error("User managed datastream destination format is invalid for connector {} and datastream {}", stream.getConnectorName(), stream.getName());
-//          throw new DatastreamValidationException(
-//                  String.format("Datastream destination format is invalid for connector %s. Datastream: %s", stream.getConnectorName(), stream)
-//          );
+          //LOG.error("User managed datastream destination format is invalid for connector {} and datastream {}", stream.getConnectorName(), stream.getName());
+          //throw new DatastreamValidationException(
+          //        String.format("Datastream destination format is invalid for connector %s. Datastream: %s", stream.getConnectorName(), stream)
+          //);
         }
       } else {
         throw new DatastreamValidationException(