Kafka Connect: Add regex for property file match (#11303)

Author: ryanjclark

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java

@@ -406,6 +406,12 @@ public JsonConverter jsonConverter() {
     return jsonConverter;
   }
 
+  @VisibleForTesting
+  static boolean checkClassName(String className) {
+    return (className.matches(".*\\.ConnectDistributed.*")
+        || className.matches(".*\\.ConnectStandalone.*"));
+  }
+
   /**
    * This method attempts to load the Kafka Connect worker properties, which are not exposed to
    * connectors. It does this by parsing the Java command used to launch the worker, extracting the
@@ -422,9 +428,7 @@ private Map<String, String> loadWorkerProps() {
     String javaCmd = System.getProperty("sun.java.command");
     if (javaCmd != null && !javaCmd.isEmpty()) {
       List<String> args = Splitter.on(' ').splitToList(javaCmd);
-      if (args.size() > 1
-          && (args.get(0).endsWith(".ConnectDistributed")
-              || args.get(0).endsWith(".ConnectStandalone"))) {
+      if (args.size() > 1 && checkClassName(args.get(0))) {
         Properties result = new Properties();
         try (InputStream in = Files.newInputStream(Paths.get(args.get(1)))) {
           result.load(in);

kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/IcebergSinkConfigTest.java

@@ -88,4 +88,26 @@ public void testStringToList() {
 
   @Test
   public void testStringWithParensToList() {}
+
+  @Test
+  public void testCheckClassName() {
+    Boolean result =
+        IcebergSinkConfig.checkClassName("org.apache.kafka.connect.cli.ConnectDistributed");
+    assertThat(result).isTrue();
+
+    result = IcebergSinkConfig.checkClassName("org.apache.kafka.connect.cli.ConnectStandalone");
+    assertThat(result).isTrue();
+
+    result = IcebergSinkConfig.checkClassName("some.other.package.ConnectDistributed");
+    assertThat(result).isTrue();
+
+    result = IcebergSinkConfig.checkClassName("some.other.package.ConnectStandalone");
+    assertThat(result).isTrue();
+
+    result = IcebergSinkConfig.checkClassName("some.package.ConnectDistributedWrapper");
+    assertThat(result).isTrue();
+
+    result = IcebergSinkConfig.checkClassName("org.apache.kafka.clients.producer.KafkaProducer");
+    assertThat(result).isFalse();
+  }
 }