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

wanghd89 · 2023-08-24T07:26:51Z

Description

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

Issues Resolved

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

…ctors Signed-off-by: Haidong <[email protected]>

Signed-off-by: Haidong <[email protected]>

sharraj · 2023-08-24T15:21:35Z

@engechas @dinujoh Please help with review.

sharraj · 2023-08-24T15:23:00Z

@hshardeesi @kkondaka as well to help with review.

kkondaka · 2023-08-26T01:01:53Z

build.gradle

@@ -143,13 +143,13 @@ subprojects {
            }
            implementation('org.eclipse.jetty:jetty-http') {
                version {
-                    require '11.0.15'
+                    require '9.4.48.v20220622'


why are going to an old revision?

Kafka Connect doesn't work with 11.0.15, the latest one is with 9.4.48.v20220622.
I mentioned this as the first thing in the high-level design, and has been approved by Rajs and Dinu

This could be a bit of a problem.

Also, why does this even need Jetty? Perhaps we can disable the admin server?

Kafka Connect needs it

the newer version is 9.4.51.v20230217

kkondaka · 2023-08-26T01:09:53Z

...plugins/src/main/java/org/opensearch/dataprepper/plugins/kafkaconnect/util/KafkaConnect.java

+ * Unique with single instance for each pipeline.
+ */
+public class KafkaConnect {
+    public static final long CONNECTOR_TIMEOUT_MS = 30000L; // 30 seconds


We should make this and the other statics below configurable.

Sounds like a good idea, I can take these two as config.

kkondaka · 2023-08-28T07:22:47Z

@wanghd89 I have a general question. How do we support codecs (like avro, json, protobuf, etc) with the Kafka connect?

wanghd89 · 2023-08-29T12:41:45Z

@wanghd89 I have a general question. How do we support codecs (like avro, json, protobuf, etc) with the Kafka connect?

Kafka Connect support those using key.converter/value.converter. It has its own converter classes.
But in our current phase (Debezium connector to read from MySQL/Postges/Mongo), JSON is the only one we support now. Because for Mysql and postgres, it read from binlog and can translate to JSON format, and mongoDB is as JSON already.

If we start bringing more Kafka Connectors who read data in other codecs, Kafka Connect has configurations support for those.

kkondaka · 2023-08-29T23:34:53Z

@wanghd89 please fix the code verification failure.

dlvenable · 2023-08-24T14:54:03Z

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

+    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.48.v20220622'


Can we exclude Jetty?

We cannot, it has to run a REST API by Kafka Connect

dlvenable · 2023-08-31T15:09:41Z

build.gradle

@@ -143,13 +143,13 @@ subprojects {
            }
            implementation('org.eclipse.jetty:jetty-http') {
                version {
-                    require '11.0.15'
+                    require '9.4.48.v20220622'


This could be a bit of a problem.

Also, why does this even need Jetty? Perhaps we can disable the admin server?

dlvenable · 2023-08-31T15:10:31Z

...per-core/src/main/java/org/opensearch/dataprepper/parser/model/DataPrepperConfiguration.java

@@ -85,9 +86,11 @@ public DataPrepperConfiguration(
            final Duration sinkShutdownTimeout,
            @JsonProperty("circuit_breakers") final CircuitBreakerConfig circuitBreakerConfig,
            @JsonProperty("source_coordination") final SourceCoordinationConfig sourceCoordinationConfig,
+            @JsonProperty("kafka_cluster_config") final KafkaClusterConfig kafkaClusterConfig,


This is a perfect place to use Data Prepper extensions.

See my comment on your issue: #3233 (comment)

dlvenable · 2023-08-31T15:13:54Z

...src/main/java/org/opensearch/dataprepper/plugins/kafkaconnect/source/KafkaConnectSource.java

+ */
+
+@SuppressWarnings("deprecation")
+@DataPrepperPlugin(name = "kafka_connect", pluginType = Source.class, pluginConfigurationType = KafkaConnectSourceConfig.class)


Please see this comment:

#3233 (comment)

As noted there, we should have different sources for each of the databases: Postgresql, MySQL, MongoDB.

A good way to solve this would be to add a Kafka Connect extensions. See the aws-plugin-api and aws-plugin projects for examples.

You could have a kafka-connect-api project which provides a basic API. Then, the mysql-source project would depend on it. (Similar for postgres-source, and mongodb-source).

The implementation can be an extensions in a project such as kakfa-connect-extension.

dlvenable · 2023-08-31T15:14:20Z

settings.gradle

@@ -36,6 +36,8 @@ dependencyResolutionManagement {
            library('spring-context', 'org.springframework', 'spring-context').versionRef('spring')
            version('guava', '32.0.1-jre')
            library('guava-core', 'com.google.guava', 'guava').versionRef('guava')
+            version('reflections', '0.9.12')


I'm a bit concerned about downgrading such a core component.

dlvenable · 2023-08-31T15:16:08Z

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;


Let's keep these out of data-prepper-api. We should reserve this project for the most common APIs, not specific technologies.

You can make use of extensions, or perhaps add a kafka-common project if you want to share with the kafka source/sink.

dlvenable · 2023-08-31T15:16:52Z

.../src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaSourceConfig.java

+        this.encryptionConfig = encryptionConfig;
+    }
+
+//    @Override


Please remove all commented-out code.

dlvenable · 2023-08-31T15:18:39Z

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

@@ -0,0 +1,30 @@
+package org.opensearch.dataprepper.plugins.sink;


All new files need a header. Use the following if the code is exclusively new code:

/* * Copyright OpenSearch Contributors * SPDX-License-Identifier: Apache-2.0 */

There is another if you are pulling from another open-source project. Let us know if you need it.

wanghd89 · 2023-09-07T08:50:02Z

There is major change for this PR Due to the introduction of Extension Plugins.
I closed this PR and please review at new place: #3313

Haidong added 2 commits August 24, 2023 14:14

Integrate with Kafka Connect and Debezium MySQL/Postres/MongoDb Conne…

220aaee

…ctors Signed-off-by: Haidong <[email protected]>

add readme

07d10b9

Signed-off-by: Haidong <[email protected]>

wanghd89 requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, asifsmohammed, dlvenable and oeyh as code owners August 24, 2023 07:26

kkondaka reviewed Aug 28, 2023

View reviewed changes

dlvenable requested changes Aug 31, 2023

View reviewed changes

wanghd89 closed this Sep 7, 2023

wanghd89 deleted the main branch October 24, 2023 01:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

wanghd89 commented Aug 24, 2023 •

edited

Loading

sharraj commented Aug 24, 2023

sharraj commented Aug 24, 2023

kkondaka Aug 26, 2023

wanghd89 Aug 29, 2023

dlvenable Aug 31, 2023

wanghd89 Aug 31, 2023

wanghd89 Aug 31, 2023

kkondaka Aug 26, 2023

wanghd89 Aug 29, 2023

kkondaka commented Aug 28, 2023

wanghd89 commented Aug 29, 2023

kkondaka commented Aug 29, 2023

dlvenable Aug 24, 2023

wanghd89 Aug 31, 2023

dlvenable Aug 31, 2023

dlvenable Aug 31, 2023

dlvenable Aug 31, 2023

dlvenable Aug 31, 2023

dlvenable Aug 31, 2023

dlvenable Aug 31, 2023

dlvenable Aug 31, 2023

wanghd89 commented Sep 7, 2023

		@@ -0,0 +1,30 @@
		package org.opensearch.dataprepper.plugins.sink;

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

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

Conversation

wanghd89 commented Aug 24, 2023 • edited Loading

Description

Issues Resolved

Check List

sharraj commented Aug 24, 2023

sharraj commented Aug 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kkondaka commented Aug 28, 2023

wanghd89 commented Aug 29, 2023

kkondaka commented Aug 29, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wanghd89 commented Sep 7, 2023

wanghd89 commented Aug 24, 2023 •

edited

Loading