Integrate CDC data from MySQL/Postgres/MongoDb data source

build.gradle

@@ -147,13 +147,13 @@ subprojects {
             }
             implementation('org.eclipse.jetty:jetty-http') {
                 version {
-                    require '11.0.15'
+                    require '9.4.51.v20230217'
                 }
                 because 'CVE from transitive dependencies'
             }
             implementation('org.eclipse.jetty:jetty-server') {
                 version {
-                    require '11.0.15'
+                    require '9.4.51.v20230217'
                 }
                 because 'CVE from transitive dependencies'
             }

data-prepper-api/src/main/java/org/opensearch/dataprepper/metrics/PluginMetrics.java

@@ -8,6 +8,7 @@
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Timer;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 
@@ -78,8 +79,11 @@ public <T> T gauge(String name, T obj, ToDoubleFunction<T> valueFunction) {
         return Metrics.gauge(getMeterName(name), obj, valueFunction);
     }
 
+    public <T> T gaugeWithTags(String name, Iterable<Tag> tags, T obj, ToDoubleFunction<T> valueFunction) {
+        return Metrics.gauge(getMeterName(name), tags, obj, valueFunction);
+    }
+
     private String getMeterName(final String name) {
         return new StringJoiner(MetricNames.DELIMITER).add(metricsPrefix).add(name).toString();
     }
-
 }

data-prepper-api/src/test/java/org/opensearch/dataprepper/metrics/PluginMetricsTest.java

@@ -17,6 +17,7 @@
 import java.util.StringJoiner;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static java.util.Collections.emptyList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -125,6 +126,18 @@ public void testReferenceGauge() {
         assertEquals(3, gauge.length());
     }
 
