S3 source connector

.github/workflows/codeql-analysis.yml

@@ -13,10 +13,14 @@ name: "CodeQL"
 
 on:
   push:
-    branches: [main]
+    branches:
+      - main
+      - s3-source-release
   pull_request:
     # The branches below must be a subset of the branches above
-    branches: [main]
+    branches:
+      - main
+      - s3-source-release
   schedule:
     - cron: "42 20 * * 6"
 

.github/workflows/main_push_workflow.yml

@@ -2,9 +2,13 @@
 name: Main and pull request checks
 on:
   push:
-    branches: [ main ]
+    branches:
+      - main
+      - s3-source-release
   pull_request:
-    branches: [ main ]
+    branches:
+      - main
+      - s3-source-release
 jobs:
   build:
     strategy:
@@ -30,4 +34,4 @@ jobs:
         run: ./gradlew build test
       - name: Build in Linux
         if: runner.os == 'Linux'
-        run: ./gradlew build check test integrationTest
+        run: ./gradlew build check test integrationTest -i

README.md

@@ -6,6 +6,7 @@
 - [Aiven GCS Sink Connector](./gcs-sink-connector/README.md)
 - [Aiven S3 Sink Connector](./s3-sink-connector/README.md)
 - [Aiven Azure Blob Sink Connector](./azure-sink-connector/README.md)
+- [Aiven S3 Source Connector](./s3-source-connector/README.md)
 
 # Development
 

commons/build.gradle.kts

