Integrate CDC data from MySQL/Postgres/MongoDb data source (opensearc…

…h-project#3313) Signed-off-by: Haidong <[email protected]>
daixba · Oct 12, 2023 · 5b822f3 · 5b822f3
1 parent 633401a
commit 5b822f3
Show file tree

Hide file tree

Showing 70 changed files with 4,045 additions and 32 deletions.
diff --git a/build.gradle b/build.gradle
@@ -89,7 +89,7 @@ subprojects {
     }
     dependencies {
         implementation platform('com.fasterxml.jackson:jackson-bom:2.15.0')
-        implementation platform('org.eclipse.jetty:jetty-bom:11.0.17')
+        implementation platform('org.eclipse.jetty:jetty-bom:9.4.53.v20231009')
         implementation platform('io.micrometer:micrometer-bom:1.10.5')
         implementation libs.guava.core
         implementation libs.slf4j.api
@@ -154,13 +154,13 @@ subprojects {
             }
             implementation('org.eclipse.jetty:http2-common') {
                 version {
-                    require '11.0.17'
+                    require '9.4.53.v20231009'
                 }
                 because 'Fixes CVE-2023-44487'
             }
             implementation('org.eclipse.jetty:http2-server') {
                 version {
-                    require '11.0.17'
+                    require '9.4.53.v20231009'
                 }
                 because 'Fixes CVE-2023-44487'
             }

diff --git a/data-prepper-api/src/main/java/org/opensearch/dataprepper/metrics/PluginMetrics.java b/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();
     }
-
 }
diff --git a/data-prepper-api/src/test/java/org/opensearch/dataprepper/metrics/PluginMetricsTest.java b/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(

diff --git a/data-prepper-core/build.gradle b/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'

diff --git a/data-prepper-plugins/common/build.gradle b/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'

diff --git a/...repper-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/sink/NoopSink.java b/...repper-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;
+    }
+}