+    @Test
+    public void testReferenceGaugeWithTags() {
+        final String testString = "abc";
+        final String gauge = objectUnderTest.gaugeWithTags("gauge", emptyList(), testString, String::length);
+        assertNotNull(
+                Metrics.globalRegistry.get(new StringJoiner(MetricNames.DELIMITER)
+                        .add(PIPELINE_NAME).add(PLUGIN_NAME)
+                        .add("gauge").toString()).meter());
+        assertEquals(3, gauge.length());
+    }
+
+
     @Test
     public void testEmptyPipelineName() {
         assertThrows(

data-prepper-core/build.gradle

@@ -20,7 +20,7 @@ dependencies {
     testImplementation project(':data-prepper-plugins:common').sourceSets.test.output
     implementation 'com.fasterxml.jackson.core:jackson-databind'
     implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
-    implementation "org.reflections:reflections:0.10.2"
+    implementation libs.reflections.core
     implementation 'io.micrometer:micrometer-core'
     implementation 'io.micrometer:micrometer-registry-prometheus'
     implementation 'io.micrometer:micrometer-registry-cloudwatch2'

data-prepper-plugins/common/build.gradle

@@ -17,7 +17,7 @@ dependencies {
     implementation libs.commons.lang3
     implementation libs.bouncycastle.bcprov
     implementation libs.bouncycastle.bcpkix
-    implementation 'org.reflections:reflections:0.10.2'
+    implementation libs.reflections.core
     implementation 'io.micrometer:micrometer-core'
     testImplementation testLibs.junit.vintage
     implementation 'org.apache.parquet:parquet-common:1.13.1'

data-prepper-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/sink/NoopSink.java

@@ -0,0 +1,30 @@
+package org.opensearch.dataprepper.plugins.sink;
+
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.sink.Sink;
+
+import java.util.Collection;
+
+@DataPrepperPlugin(name = "noop", pluginType = Sink.class)
+public class NoopSink implements Sink<Record<Object>> {
+    @Override
+    public void output(Collection<Record<Object>> records) {
+        // empty by design.
+    }
+
+    @Override
+    public void shutdown() {
+        // empty by design.
+    }
+
+    @Override
+    public void initialize() {
+        // empty by design.
+    }
+
+    @Override
+    public boolean isReady() {
+        return true;
+    }
+}

data-prepper-plugins/kafka-connect-plugins/build.gradle

@@ -0,0 +1,68 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+plugins {
+    id 'java'
+}
+
+dependencies {
+    implementation project(':data-prepper-plugins:common')
+    implementation project(':data-prepper-plugins:kafka-plugins')
+    implementation 'org.apache.kafka:connect-runtime:3.5.1'
+    implementation 'software.amazon.awssdk:sts'
+    implementation 'software.amazon.awssdk:secretsmanager'
+    implementation 'javax.validation:validation-api:2.0.1.Final'
+    implementation libs.reflections.core
+    implementation 'io.micrometer:micrometer-core'
+    implementation 'org.eclipse.jetty:jetty-server:9.4.51.v20230217'
+    implementation 'org.eclipse.jetty:jetty-servlet:9.4.51.v20230217'
+    implementation 'org.eclipse.jetty:jetty-servlets:9.4.51.v20230217'
+    implementation 'org.eclipse.jetty:jetty-client:9.4.51.v20230217'
+    implementation ('io.confluent:kafka-schema-registry:7.5.0') {
+        exclude group: 'org.glassfish.jersey.containers', module: 'jersey-container-servlet'
+        exclude group: 'org.glassfish.jersey.inject', module: 'jersey-hk2'
+        exclude group: 'org.glassfish.jersey.ext', module: 'jersey-bean-validation'
+    }
+    runtimeOnly 'org.jetbrains.kotlin:kotlin-compiler:1.8.21'
+    // Common Debezium Connector
+    implementation 'io.debezium:debezium-api:2.3.0.Final'
+    implementation 'io.debezium:debezium-core:2.3.0.Final'
+    implementation 'io.debezium:debezium-storage-kafka:2.3.0.Final'
+    implementation 'io.debezium:debezium-storage-file:2.3.0.Final'
+    // Debezium MySQL Connector
+    implementation 'org.antlr:antlr4-runtime:4.10.1'
+    implementation 'io.debezium:debezium-connector-mysql:2.3.0.Final'
+    implementation 'io.debezium:debezium-ddl-parser:2.3.0.Final'
+    implementation 'com.zendesk:mysql-binlog-connector-java:0.28.1'
+    implementation 'com.mysql:mysql-connector-j:8.0.33'
+    implementation 'com.github.luben:zstd-jni:1.5.0-2'
+    // Debezium Postgres connector
+    implementation 'io.debezium:debezium-connector-postgres:2.3.0.Final'
+    implementation 'org.postgresql:postgresql:42.5.1'
+    implementation 'com.google.protobuf:protobuf-java:3.19.6'
+    // Debezium Mongodb connector
+    implementation 'io.debezium:debezium-connector-mongodb:2.3.0.Final'
+    implementation 'org.mongodb:mongodb-driver-core:4.7.1'
+    implementation 'org.mongodb:mongodb-driver-sync:4.7.1'
+    implementation 'org.mongodb:bson:4.7.1'
+    runtimeOnly 'org.mongodb:bson-record-codec:4.7.1'
+    // test
+    testImplementation project(':data-prepper-test-common')
+    testImplementation project(':data-prepper-core')
+    testImplementation 'org.yaml:snakeyaml:2.0'
+    testImplementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.14.2'
+    testImplementation testLibs.mockito.inline
+}
+
+jacocoTestCoverageVerification {
+    dependsOn jacocoTestReport
+    violationRules {
+        rule { //in addition to core projects rule
+            limit {
+                minimum = 0.90
+            }
+        }
+    }
+}

data-prepper-plugins/kafka-connect-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafkaconnect/configuration/ConnectorConfig.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.kafkaconnect.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.opensearch.dataprepper.plugins.kafkaconnect.util.Connector;
+
+import java.util.List;
+import java.util.Properties;
+
+public abstract class ConnectorConfig {
+    @JsonProperty("force_update")
+    public Boolean forceUpdate = false;
+    private String bootstrapServers;
+    private Properties authProperties;
+
+    public abstract List<Connector> buildConnectors();
+
+    public Properties getAuthProperties() {
+        return this.authProperties;
+    }
+
+    public void setAuthProperties(Properties authProperties) {
+        this.authProperties = authProperties;
+    }
+
+    public String getBootstrapServers() {
+        return this.bootstrapServers;
+    }
+
+    public void setBootstrapServers(String bootStrapServers) {
+        this.bootstrapServers = bootStrapServers;
+    }
+}

data-prepper-plugins/kafka-connect-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafkaconnect/configuration/CredentialsConfig.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.kafkaconnect.configuration;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.opensearch.dataprepper.plugins.kafkaconnect.util.SecretManagerHelper;
+
+import java.util.Map;
+
+public class CredentialsConfig {
+    private final String username;
+    private final String password;
+
+    @JsonCreator
+    public CredentialsConfig(@JsonProperty("plaintext") final PlainText plainText,
+                             @JsonProperty("secret_manager") final SecretManager secretManager) {
+        if (plainText != null && secretManager != null) {
+            throw new IllegalArgumentException("plaintext and secret_manager cannot both be set");
+        }
+        if (plainText != null) {
+            if (plainText.username == null || plainText.password == null) {
+                throw new IllegalArgumentException("user and password must be set for plaintext credentials");
+            }
+            this.username = plainText.username;
+            this.password = plainText.password;
+        } else if (secretManager != null) {
+            if (secretManager.secretId == null || secretManager.region == null) {
+                throw new IllegalArgumentException("secretId and region must be set for aws credential type");
+            }
+            final Map<String, String> secretMap = this.getSecretValueMap(secretManager.stsRoleArn, secretManager.region, secretManager.secretId);
+            if (!secretMap.containsKey("username") || !secretMap.containsKey("password")) {
+                throw new RuntimeException("username or password missing in secret manager.");
+            }
+            this.username = secretMap.get("username");
+            this.password = secretMap.get("password");
+        } else {
+            throw new IllegalArgumentException("plaintext or secret_manager must be set");
+        }
+    }
+
+    private Map<String, String> getSecretValueMap(String stsRoleArn, String region, String secretId) {
+        ObjectMapper objectMapper = new ObjectMapper();
+        try {
+            final String secretValue = SecretManagerHelper.getSecretValue(stsRoleArn, region, secretId);
+            return objectMapper.readValue(secretValue, new TypeReference<>() {});
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to get credentials.", e);
+        }
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    static class PlainText {
+        private String username;
+        private String password;
+
+        @JsonCreator
+        public PlainText(@JsonProperty("username") String username,
+                         @JsonProperty("password") String password) {
+            this.username = username;
+            this.password = password;
+        }
+    }
+
+    static class SecretManager {
+        private String region;
+        private String secretId;
+        private String stsRoleArn;
+
+        @JsonCreator
+        public SecretManager(@JsonProperty("sts_role_arn") String stsRoleArn,
+                             @JsonProperty("region") String region,
+                             @JsonProperty("secretId") String secretId) {
+            this.stsRoleArn = stsRoleArn;
+            this.region = region;
+            this.secretId = secretId;
+        }
+    }
+}