@@ -27,7 +27,7 @@ dependencies {
   implementation(confluent.kafka.connect.avro.data) {
     exclude(group = "org.apache.kafka", module = "kafka-clients")
   }
-
+  implementation("commons-io:commons-io:2.18.0")
   implementation(tools.spotbugs.annotations)
   implementation(compressionlibs.snappy)
   implementation(compressionlibs.zstd.jni)
@@ -41,6 +41,7 @@ dependencies {
     exclude(group = "org.slf4j", module = "slf4j-api")
     exclude(group = "org.apache.avro", module = "avro")
   }
+
   implementation(apache.hadoop.common) {
     exclude(group = "org.apache.hadoop.thirdparty", module = "hadoop-shaded-protobuf_3_7")
     exclude(group = "com.google.guava", module = "guava")
@@ -86,11 +87,12 @@ dependencies {
   testImplementation(jackson.databind)
   testImplementation(testinglibs.mockito.core)
   testImplementation(testinglibs.assertj.core)
+  testImplementation(testinglibs.awaitility)
   testImplementation(testFixtures(project(":commons")))
-
   testImplementation(testinglibs.woodstox.stax2.api)
   testImplementation(apache.hadoop.mapreduce.client.core)
   testImplementation(confluent.kafka.connect.avro.converter)
+  testImplementation("org.mockito:mockito-junit-jupiter:5.14.2")
 
   testRuntimeOnly(testinglibs.junit.jupiter.engine)
   testRuntimeOnly(logginglibs.logback.classic)

commons/src/main/java/io/aiven/kafka/connect/common/config/CommonConfig.java

@@ -27,6 +27,8 @@
 public class CommonConfig extends AbstractConfig {
     protected static final String GROUP_COMPRESSION = "File Compression";
     protected static final String GROUP_FORMAT = "Format";
+    public static final String TASK_ID = "task.id";
+    public static final String MAX_TASKS = "tasks.max";
 
     /**
      * @deprecated No longer needed.
@@ -58,4 +60,25 @@ public Long getKafkaRetryBackoffMs() {
         return new BackoffPolicyConfig(this).getKafkaRetryBackoffMs();
     }
 
+    /**
+     *
+     * Get the maximum number of tasks that should be run by this connector configuration Max Tasks is set within the
+     * Kafka Connect framework and so is retrieved slightly differently in ConnectorConfig.java
+     *
+     * @return The maximum number of tasks that should be run by this connector configuration
+     */
+    public int getMaxTasks() {
+        // TODO when Connect framework is upgraded it will be possible to retrieve this information from the configDef
+        // as tasksMax
+        return Integer.parseInt(this.originalsStrings().get(MAX_TASKS));
+    }
+    /**
+     * Get the task id for this configuration
+     *
+     * @return The task id for this configuration
+     */
+    public int getTaskId() {
+        return Integer.parseInt(this.originalsStrings().get(TASK_ID));
+    }
+
 }

commons/src/main/java/io/aiven/kafka/connect/common/config/FileNameFragment.java

@@ -43,9 +43,12 @@ public final class FileNameFragment extends ConfigFragment {
     static final String FILE_MAX_RECORDS = "file.max.records";
     static final String FILE_NAME_TIMESTAMP_TIMEZONE = "file.name.timestamp.timezone";
     static final String FILE_NAME_TIMESTAMP_SOURCE = "file.name.timestamp.source";
-    static final String FILE_NAME_TEMPLATE_CONFIG = "file.name.template";
+    public static final String FILE_NAME_TEMPLATE_CONFIG = "file.name.template";
     static final String DEFAULT_FILENAME_TEMPLATE = "{{topic}}-{{partition}}-{{start_offset}}";
 
+    public static final String FILE_PATH_PREFIX_TEMPLATE_CONFIG = "file.prefix.template";
+    static final String DEFAULT_FILE_PATH_PREFIX_TEMPLATE = "topics/{{topic}}/partition={{partition}}/";
+
     public FileNameFragment(final AbstractConfig cfg) {
         super(cfg);
     }
@@ -109,9 +112,18 @@ public void ensureValid(final String name, final Object value) {
         configDef.define(FILE_NAME_TIMESTAMP_SOURCE, ConfigDef.Type.STRING, TimestampSource.Type.WALLCLOCK.name(),
                 new TimestampSourceValidator(), ConfigDef.Importance.LOW,
                 "Specifies the the timestamp variable source. Default is wall-clock.", GROUP_FILE, fileGroupCounter++, // NOPMD
-                // UnusedAssignment
                 ConfigDef.Width.SHORT, FILE_NAME_TIMESTAMP_SOURCE);
 
+        configDef.define(FILE_PATH_PREFIX_TEMPLATE_CONFIG, ConfigDef.Type.STRING, DEFAULT_FILE_PATH_PREFIX_TEMPLATE,
+                new ConfigDef.NonEmptyString(), ConfigDef.Importance.MEDIUM,
+                "The template for file prefix on S3. "
+                        + "Supports `{{ variable }}` placeholders for substituting variables. "
+                        + "Currently supported variables are `topic` and `partition` "
+                        + "and are mandatory to have these in the directory structure."
+                        + "Example prefix : topics/{{topic}}/partition/{{partition}}/",
+                GROUP_FILE, fileGroupCounter++, // NOPMD UnusedAssignment
+                ConfigDef.Width.LONG, FILE_PATH_PREFIX_TEMPLATE_CONFIG);
+
         return configDef;
     }
 
@@ -185,4 +197,8 @@ public int getMaxRecordsPerFile() {
         return cfg.getInt(FILE_MAX_RECORDS);
     }
 
+    public String getFilePathPrefixTemplateConfig() {
+        return cfg.getString(FILE_PATH_PREFIX_TEMPLATE_CONFIG);
+    }
+
 }

commons/src/main/java/io/aiven/kafka/connect/common/config/SchemaRegistryFragment.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2024 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.connect.common.config;
+
+import java.util.Locale;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+
+import io.aiven.kafka.connect.common.source.input.InputFormat;
+
+public final class SchemaRegistryFragment extends ConfigFragment {
+    private static final String SCHEMAREGISTRY_GROUP = "Schema registry group";
+    public static final String SCHEMA_REGISTRY_URL = "schema.registry.url";
+    public static final String VALUE_CONVERTER_SCHEMA_REGISTRY_URL = "value.converter.schema.registry.url";
+    public static final String AVRO_VALUE_SERIALIZER = "value.serializer";
+    public static final String INPUT_FORMAT_KEY = "input.format";
+    public static final String SCHEMAS_ENABLE = "schemas.enable";
+
+    /**
+     * Construct the ConfigFragment..
+     *
+     * @param cfg
+     *            the configuration that this fragment is associated with.
+     */
+    public SchemaRegistryFragment(final AbstractConfig cfg) {
+        super(cfg);
+    }
+
+    public static ConfigDef update(final ConfigDef configDef) {
+        int srCounter = 0;
+        configDef.define(SCHEMA_REGISTRY_URL, ConfigDef.Type.STRING, null, new ConfigDef.NonEmptyString(),
+                ConfigDef.Importance.MEDIUM, "SCHEMA REGISTRY URL", SCHEMAREGISTRY_GROUP, srCounter++,
+                ConfigDef.Width.NONE, SCHEMA_REGISTRY_URL);
+        configDef.define(VALUE_CONVERTER_SCHEMA_REGISTRY_URL, ConfigDef.Type.STRING, null,
+                new ConfigDef.NonEmptyString(), ConfigDef.Importance.MEDIUM, "SCHEMA REGISTRY URL",
+                SCHEMAREGISTRY_GROUP, srCounter++, ConfigDef.Width.NONE, VALUE_CONVERTER_SCHEMA_REGISTRY_URL);
+        configDef.define(INPUT_FORMAT_KEY, ConfigDef.Type.STRING, InputFormat.BYTES.getValue(),
+                new ConfigDef.NonEmptyString(), ConfigDef.Importance.MEDIUM,
+                "Input format of messages read from source avro/json/parquet/bytes", SCHEMAREGISTRY_GROUP, srCounter++, // NOPMD
+                ConfigDef.Width.NONE, INPUT_FORMAT_KEY);
+
+        configDef.define(AVRO_VALUE_SERIALIZER, ConfigDef.Type.CLASS, null, ConfigDef.Importance.MEDIUM,
+                "Avro value serializer", SCHEMAREGISTRY_GROUP, srCounter++, // NOPMD
+                // UnusedAssignment
+                ConfigDef.Width.NONE, AVRO_VALUE_SERIALIZER);
+        return configDef;
+    }
+
+    public InputFormat getInputFormat() {
+        return InputFormat.valueOf(cfg.getString(INPUT_FORMAT_KEY).toUpperCase(Locale.ROOT));
+    }
+
+    public String getSchemaRegistryUrl() {
+        return cfg.getString(SCHEMA_REGISTRY_URL);
+    }
+
+    public Class<?> getAvroValueSerializer() {
+        return cfg.getClass(AVRO_VALUE_SERIALIZER);
+    }
+
+}

commons/src/main/java/io/aiven/kafka/connect/common/config/SourceCommonConfig.java

@@ -20,8 +20,66 @@
 
 import org.apache.kafka.common.config.ConfigDef;
 
+import io.aiven.kafka.connect.common.config.enums.ErrorsTolerance;
+import io.aiven.kafka.connect.common.source.input.InputFormat;
+import io.aiven.kafka.connect.common.source.input.Transformer;
+import io.aiven.kafka.connect.common.source.input.TransformerFactory;
+import io.aiven.kafka.connect.common.source.task.DistributionType;
+
 public class SourceCommonConfig extends CommonConfig {
+
+    private final SchemaRegistryFragment schemaRegistryFragment;
+    private final SourceConfigFragment sourceConfigFragment;
+    private final FileNameFragment fileNameFragment;
+    private final OutputFormatFragment outputFormatFragment;
+
     public SourceCommonConfig(ConfigDef definition, Map<?, ?> originals) {// NOPMD
         super(definition, originals);
+        // Construct Fragments
+        schemaRegistryFragment = new SchemaRegistryFragment(this);
+        sourceConfigFragment = new SourceConfigFragment(this);
+        fileNameFragment = new FileNameFragment(this);
+        outputFormatFragment = new OutputFormatFragment(this);
+
+        validate(); // NOPMD ConstructorCallsOverridableMethod
+    }
+
+    private void validate() {
+        schemaRegistryFragment.validate();
+        sourceConfigFragment.validate();
+        fileNameFragment.validate();
+        outputFormatFragment.validate();
+    }
+
+    public InputFormat getInputFormat() {
+        return schemaRegistryFragment.getInputFormat();
+    }
+
+    public String getSchemaRegistryUrl() {
+        return schemaRegistryFragment.getSchemaRegistryUrl();
+    }
+
+    public String getTargetTopics() {
+        return sourceConfigFragment.getTargetTopics();
+    }
+    public String getTargetTopicPartitions() {
+        return sourceConfigFragment.getTargetTopicPartitions();
+    }
+
+    public ErrorsTolerance getErrorsTolerance() {
+        return sourceConfigFragment.getErrorsTolerance();
+    }
+
+    public DistributionType getDistributionType() {
+        return sourceConfigFragment.getDistributionType();
     }
+
+    public int getMaxPollRecords() {
+        return sourceConfigFragment.getMaxPollRecords();
+    }
+
+    public Transformer getTransformer() {
+        return TransformerFactory.getTransformer(schemaRegistryFragment.getInputFormat());
+    }
+
 }