Merge branch '10.2.x' into master

Author: semaphore-agent-production[bot]

src/main/java/io/confluent/connect/hdfs/FileUtils.java

@@ -15,6 +15,8 @@
 
 package io.confluent.connect.hdfs;
 
+import com.google.re2j.Matcher;
+
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -27,7 +29,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
-import java.util.regex.Matcher;
 
 import io.confluent.connect.hdfs.filter.CommittedFileFilter;
 import io.confluent.connect.hdfs.storage.Storage;

src/main/java/io/confluent/connect/hdfs/HdfsSinkConnectorConfig.java

@@ -15,12 +15,12 @@
 
 package io.confluent.connect.hdfs;
 
+import com.google.re2j.Matcher;
+import com.google.re2j.Pattern;
+import com.google.re2j.PatternSyntaxException;
 import io.confluent.connect.hdfs.orc.OrcFormat;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.regex.PatternSyntaxException;
 import io.confluent.connect.hdfs.parquet.ParquetFormat;
 import io.confluent.connect.hdfs.string.StringFormat;
 import org.apache.commons.lang.StringUtils;
@@ -402,10 +402,14 @@ protected HdfsSinkConnectorConfig(ConfigDef configDef, Map<String, String> props
     this.url = extractUrl();
     try {
       String topicRegex = getString(TOPIC_CAPTURE_GROUPS_REGEX_CONFIG);
-      this.topicRegexCaptureGroup = topicRegex != null ? Pattern.compile(topicRegex) : null;
+      if (topicRegex != null) {
+        this.topicRegexCaptureGroup = Pattern.compile(topicRegex);
+      } else {
+        this.topicRegexCaptureGroup = null;
+      }
     } catch (PatternSyntaxException e) {
       throw new ConfigException(
-          TOPIC_CAPTURE_GROUPS_REGEX_CONFIG + " is an invalid regex pattern: ",
+          TOPIC_CAPTURE_GROUPS_REGEX_CONFIG + " is an invalid regex pattern: " + e.getMessage(),
           e
       );
     }

src/main/java/io/confluent/connect/hdfs/HdfsSinkConnectorConstants.java

@@ -15,7 +15,7 @@
 
 package io.confluent.connect.hdfs;
 
-import java.util.regex.Pattern;
+import com.google.re2j.Pattern;
 
 public class HdfsSinkConnectorConstants {
 

src/main/java/io/confluent/connect/hdfs/filter/CommittedFileFilter.java

@@ -17,12 +17,10 @@
 
 import io.confluent.connect.hdfs.HdfsSinkConnectorConstants;
 
+import com.google.re2j.Matcher;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 
-import java.util.regex.Matcher;
-
-
 public class CommittedFileFilter implements PathFilter {
   @Override
   public boolean accept(Path path) {

src/main/java/io/confluent/connect/hdfs/filter/TopicCommittedFileFilter.java

@@ -15,12 +15,11 @@
 
 package io.confluent.connect.hdfs.filter;
 
+import com.google.re2j.Matcher;
 import org.apache.hadoop.fs.Path;
 
 import io.confluent.connect.hdfs.HdfsSinkConnectorConstants;
 
-import java.util.regex.Matcher;
-
 public class TopicCommittedFileFilter extends CommittedFileFilter {
   private String topic;
 

src/main/java/io/confluent/connect/hdfs/filter/TopicPartitionCommittedFileFilter.java

@@ -15,11 +15,10 @@
 
 package io.confluent.connect.hdfs.filter;
 
+import com.google.re2j.Matcher;
 import org.apache.hadoop.fs.Path;
 import org.apache.kafka.common.TopicPartition;
 
-import java.util.regex.Matcher;
-
 import io.confluent.connect.hdfs.HdfsSinkConnectorConstants;
 
 public class TopicPartitionCommittedFileFilter extends CommittedFileFilter {

src/test/java/io/confluent/connect/hdfs/HdfsSinkConnectorConfigTest.java

@@ -82,7 +82,6 @@ public void testHiveTableNameValidation() {
             configException.getMessage());
   }
 
-  @Test
   public void testValidRegexCaptureGroup() {
     String topic = "topica";
     String topicDir = "topic.another.${topic}.again";
@@ -172,15 +171,52 @@ public void testInvalidRegexCaptureGroup() {
     connectorConfig = new HdfsSinkConnectorConfig(properties);
   }
 
-  @Test(expected = ConfigException.class)
-  public void testInvalidRegexCaptureGroupDoesntMatchTopic() {
+  @Test
+  public void testRegexDoesNotFullyMatchButTopicDirUsesOnlyTopicPlaceholder() {
     String topic = "topica";
-    String topicDir = "topic.another.${topic}.again";
+    String topicDirPattern = "logs/${topic}/data";
+    properties.put(HdfsSinkConnectorConfig.TOPIC_CAPTURE_GROUPS_REGEX_CONFIG, "[a-z]+");
+    properties.put(StorageCommonConfig.TOPICS_DIR_CONFIG, topicDirPattern);
+    connectorConfig = new HdfsSinkConnectorConfig(properties);
 
-    properties.put(HdfsSinkConnectorConfig.TOPIC_CAPTURE_GROUPS_REGEX_CONFIG, "[a-z]");
-    properties.put(StorageCommonConfig.TOPICS_DIR_CONFIG, topicDir);
+    String expectedDir = "logs/topica/data";
+    assertEquals(expectedDir, connectorConfig.getTopicsDirFromTopic(topic));
+  }
+
+  @Test
+  public void testRegexDoesNotMatchAndTopicDirUsesCaptureGroupsShouldThrow() {
+    String topic = "this_topic_will_not_match_pattern";
+    String topicDirPattern = "groups/${1}/data";
+    properties.put(HdfsSinkConnectorConfig.TOPIC_CAPTURE_GROUPS_REGEX_CONFIG, "(\\d+)");
+    properties.put(StorageCommonConfig.TOPICS_DIR_CONFIG, topicDirPattern);
     connectorConfig = new HdfsSinkConnectorConfig(properties);
-    connectorConfig.getTopicsDirFromTopic(topic);
+    ConfigException ex = assertThrows(ConfigException.class, () -> {
+      connectorConfig.getTopicsDirFromTopic(topic);
+    });
+
+    assertTrue(
+        "Exception message should indicate topic-regex mismatch",
+        ex.getMessage().contains("does not fully match the specified regex")
+            || ex.getMessage().contains("Requested regex group 1 is not available")
+    );
+  }
+
+  @Test
+  public void testRegexMatchesButTopicDirUsesNonExistentCaptureGroupShouldThrow() {
+    String topic = "data-set-alpha";
+    String topicDirPattern = "output/${1}/${2}";
+
+    properties.put(HdfsSinkConnectorConfig.TOPIC_CAPTURE_GROUPS_REGEX_CONFIG, "data-set-([a-z]+)");
+    properties.put(StorageCommonConfig.TOPICS_DIR_CONFIG, topicDirPattern);
+    connectorConfig = new HdfsSinkConnectorConfig(properties);
+
+    ConfigException ex = assertThrows(ConfigException.class, () -> {
+      connectorConfig.getTopicsDirFromTopic(topic);
+    });
+    assertTrue(
+        "Exception message should indicate missing regex group",
+        ex.getMessage().contains("actually had 1 capture groups")
+    );
   }
 
   @Test(expected = ConfigException.class)