Merge pull request #297 from stephen-harris/feat/add-support-for-KeyAndTopicPartitionRecordGrouper

Bump commons-for-apache-kafka-connect to 0.11.0

Author: jeqo

build.gradle

@@ -80,7 +80,7 @@ ext {
     confluentPlatformVersion = "4.1.4"
     // Align with version used by commons
     avroConverterVersion = "7.2.2"
-    aivenConnectCommonsVersion = "0.10.1"
+    aivenConnectCommonsVersion = "0.11.0"
 
     amazonS3Version = "1.12.520"
     amazonSTSVersion = "1.12.519"

src/main/java/io/aiven/kafka/connect/s3/config/S3SinkConfig.java

@@ -45,7 +45,6 @@
 import io.aiven.kafka.connect.common.config.validators.TimeZoneValidator;
 import io.aiven.kafka.connect.common.config.validators.TimestampSourceValidator;
 import io.aiven.kafka.connect.common.config.validators.UrlValidator;
-import io.aiven.kafka.connect.common.grouper.RecordGrouperFactory;
 import io.aiven.kafka.connect.common.templating.Template;
 import io.aiven.kafka.connect.s3.S3OutputStream;
 
@@ -726,15 +725,6 @@ private void validate() {
             );
         }
 
-        // Special checks for {{key}} filename template.
-        final Template filenameTemplate = getFilenameTemplate();
-        if (RecordGrouperFactory.KEY_RECORD.equals(RecordGrouperFactory.resolveRecordGrouperType(filenameTemplate))) {
-            if (getMaxRecordsPerFile() > 1) {
-                final String msg = String.format("When %s is %s, %s must be either 1 or not set",
-                    FILE_NAME_TEMPLATE_CONFIG, filenameTemplate, FILE_MAX_RECORDS);
-                throw new ConfigException(msg);
-            }
-        }
     }
 
     public AwsAccessSecret getOldAwsCredentials() {

src/test/java/io/aiven/kafka/connect/s3/config/S3SinkConfigTest.java

@@ -934,4 +934,28 @@ void stsEndpointShouldNotBeSetWithoutRegion() {
         );
 
     }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"{{key}}", "{{topic}}/{{partition}}/{{key}}"})
+    void notSupportedFileMaxRecords(final String fileNameTemplate) {
+        final Map<String, String> properties =
+            Map.of(
+                S3SinkConfig.FILE_NAME_TEMPLATE_CONFIG, fileNameTemplate,
+                S3SinkConfig.FILE_MAX_RECORDS, "2",
+                S3SinkConfig.AWS_ACCESS_KEY_ID_CONFIG, "any_access_key_id",
+                S3SinkConfig.AWS_SECRET_ACCESS_KEY_CONFIG, "any_secret_key",
+                S3SinkConfig.AWS_S3_BUCKET_NAME_CONFIG, "any_bucket"
+            );
+        final Throwable t = assertThrows(
+            ConfigException.class,
+            () -> new S3SinkConfig(properties)
+        );
+        assertEquals(
+            String.format(
+                "When file.name.template is %s, file.max.records must be either 1 or not set", 
+                fileNameTemplate
+            ),
+            t.getMessage());
+    }
+
 }