Integrate with Kafka Connect and Debezium MySQL/Postgres/MongoDb plugin for CDC

build.gradle

@@ -143,13 +143,13 @@ subprojects {
             }
             implementation('org.eclipse.jetty:jetty-http') {
                 version {
-                    require '11.0.15'
+                    require '9.4.48.v20220622'
                 }
                 because 'CVE from transitive dependencies'
             }
             implementation('org.eclipse.jetty:jetty-server') {
                 version {
-                    require '11.0.15'
+                    require '9.4.48.v20220622'
                 }
                 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-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/AuthConfig.java

@@ -3,20 +3,42 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
-import jakarta.validation.constraints.AssertTrue;
-
-import java.util.stream.Stream;
 
 /**
  * A helper class that helps to read auth related configuration values from
  * pipelines.yaml
  */
 public class AuthConfig {
 
+    @JsonProperty("sasl")
+    private SaslAuthConfig saslAuthConfig;
+
+    /*
+     * TODO
+        public static class SslAuthConfig {
+            // TODO Add Support for SSL authentication types like
+            // one-way or two-way authentication
+
+            public SslAuthConfig() {
+            }
+        }
+
+        @JsonProperty("ssl")
+        private SslAuthConfig sslAuthConfig;
+
+        public SslAuthConfig getSslAuthConfig() {
+            return sslAuthConfig;
+        }
+
+    */
+
+    public SaslAuthConfig getSaslAuthConfig() {
+        return saslAuthConfig;
+    }
+
     public static class SaslAuthConfig {
         @JsonProperty("plaintext")
         private PlainTextAuthConfig plainTextAuthConfig;
@@ -45,40 +67,6 @@ public OAuthConfig getOAuthConfig() {
         public String getSslEndpointAlgorithm() {
             return sslEndpointAlgorithm;
         }
-
-        @AssertTrue(message = "Only one of AwsIam or oAuth or PlainText auth config must be specified")
-        public boolean hasOnlyOneConfig() {
-            return Stream.of(awsIamAuthConfig, plainTextAuthConfig, oAuthConfig).filter(n -> n != null).count() == 1;
-        }
-
-    }
-
-
-    /*
-     * TODO 
-        public static class SslAuthConfig {
-            // TODO Add Support for SSL authentication types like
-            // one-way or two-way authentication
-
-            public SslAuthConfig() {
-            }
-        }
-
-        @JsonProperty("ssl")
-        private SslAuthConfig sslAuthConfig;
-
-        public SslAuthConfig getSslAuthConfig() {
-            return sslAuthConfig;
-        }
-
-    */
-
-    @Valid
-    @JsonProperty("sasl")
-    private SaslAuthConfig saslAuthConfig;
-
-    public SaslAuthConfig getSaslAuthConfig() {
-        return saslAuthConfig;
     }
 
     /*

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AwsConfig.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/AwsConfig.java

@@ -3,17 +3,13 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.constraints.Size;
-import jakarta.validation.Valid;
 
 public class AwsConfig {
 
     public static class AwsMskConfig {
-        @Valid
-        @Size(min = 20, max = 2048, message = "msk_arn length should be between 20 and 2048 characters")
         @JsonProperty("arn")
         private String arn;
 
@@ -32,13 +28,9 @@ public MskBrokerConnectionType getBrokerConnectionType() {
     @JsonProperty("msk")
     private AwsMskConfig awsMskConfig;
 
-    @Valid
-    @Size(min = 1, message = "Region cannot be empty string")
     @JsonProperty("region")
     private String region;
 
-    @Valid
-    @Size(min = 20, max = 2048, message = "sts_role_arn length should be between 20 and 2048 characters")
     @JsonProperty("sts_role_arn")
     private String stsRoleArn;
 

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AwsIamAuthConfig.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/AwsIamAuthConfig.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import java.util.Map;
@@ -28,7 +28,7 @@ public enum AwsIamAuthConfig {
     }
 
     @JsonCreator
-    static AwsIamAuthConfig fromOptionValue(final String option) {
+    public static AwsIamAuthConfig fromOptionValue(final String option) {
         return OPTIONS_MAP.get(option.toLowerCase());
     }
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/EncryptionConfig.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.plugin.kafka;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class EncryptionConfig {
+    @JsonProperty("type")
+    private EncryptionType type = EncryptionType.SSL;
+
+    @JsonProperty("insecure")
+    private boolean insecure = false;
+
+    public EncryptionType getType() {
+        return type;
+    }
+
+    public boolean getInsecure() {
+        return insecure;
+    }
+}

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/EncryptionType.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/EncryptionType.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 
@@ -28,7 +28,7 @@ public enum EncryptionType {
     }
 
     @JsonCreator
-    static EncryptionType fromTypeValue(final String type) {
+    public static EncryptionType fromTypeValue(final String type) {
         return OPTIONS_MAP.get(type.toLowerCase());
     }
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/KafkaClusterAuthConfig.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.plugin.kafka;
+
+public interface KafkaClusterAuthConfig {
+    AwsConfig getAwsConfig();
+
+    AuthConfig getAuthConfig();
+
+    EncryptionConfig getEncryptionConfig();
+
+    String getBootStrapServers();
+}

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/MskBrokerConnectionType.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/MskBrokerConnectionType.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 
@@ -29,7 +29,7 @@ public enum MskBrokerConnectionType {
     }
 
     @JsonCreator
-    static MskBrokerConnectionType fromTypeValue(final String type) {
+    public static MskBrokerConnectionType fromTypeValue(final String type) {
         return OPTIONS_MAP.get(type.toLowerCase());
     }
 }

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/OAuthConfig.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/OAuthConfig.java

@@ -2,7 +2,7 @@
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
  */
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 

data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/PlainTextAuthConfig.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/PlainTextAuthConfig.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.kafka.configuration;
+package org.opensearch.dataprepper.model.plugin.kafka;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/plugin/kafka/UsesKafkaClusterConfig.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.plugin.kafka;
+
+import java.util.List;
+
+/**
+ * An interface that a plugin who uses Kafka Cluster will implement when kafka_cluster_config is configured
+ */
+public interface UsesKafkaClusterConfig {
+
+    void setBootstrapServers(final List<String> bootstrapServers);
+
+    void setKafkaClusterAuthConfig(final AuthConfig authConfig,
+                                   final AwsConfig awsConfig,
+                                   final EncryptionConfig encryptionConfig);
+}

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-core/src/main/java/org/opensearch/dataprepper/parser/PipelineParser.java

@@ -16,11 +16,13 @@
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.peerforwarder.RequiresPeerForwarding;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
+import org.opensearch.dataprepper.model.plugin.kafka.UsesKafkaClusterConfig;
 import org.opensearch.dataprepper.model.processor.Processor;
 import org.opensearch.dataprepper.model.sink.Sink;
 import org.opensearch.dataprepper.model.source.Source;
 import org.opensearch.dataprepper.model.sink.SinkContext;
 import org.opensearch.dataprepper.parser.model.DataPrepperConfiguration;
+import org.opensearch.dataprepper.parser.model.KafkaClusterConfig;
 import org.opensearch.dataprepper.parser.model.PipelineConfiguration;
 import org.opensearch.dataprepper.parser.model.SinkContextPluginSetting;
 import org.opensearch.dataprepper.peerforwarder.PeerForwarderConfiguration;
@@ -233,6 +235,8 @@ private void buildPipelineFromConfiguration(
 
             final Router router = routerFactory.createRouter(pipelineConfiguration.getRoutes());
 
+            configPluginsUsesKafka(source, buffer, decoratedProcessorSets, sinks);
+
             final Pipeline pipeline = new Pipeline(pipelineName, source, buffer, decoratedProcessorSets, sinks, router,
                     eventFactory, acknowledgementSetManager, sourceCoordinatorFactory, processorThreads, readBatchDelay,
                     dataPrepperConfiguration.getProcessorShutdownTimeout(), dataPrepperConfiguration.getSinkShutdownTimeout(),
@@ -247,6 +251,20 @@ private void buildPipelineFromConfiguration(
 
     }
 
+    private void configPluginsUsesKafka(final Source source,
+                                        final Buffer buffer,
+                                        final List<List<Processor>> decoratedProcessorSets,
+                                        final List<DataFlowComponent<Sink>> sinks) {
+        if (source instanceof UsesKafkaClusterConfig) {
+            final KafkaClusterConfig kafkaClusterConfig = dataPrepperConfiguration.getKafkaClusterConfig();
+            ((UsesKafkaClusterConfig) source).setBootstrapServers(kafkaClusterConfig.getBootStrapServers());
+            ((UsesKafkaClusterConfig) source).setKafkaClusterAuthConfig(
+                    kafkaClusterConfig.getAuthConfig(),
+                    kafkaClusterConfig.getAwsConfig(),
+                    kafkaClusterConfig.getEncryptionConfig());
+        }
+    }
+
     private List<IdentifiedComponent<Processor>> newProcessor(final PluginSetting pluginSetting) {
         final List<Processor> processors = pluginFactory.loadPlugins(
                 Processor.class,