This is an automated email from the ASF dual-hosted git repository.
ruanhang1993 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 3194adb15 [FLINK-39410][pipeline-connector][elasticsearch] The
`flink-cdc-pipeline-connector-elasticsearch` module offers better compatibility
with Flink 1.20 and Flink 2.x (#4368)
3194adb15 is described below
commit 3194adb15f6e3102bc5b30058baa921cadb7b299
Author: Pei Yu <[email protected]>
AuthorDate: Wed Sep 23 11:44:20 2026 +0800
[FLINK-39410][pipeline-connector][elasticsearch] The
`flink-cdc-pipeline-connector-elasticsearch` module offers better compatibility
with Flink 1.20 and Flink 2.x (#4368)
---
.../pipeline-connectors/elasticsearch.md | 2 +-
.../pipeline-connectors/elasticsearch.md | 2 +-
.../pom.xml | 11 +-
.../serializer/ElasticsearchEventSerializer.java | 6 -
.../sink/ElasticsearchDataSinkOptions.java | 2 +-
.../elasticsearch/v2/Elasticsearch8AsyncSink.java | 42 ++-
.../v2/Elasticsearch8AsyncWriter.java | 39 ++-
.../flink-cdc-pipeline-e2e-tests/pom.xml | 21 ++
.../stage2/MysqlToElasticsearch6E2eITCase.java | 329 ++++++++++++++++++
.../stage2/MysqlToElasticsearch7E2eITCase.java | 329 ++++++++++++++++++
.../stage2/MysqlToElasticsearch8E2eITCase.java | 380 +++++++++++++++++++++
.../connector/sink2/StatefulSinkWriterAdapter.java | 27 ++
.../connector/sink2/WriterInitContextAdapter.java | 94 +++++
.../connector/base/sink/AsyncSinkBaseAdapter.java | 68 ++++
.../base/sink/writer/AsyncSinkWriterAdapter.java | 42 +++
.../connector/sink2/StatefulSinkWriterAdapter.java | 27 ++
.../connector/sink2/WriterInitContextAdapter.java | 94 +++++
.../connector/base/sink/AsyncSinkBaseAdapter.java | 69 ++++
.../base/sink/writer/AsyncSinkWriterAdapter.java | 64 ++++
19 files changed, 1609 insertions(+), 39 deletions(-)
diff --git
a/docs/content.zh/docs/connectors/pipeline-connectors/elasticsearch.md
b/docs/content.zh/docs/connectors/pipeline-connectors/elasticsearch.md
index 0b260c761..203e6d29e 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/elasticsearch.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/elasticsearch.md
@@ -160,7 +160,7 @@ Pipeline Connector Options
<tr>
<td>record.size.max.bytes</td>
<td>optional</td>
- <td style="word-wrap: break-word;">10485760</td>
+ <td style="word-wrap: break-word;">5242880</td>
<td>Long</td>
<td>单个记录的最大大小（以byte为单位）。</td>
</tr>
diff --git a/docs/content/docs/connectors/pipeline-connectors/elasticsearch.md
b/docs/content/docs/connectors/pipeline-connectors/elasticsearch.md
index 3f3d693fe..60b316872 100644
--- a/docs/content/docs/connectors/pipeline-connectors/elasticsearch.md
+++ b/docs/content/docs/connectors/pipeline-connectors/elasticsearch.md
@@ -160,7 +160,7 @@ Pipeline Connector Options
<tr>
<td>record.size.max.bytes</td>
<td>optional</td>
- <td style="word-wrap: break-word;">10485760</td>
+ <td style="word-wrap: break-word;">5242880</td>
<td>Long</td>
<td>The maximum size of a single record in bytes.</td>
</tr>
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/pom.xml
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/pom.xml
index de0061449..1cb38ef05 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/pom.xml
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/pom.xml
@@ -35,7 +35,7 @@ limitations under the License.
<properties>
<elasticsearch.version>8.12.1</elasticsearch.version>
-
<flink.connector.elasticsearch.version>3.0.1-1.17</flink.connector.elasticsearch.version>
+
<flink.connector.elasticsearch.version>3.1.0-1.20</flink.connector.elasticsearch.version>
<httpclient.version>4.5.13</httpclient.version>
<jakarta.json.version>2.0.2</jakarta.json.version>
</properties>
@@ -195,4 +195,13 @@ limitations under the License.
</plugin>
</plugins>
</build>
+
+ <profiles>
+ <profile>
+ <id>flink2</id>
+ <properties>
+
<flink.connector.elasticsearch.version>4.0.0-2.0</flink.connector.elasticsearch.version>
+ </properties>
+ </profile>
+ </profiles>
</project>
\ No newline at end of file
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/serializer/ElasticsearchEventSerializer.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/serializer/ElasticsearchEventSerializer.java
index 800c0f359..afae409b5 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/serializer/ElasticsearchEventSerializer.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/serializer/ElasticsearchEventSerializer.java
@@ -17,7 +17,6 @@
package org.apache.flink.cdc.connectors.elasticsearch.serializer;
-import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.cdc.common.data.RecordData;
import org.apache.flink.cdc.common.event.AddColumnEvent;
@@ -251,9 +250,4 @@ public class ElasticsearchEventSerializer implements
ElementConverter<Event, Bul
throw new IndexOutOfBoundsException("Index: " + index + ", Size: "
+ size);
}
}
-
- @Override
- public void open(Sink.InitContext context) {
- ElementConverter.super.open(context);
- }
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/sink/ElasticsearchDataSinkOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/sink/ElasticsearchDataSinkOptions.java
index 213b446c8..45183a187 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/sink/ElasticsearchDataSinkOptions.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/sink/ElasticsearchDataSinkOptions.java
@@ -74,7 +74,7 @@ public class ElasticsearchDataSinkOptions {
public static final ConfigOption<Long> MAX_RECORD_SIZE_IN_BYTES =
ConfigOptions.key("record.size.max.bytes")
.longType()
- .defaultValue(10L * 1024L * 1024L)
+ .defaultValue(5L * 1024L * 1024L)
.withDescription("The maximum size of a single record in
bytes.");
/** The version of Elasticsearch to connect to. */
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncSink.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncSink.java
index 4a237e3a3..1250c94b4 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncSink.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncSink.java
@@ -20,14 +20,16 @@
package org.apache.flink.cdc.connectors.elasticsearch.v2;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.api.connector.sink2.InitContextAdapter;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriterAdapter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.connector.base.sink.AsyncSinkBaseAdapter;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
@@ -38,8 +40,7 @@ import java.util.Collections;
* @param <InputT> type of records that will be converted into {@link
Operation}. See {@link
* Elasticsearch8AsyncSinkBuilder} on how to construct valid instances.
*/
-public class Elasticsearch8AsyncSink<InputT> extends AsyncSinkBase<InputT,
Operation> {
- private static final Logger LOG =
LoggerFactory.getLogger(Elasticsearch8AsyncSink.class);
+public class Elasticsearch8AsyncSink<InputT> extends
AsyncSinkBaseAdapter<InputT, Operation> {
@VisibleForTesting protected final NetworkConfig networkConfig;
@@ -78,14 +79,13 @@ public class Elasticsearch8AsyncSink<InputT> extends
AsyncSinkBase<InputT, Opera
}
/**
- * Creates a new {@link StatefulSinkWriter} for writing elements to
Elasticsearch.
+ * Creates a new {@link SinkWriter} for writing elements to Elasticsearch.
*
* @param context the initialization context.
* @return a new instance of {@link Elasticsearch8AsyncWriter}.
*/
@Override
- public StatefulSinkWriter<InputT, BufferedRequestState<Operation>>
createWriter(
- InitContext context) {
+ public SinkWriter<InputT> createWriter(InitContext context) {
return new Elasticsearch8AsyncWriter<>(
getElementConverter(),
context,
@@ -99,19 +99,35 @@ public class Elasticsearch8AsyncSink<InputT> extends
AsyncSinkBase<InputT, Opera
Collections.emptyList());
}
+ @Override
+ public SinkWriter<InputT> createWriter(WriterInitContext context) throws
IOException {
+ return new Elasticsearch8AsyncWriter<>(
+ getElementConverter(),
+ new InitContextAdapter(context),
+ getMaxBatchSize(),
+ getMaxInFlightRequests(),
+ getMaxBufferedRequests(),
+ getMaxBatchSizeInBytes(),
+ getMaxTimeInBufferMS(),
+ getMaxRecordSizeInBytes(),
+ networkConfig,
+ Collections.emptyList());
+ }
+
/**
- * Restores a {@link StatefulSinkWriter} from a previously saved state.
+ * Restores a {@link StatefulSinkWriterAdapter} from a previously saved
state.
*
* @param context the initialization context.
* @param recoveredState the recovered state.
* @return a restored instance of {@link Elasticsearch8AsyncWriter}.
*/
@Override
- public StatefulSinkWriter<InputT, BufferedRequestState<Operation>>
restoreWriter(
- InitContext context, Collection<BufferedRequestState<Operation>>
recoveredState) {
+ public StatefulSinkWriterAdapter<InputT, BufferedRequestState<Operation>>
restoreWriterAdapter(
+ WriterInitContext context,
Collection<BufferedRequestState<Operation>> recoveredState)
+ throws IOException {
return new Elasticsearch8AsyncWriter<>(
getElementConverter(),
- context,
+ new InitContextAdapter(context),
getMaxBatchSize(),
getMaxInFlightRequests(),
getMaxBufferedRequests(),
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncWriter.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncWriter.java
index 86d081697..b3246b1aa 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncWriter.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncWriter.java
@@ -20,10 +20,12 @@
package org.apache.flink.cdc.connectors.elasticsearch.v2;
import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriterAdapter;
import org.apache.flink.connector.base.sink.throwable.FatalExceptionClassifier;
-import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter;
+import org.apache.flink.connector.base.sink.writer.AsyncSinkWriterAdapter;
import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
import
org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
@@ -41,9 +43,7 @@ import java.net.ConnectException;
import java.net.NoRouteToHostException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
-import java.util.function.Consumer;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -53,7 +53,8 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
*
* @param <InputT> type of Operations
*/
-public class Elasticsearch8AsyncWriter<InputT> extends AsyncSinkWriter<InputT,
Operation> {
+public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriterAdapter<InputT, Operation>
+ implements StatefulSinkWriterAdapter<InputT,
BufferedRequestState<Operation>> {
private static final Logger LOG =
LoggerFactory.getLogger(Elasticsearch8AsyncWriter.class);
private final ElasticsearchAsyncClient esClient;
@@ -118,7 +119,7 @@ public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriter<InputT, O
@Override
protected void submitRequestEntries(
- List<Operation> requestEntries, Consumer<List<Operation>>
requestResult) {
+ List<Operation> requestEntries, ResultHandler<Operation>
resultHandler) {
numRequestSubmittedCounter.inc();
LOG.debug("submitRequestEntries with {} items", requestEntries.size());
@@ -136,7 +137,7 @@ public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriter<InputT, O
LOG.debug(
"Skipping empty BulkRequest, all {} operation(s) have null
BulkOperationVariant",
requestEntries.size());
- requestResult.accept(Collections.emptyList());
+ resultHandler.complete();
return;
}
@@ -144,19 +145,19 @@ public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriter<InputT, O
.whenComplete(
(response, error) -> {
if (error != null) {
- handleFailedRequest(requestEntries,
requestResult, error);
+ handleFailedRequest(requestEntries,
resultHandler, error);
} else if (response.errors()) {
handlePartiallyFailedRequest(
- requestEntries, requestResult,
response);
+ requestEntries, resultHandler,
response);
} else {
- handleSuccessfulRequest(requestResult,
response);
+ handleSuccessfulRequest(resultHandler,
response);
}
});
}
private void handleFailedRequest(
List<Operation> requestEntries,
- Consumer<List<Operation>> requestResult,
+ ResultHandler<Operation> resultHandler,
Throwable error) {
LOG.warn(
"The BulkRequest of {} operation(s) has failed due to: {}",
@@ -165,14 +166,20 @@ public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriter<InputT, O
LOG.debug("The BulkRequest has failed", error);
numRecordsOutErrorsCounter.inc(requestEntries.size());
- if (isRetryable(error.getCause())) {
- requestResult.accept(requestEntries);
+ Throwable retryableError = error.getCause() != null ? error.getCause()
: error;
+ if (isRetryable(retryableError)) {
+ resultHandler.retryForEntries(requestEntries);
+ } else {
+ resultHandler.completeExceptionally(
+ retryableError instanceof Exception
+ ? (Exception) retryableError
+ : new FlinkRuntimeException(retryableError));
}
}
private void handlePartiallyFailedRequest(
List<Operation> requestEntries,
- Consumer<List<Operation>> requestResult,
+ ResultHandler<Operation> resultHandler,
BulkResponse response) {
LOG.debug("The BulkRequest has failed partially. Response: {}",
response);
ArrayList<Operation> failedItems = new ArrayList<>();
@@ -192,16 +199,16 @@ public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriter<InputT, O
requestEntries.size(),
failedItems.size(),
response.took());
- requestResult.accept(failedItems);
+ resultHandler.retryForEntries(failedItems);
}
private void handleSuccessfulRequest(
- Consumer<List<Operation>> requestResult, BulkResponse response) {
+ ResultHandler<Operation> resultHandler, BulkResponse response) {
LOG.debug(
"The BulkRequest of {} operation(s) completed successfully. It
took {}ms",
response.items().size(),
response.took());
- requestResult.accept(Collections.emptyList());
+ resultHandler.complete();
}
private boolean isRetryable(Throwable error) {
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
index 2619583c3..c97125726 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/pom.xml
@@ -205,6 +205,12 @@ limitations under the License.
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-connector-elasticsearch</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-postgres-cdc</artifactId>
@@ -271,6 +277,12 @@ limitations under the License.
<version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>${testcontainers.version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- benchmark -->
<dependency>
@@ -763,6 +775,15 @@ limitations under the License.
<outputDirectory>${project.build.directory}/dependencies
</outputDirectory>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-cdc-pipeline-connector-elasticsearch</artifactId>
+ <version>${project.version}</version>
+
<destFileName>elasticsearch-cdc-pipeline-connector.jar</destFileName>
+ <type>jar</type>
+
<outputDirectory>${project.build.directory}/dependencies
+ </outputDirectory>
+ </artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-parquet</artifactId>
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch6E2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch6E2eITCase.java
new file mode 100644
index 000000000..77a05c524
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch6E2eITCase.java
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.pipeline.tests.stage2;
+
+import org.apache.flink.cdc.common.test.utils.TestUtils;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+import org.apache.flink.elasticsearch6.shaded.org.apache.http.HttpHost;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.action.get.GetRequest;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.action.get.GetResponse;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.client.RequestOptions;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.client.RestClient;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.client.RestHighLevelClient;
+import
org.apache.flink.elasticsearch6.shaded.org.elasticsearch.common.xcontent.XContentType;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.stream.Stream;
+
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
+class MysqlToElasticsearch6E2eITCase extends PipelineTestEnvironment {
+ private static final Logger LOG =
LoggerFactory.getLogger(MysqlToElasticsearch6E2eITCase.class);
+
+ protected final UniqueDatabase inventoryDatabase =
+ new UniqueDatabase(MYSQL, "mysql_inventory", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+
+ private static final String ELASTICSEARCH_VERSION = "6.8.20";
+ private static final String INTER_CONTAINER_ES_ALIAS = "elasticsearch";
+
+ private RestHighLevelClient client;
+
+ @Container
+ private static final ElasticsearchContainer ELASTICSEARCH_CONTAINER =
+ createElasticsearchContainer();
+
+ MysqlToElasticsearch6E2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
+ @BeforeAll
+ static void initializeContainers() {
+ LOG.info("Starting containers...");
+ Startables.deepStart(Stream.of(MYSQL, ELASTICSEARCH_CONTAINER)).join();
+ LOG.info("Containers are started.");
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ inventoryDatabase.createAndInitialize();
+ client = createElasticsearchClient();
+ initEsData();
+ }
+
+ @AfterEach
+ public void after() {
+ super.after();
+ inventoryDatabase.dropDatabase();
+ if (client != null) {
+ try {
+ client.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Test
+ void testSyncWholeDatabase() throws Exception {
+ String databaseName = inventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: elasticsearch\n"
+ + " hosts: %s:9200\n"
+ + " version: 6\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ databaseName,
+ INTER_CONTAINER_ES_ALIAS,
+ parallelism);
+ Path esConnectorJar =
TestUtils.getResource("elasticsearch-cdc-pipeline-connector.jar");
+ submitPipelineJob(pipelineJob, esConnectorJar);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ LOG.info("Pipeline job is running");
+
+ verifySnapshotData(databaseName);
+ verifyIncrementalData(databaseName);
+ }
+
+ private void verifySnapshotData(String databaseName) throws Exception {
+ String productsIndex = databaseName + ".products";
+ String customersIndex = databaseName + ".customers";
+
+ // products id=101 (all fields populated)
+ waitForEsDocument(productsIndex, "101");
+ GetResponse resp =
+ client.get(new GetRequest(productsIndex).id("101"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "scooter")
+ .containsEntry("description", "Small 2-wheel scooter")
+ .containsEntry("weight", 3.14)
+ .containsEntry("enum_c", "red")
+ .containsEntry("json_c", "{\"key1\": \"value1\"}")
+ .containsEntry("point_c",
"{\"coordinates\":[1,1],\"type\":\"Point\",\"srid\":0}");
+
+ // products id=106 (enum_c, json_c, point_c are null)
+ waitForEsDocument(productsIndex, "106");
+ resp = client.get(new GetRequest(productsIndex).id("106"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "hammer")
+ .containsEntry("description", "16oz carpenter's hammer")
+ .containsEntry("weight", 1.0);
+
+ // products id=109 (last snapshot row)
+ waitForEsDocument(productsIndex, "109");
+ resp = client.get(new GetRequest(productsIndex).id("109"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "spare tire")
+ .containsEntry("description", "24 inch spare tire")
+ .containsEntry("weight", 22.2);
+
+ // customers id=101
+ waitForEsDocument(customersIndex, "101");
+ resp = client.get(new GetRequest(customersIndex).id("101"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "user_1")
+ .containsEntry("address", "Shanghai")
+ .containsEntry("phone_number", "123567891234");
+ }
+
+ private void verifyIncrementalData(String databaseName) throws Exception {
+ LOG.info("Begin incremental reading stage.");
+ String productsIndex = databaseName + ".products";
+ String mysqlJdbcUrl =
+ String.format(
+ "jdbc:mysql://%s:%s/%s",
+ MYSQL.getHost(), MYSQL.getDatabasePort(),
databaseName);
+ try (Connection conn =
+ DriverManager.getConnection(
+ mysqlJdbcUrl, MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ Statement stat = conn.createStatement()) {
+
+ // INSERT
+ stat.execute(
+ "INSERT INTO products VALUES "
+ + "(default,'jacket','water resistent white wind
breaker',0.2, null, null, null);");
+ waitForEsDocument(productsIndex, "110");
+ GetResponse resp =
+ client.get(new GetRequest(productsIndex).id("110"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "jacket")
+ .containsEntry("description", "water resistent white wind
breaker")
+ .containsEntry("weight", 0.2);
+
+ // UPDATE
+ stat.execute("UPDATE products SET description='18oz carpenter
hammer' WHERE id=106;");
+ stat.execute("UPDATE products SET weight='5.1' WHERE id=107;");
+ waitForEsDocumentField(productsIndex, "106", "description", "18oz
carpenter hammer");
+ waitForEsDocumentField(productsIndex, "107", "weight", 5.1);
+
+ // DELETE
+ stat.execute("DELETE FROM products WHERE id=101;");
+ waitForEsDocumentDeleted(productsIndex, "101");
+ }
+ }
+
+ private static ElasticsearchContainer createElasticsearchContainer() {
+ DockerImageName imageName =
+ DockerImageName.parse(
+
"docker.elastic.co/elasticsearch/elasticsearch:"
+ + ELASTICSEARCH_VERSION)
+
.asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch");
+ ElasticsearchContainer esContainer = new
ElasticsearchContainer(imageName);
+ esContainer
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_ES_ALIAS)
+ .withEnv("xpack.security.enabled", "false")
+ .withEnv("ES_JAVA_OPTS", "-Xms2g -Xmx2g")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return esContainer;
+ }
+
+ private RestHighLevelClient createElasticsearchClient() {
+ return new RestHighLevelClient(
+ RestClient.builder(
+ new HttpHost(
+ ELASTICSEARCH_CONTAINER.getHost(),
+ ELASTICSEARCH_CONTAINER.getFirstMappedPort(),
+ "http")));
+ }
+
+ private void initEsData() throws IOException {
+ String dbName = inventoryDatabase.getDatabaseName();
+ createProductsIndex(dbName);
+ createCustomersIndex(dbName);
+ }
+
+ private void createProductsIndex(String dbName) throws IOException {
+ String indexName = dbName + ".products";
+ String source =
+ "{\"mappings\":{\"_doc\":{\"properties\":{"
+ + "\"id\":{\"type\":\"integer\"},"
+ + "\"name\":{\"type\":\"keyword\"},"
+ + "\"description\":{\"type\":\"text\"},"
+ + "\"weight\":{\"type\":\"float\"},"
+ + "\"enum_c\":{\"type\":\"keyword\"},"
+ + "\"json_c\":{\"type\":\"keyword\"},"
+ + "\"point_c\":{\"type\":\"keyword\"}"
+ + "}}}}";
+ client.indices()
+ .create(
+ new CreateIndexRequest(indexName).source(source,
XContentType.JSON),
+ RequestOptions.DEFAULT);
+ }
+
+ private void createCustomersIndex(String dbName) throws IOException {
+ String indexName = dbName + ".customers";
+ String source =
+ "{\"mappings\":{\"_doc\":{\"properties\":{"
+ + "\"id\":{\"type\":\"integer\"},"
+ + "\"name\":{\"type\":\"keyword\"},"
+ + "\"address\":{\"type\":\"text\"},"
+ + "\"phone_number\":{\"type\":\"keyword\"}"
+ + "}}}}";
+ client.indices()
+ .create(
+ new CreateIndexRequest(indexName).source(source,
XContentType.JSON),
+ RequestOptions.DEFAULT);
+ }
+
+ private void waitForEsDocument(String indexName, String docId) throws
Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetRequest getRequest = new GetRequest(indexName).id(docId);
+ GetResponse response = client.get(getRequest,
RequestOptions.DEFAULT);
+ if (response.isExists()) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail("Timed out waiting for ES document: " + indexName +
"/" + docId);
+ }
+
+ private void waitForEsDocumentField(
+ String indexName, String docId, String field, Object
expectedValue) throws Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetRequest getRequest = new GetRequest(indexName).id(docId);
+ GetResponse response = client.get(getRequest,
RequestOptions.DEFAULT);
+ if (response.isExists() &&
expectedValue.equals(response.getSource().get(field))) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail(
+ "Timed out waiting for ES document field: "
+ + indexName
+ + "/"
+ + docId
+ + " "
+ + field
+ + "="
+ + expectedValue);
+ }
+
+ private void waitForEsDocumentDeleted(String indexName, String docId)
throws Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetRequest getRequest = new GetRequest(indexName).id(docId);
+ GetResponse response = client.get(getRequest,
RequestOptions.DEFAULT);
+ if (!response.isExists()) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail("Document was not deleted within timeout: " +
indexName + "/" + docId);
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch7E2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch7E2eITCase.java
new file mode 100644
index 000000000..78b46894c
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch7E2eITCase.java
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.pipeline.tests.stage2;
+
+import org.apache.flink.cdc.common.test.utils.TestUtils;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+import org.apache.flink.elasticsearch7.shaded.org.apache.http.HttpHost;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.action.get.GetRequest;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.action.get.GetResponse;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.client.RequestOptions;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.client.RestClient;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.client.RestHighLevelClient;
+import
org.apache.flink.elasticsearch7.shaded.org.elasticsearch.common.xcontent.XContentType;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.stream.Stream;
+
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
+class MysqlToElasticsearch7E2eITCase extends PipelineTestEnvironment {
+ private static final Logger LOG =
LoggerFactory.getLogger(MysqlToElasticsearch7E2eITCase.class);
+
+ protected final UniqueDatabase inventoryDatabase =
+ new UniqueDatabase(MYSQL, "mysql_inventory", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+
+ private static final String ELASTICSEARCH_VERSION = "7.10.2";
+ private static final String INTER_CONTAINER_ES_ALIAS = "elasticsearch7";
+
+ private RestHighLevelClient client;
+
+ @Container
+ private static final ElasticsearchContainer ELASTICSEARCH_CONTAINER =
+ createElasticsearchContainer();
+
+ MysqlToElasticsearch7E2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
+ @BeforeAll
+ static void initializeContainers() {
+ LOG.info("Starting containers...");
+ Startables.deepStart(Stream.of(MYSQL, ELASTICSEARCH_CONTAINER)).join();
+ LOG.info("Containers are started.");
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ inventoryDatabase.createAndInitialize();
+ client = createElasticsearchClient();
+ initEsData();
+ }
+
+ @AfterEach
+ public void after() {
+ super.after();
+ inventoryDatabase.dropDatabase();
+ if (client != null) {
+ try {
+ client.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Test
+ void testSyncWholeDatabase() throws Exception {
+ String databaseName = inventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: elasticsearch\n"
+ + " hosts: %s:9200\n"
+ + " version: 7\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ databaseName,
+ INTER_CONTAINER_ES_ALIAS,
+ parallelism);
+ Path esConnectorJar =
TestUtils.getResource("elasticsearch-cdc-pipeline-connector.jar");
+ submitPipelineJob(pipelineJob, esConnectorJar);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ LOG.info("Pipeline job is running");
+
+ verifySnapshotData(databaseName);
+ verifyIncrementalData(databaseName);
+ }
+
+ private void verifySnapshotData(String databaseName) throws Exception {
+ String productsIndex = databaseName + ".products";
+ String customersIndex = databaseName + ".customers";
+
+ // products id=101 (all fields populated)
+ waitForEsDocument(productsIndex, "101");
+ GetResponse resp =
+ client.get(new GetRequest(productsIndex).id("101"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "scooter")
+ .containsEntry("description", "Small 2-wheel scooter")
+ .containsEntry("weight", 3.14)
+ .containsEntry("enum_c", "red")
+ .containsEntry("json_c", "{\"key1\": \"value1\"}")
+ .containsEntry("point_c",
"{\"coordinates\":[1,1],\"type\":\"Point\",\"srid\":0}");
+
+ // products id=106 (enum_c, json_c, point_c are null)
+ waitForEsDocument(productsIndex, "106");
+ resp = client.get(new GetRequest(productsIndex).id("106"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "hammer")
+ .containsEntry("description", "16oz carpenter's hammer")
+ .containsEntry("weight", 1.0);
+
+ // products id=109 (last snapshot row)
+ waitForEsDocument(productsIndex, "109");
+ resp = client.get(new GetRequest(productsIndex).id("109"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "spare tire")
+ .containsEntry("description", "24 inch spare tire")
+ .containsEntry("weight", 22.2);
+
+ // customers id=101
+ waitForEsDocument(customersIndex, "101");
+ resp = client.get(new GetRequest(customersIndex).id("101"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "user_1")
+ .containsEntry("address", "Shanghai")
+ .containsEntry("phone_number", "123567891234");
+ }
+
+ private void verifyIncrementalData(String databaseName) throws Exception {
+ LOG.info("Begin incremental reading stage.");
+ String productsIndex = databaseName + ".products";
+ String mysqlJdbcUrl =
+ String.format(
+ "jdbc:mysql://%s:%s/%s",
+ MYSQL.getHost(), MYSQL.getDatabasePort(),
databaseName);
+ try (Connection conn =
+ DriverManager.getConnection(
+ mysqlJdbcUrl, MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ Statement stat = conn.createStatement()) {
+
+ // INSERT
+ stat.execute(
+ "INSERT INTO products VALUES "
+ + "(default,'jacket','water resistent white wind
breaker',0.2, null, null, null);");
+ waitForEsDocument(productsIndex, "110");
+ GetResponse resp =
+ client.get(new GetRequest(productsIndex).id("110"),
RequestOptions.DEFAULT);
+ Assertions.assertThat(resp.getSource())
+ .containsEntry("name", "jacket")
+ .containsEntry("description", "water resistent white wind
breaker")
+ .containsEntry("weight", 0.2);
+
+ // UPDATE
+ stat.execute("UPDATE products SET description='18oz carpenter
hammer' WHERE id=106;");
+ stat.execute("UPDATE products SET weight='5.1' WHERE id=107;");
+ waitForEsDocumentField(productsIndex, "106", "description", "18oz
carpenter hammer");
+ waitForEsDocumentField(productsIndex, "107", "weight", 5.1);
+
+ // DELETE
+ stat.execute("DELETE FROM products WHERE id=101;");
+ waitForEsDocumentDeleted(productsIndex, "101");
+ }
+ }
+
+ private static ElasticsearchContainer createElasticsearchContainer() {
+ DockerImageName imageName =
+ DockerImageName.parse(
+
"docker.elastic.co/elasticsearch/elasticsearch:"
+ + ELASTICSEARCH_VERSION)
+
.asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch");
+ ElasticsearchContainer esContainer = new
ElasticsearchContainer(imageName);
+ esContainer
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_ES_ALIAS)
+ .withEnv("xpack.security.enabled", "false")
+ .withEnv("ES_JAVA_OPTS", "-Xms2g -Xmx2g")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return esContainer;
+ }
+
+ private RestHighLevelClient createElasticsearchClient() {
+ return new RestHighLevelClient(
+ RestClient.builder(
+ new HttpHost(
+ ELASTICSEARCH_CONTAINER.getHost(),
+ ELASTICSEARCH_CONTAINER.getFirstMappedPort(),
+ "http")));
+ }
+
+ private void initEsData() throws IOException {
+ String dbName = inventoryDatabase.getDatabaseName();
+ createProductsIndex(dbName);
+ createCustomersIndex(dbName);
+ }
+
+ private void createProductsIndex(String dbName) throws IOException {
+ String indexName = dbName + ".products";
+ String source =
+ "{\"mappings\":{\"_doc\":{\"properties\":{"
+ + "\"id\":{\"type\":\"integer\"},"
+ + "\"name\":{\"type\":\"keyword\"},"
+ + "\"description\":{\"type\":\"text\"},"
+ + "\"weight\":{\"type\":\"float\"},"
+ + "\"enum_c\":{\"type\":\"keyword\"},"
+ + "\"json_c\":{\"type\":\"keyword\"},"
+ + "\"point_c\":{\"type\":\"keyword\"}"
+ + "}}}}";
+ client.indices()
+ .create(
+ new CreateIndexRequest(indexName).source(source,
XContentType.JSON),
+ RequestOptions.DEFAULT);
+ }
+
+ private void createCustomersIndex(String dbName) throws IOException {
+ String indexName = dbName + ".customers";
+ String source =
+ "{\"mappings\":{\"_doc\":{\"properties\":{"
+ + "\"id\":{\"type\":\"integer\"},"
+ + "\"name\":{\"type\":\"keyword\"},"
+ + "\"address\":{\"type\":\"text\"},"
+ + "\"phone_number\":{\"type\":\"keyword\"}"
+ + "}}}}";
+ client.indices()
+ .create(
+ new CreateIndexRequest(indexName).source(source,
XContentType.JSON),
+ RequestOptions.DEFAULT);
+ }
+
+ private void waitForEsDocument(String indexName, String docId) throws
Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetRequest getRequest = new GetRequest(indexName).id(docId);
+ GetResponse response = client.get(getRequest,
RequestOptions.DEFAULT);
+ if (response.isExists()) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail("Timed out waiting for ES document: " + indexName +
"/" + docId);
+ }
+
+ private void waitForEsDocumentField(
+ String indexName, String docId, String field, Object
expectedValue) throws Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetRequest getRequest = new GetRequest(indexName).id(docId);
+ GetResponse response = client.get(getRequest,
RequestOptions.DEFAULT);
+ if (response.isExists() &&
expectedValue.equals(response.getSource().get(field))) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail(
+ "Timed out waiting for ES document field: "
+ + indexName
+ + "/"
+ + docId
+ + " "
+ + field
+ + "="
+ + expectedValue);
+ }
+
+ private void waitForEsDocumentDeleted(String indexName, String docId)
throws Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetRequest getRequest = new GetRequest(indexName).id(docId);
+ GetResponse response = client.get(getRequest,
RequestOptions.DEFAULT);
+ if (!response.isExists()) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail("Document was not deleted within timeout: " +
indexName + "/" + docId);
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch8E2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch8E2eITCase.java
new file mode 100644
index 000000000..fb96c6d7f
--- /dev/null
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/MysqlToElasticsearch8E2eITCase.java
@@ -0,0 +1,380 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.flink.cdc.pipeline.tests.stage2;
+
+import org.apache.flink.cdc.common.test.utils.TestUtils;
+import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.elasticsearch.core.GetRequest;
+import co.elastic.clients.elasticsearch.core.GetResponse;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.assertj.core.api.Assertions;
+import org.elasticsearch.client.RestClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.Map;
+import java.util.stream.Stream;
+
+@ParameterizedClass
+@ValueSource(ints = {1, 4})
+class MysqlToElasticsearch8E2eITCase extends PipelineTestEnvironment {
+ private static final Logger LOG =
LoggerFactory.getLogger(MysqlToElasticsearch8E2eITCase.class);
+
+ protected final UniqueDatabase inventoryDatabase =
+ new UniqueDatabase(MYSQL, "mysql_inventory", MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+
+ private static final String ELASTICSEARCH_VERSION = "8.12.1";
+ private static final String INTER_CONTAINER_ES_ALIAS = "elasticsearch8";
+ private static final String DEFAULT_USERNAME = "elastic";
+ private static final String DEFAULT_PASSWORD = "123456";
+
+ private ElasticsearchClient client;
+
+ @Container
+ private static final ElasticsearchContainer ELASTICSEARCH_CONTAINER =
+ createElasticsearchContainer();
+
+ MysqlToElasticsearch8E2eITCase(int parallelism) {
+ super(parallelism);
+ }
+
+ @BeforeAll
+ static void initializeContainers() {
+ LOG.info("Starting containers...");
+ Startables.deepStart(Stream.of(MYSQL, ELASTICSEARCH_CONTAINER)).join();
+ LOG.info("Containers are started.");
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ inventoryDatabase.createAndInitialize();
+ client = createElasticsearchClient();
+ initEsData();
+ }
+
+ @AfterEach
+ public void after() {
+ super.after();
+ inventoryDatabase.dropDatabase();
+ if (client != null) {
+ try {
+ client.shutdown();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ @Test
+ void testSyncWholeDatabase() throws Exception {
+ String databaseName = inventoryDatabase.getDatabaseName();
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + "\n"
+ + "sink:\n"
+ + " type: elasticsearch\n"
+ + " hosts: %s:9200\n"
+ + " version: 8\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ databaseName,
+ INTER_CONTAINER_ES_ALIAS,
+ DEFAULT_USERNAME,
+ DEFAULT_PASSWORD,
+ parallelism);
+ Path esConnectorJar =
TestUtils.getResource("elasticsearch-cdc-pipeline-connector.jar");
+ submitPipelineJob(pipelineJob, esConnectorJar);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ LOG.info("Pipeline job is running");
+
+ verifySnapshotData(databaseName);
+ verifyIncrementalData(databaseName);
+ }
+
+ private void verifySnapshotData(String databaseName) throws Exception {
+ String productsIndex = databaseName + ".products";
+ String customersIndex = databaseName + ".customers";
+
+ // products id=101 (all fields populated)
+ waitForEsDocument(productsIndex, "101");
+ GetResponse<Map> resp =
+ client.get(
+ new
GetRequest.Builder().index(productsIndex).id("101").build(), Map.class);
+ Assertions.assertThat(resp.source())
+ .containsEntry("name", "scooter")
+ .containsEntry("description", "Small 2-wheel scooter")
+ .containsEntry("weight", 3.14)
+ .containsEntry("enum_c", "red")
+ .containsEntry("json_c", "{\"key1\": \"value1\"}")
+ .containsEntry("point_c",
"{\"coordinates\":[1,1],\"type\":\"Point\",\"srid\":0}");
+
+ // products id=106 (enum_c, json_c, point_c are null)
+ waitForEsDocument(productsIndex, "106");
+ resp =
+ client.get(
+ new
GetRequest.Builder().index(productsIndex).id("106").build(), Map.class);
+ Assertions.assertThat(resp.source())
+ .containsEntry("name", "hammer")
+ .containsEntry("description", "16oz carpenter's hammer")
+ .containsEntry("weight", 1.0);
+
+ // products id=109 (last snapshot row)
+ waitForEsDocument(productsIndex, "109");
+ resp =
+ client.get(
+ new
GetRequest.Builder().index(productsIndex).id("109").build(), Map.class);
+ Assertions.assertThat(resp.source())
+ .containsEntry("name", "spare tire")
+ .containsEntry("description", "24 inch spare tire")
+ .containsEntry("weight", 22.2);
+
+ // customers id=101
+ waitForEsDocument(customersIndex, "101");
+ resp =
+ client.get(
+ new
GetRequest.Builder().index(customersIndex).id("101").build(),
+ Map.class);
+ Assertions.assertThat(resp.source())
+ .containsEntry("name", "user_1")
+ .containsEntry("address", "Shanghai")
+ .containsEntry("phone_number", "123567891234");
+ }
+
+ private void verifyIncrementalData(String databaseName) throws Exception {
+ LOG.info("Begin incremental reading stage.");
+ String productsIndex = databaseName + ".products";
+ String mysqlJdbcUrl =
+ String.format(
+ "jdbc:mysql://%s:%s/%s",
+ MYSQL.getHost(), MYSQL.getDatabasePort(),
databaseName);
+ try (Connection conn =
+ DriverManager.getConnection(
+ mysqlJdbcUrl, MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD);
+ Statement stat = conn.createStatement()) {
+
+ // INSERT
+ stat.execute(
+ "INSERT INTO products VALUES "
+ + "(default,'jacket','water resistent white wind
breaker',0.2, null, null, null);");
+ waitForEsDocument(productsIndex, "110");
+ GetResponse<Map> resp =
+ client.get(
+ new
GetRequest.Builder().index(productsIndex).id("110").build(),
+ Map.class);
+ Assertions.assertThat(resp.source())
+ .containsEntry("name", "jacket")
+ .containsEntry("description", "water resistent white wind
breaker")
+ .containsEntry("weight", 0.2);
+
+ // UPDATE
+ stat.execute("UPDATE products SET description='18oz carpenter
hammer' WHERE id=106;");
+ stat.execute("UPDATE products SET weight='5.1' WHERE id=107;");
+ waitForEsDocumentField(productsIndex, "106", "description", "18oz
carpenter hammer");
+ waitForEsDocumentField(productsIndex, "107", "weight", 5.1);
+
+ // DELETE
+ stat.execute("DELETE FROM products WHERE id=101;");
+ waitForEsDocumentDeleted(productsIndex, "101");
+ }
+ }
+
+ private static ElasticsearchContainer createElasticsearchContainer() {
+ DockerImageName imageName =
+ DockerImageName.parse(
+ "docker.elastic.co/elasticsearch/elasticsearch:" +
ELASTICSEARCH_VERSION);
+ ElasticsearchContainer esContainer = new
ElasticsearchContainer(imageName);
+ esContainer
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_ES_ALIAS)
+ .withEnv("discovery.type", "single-node")
+ .withEnv("xpack.security.enabled", "true")
+ .withEnv("xpack.security.http.ssl.enabled", "false")
+ .withEnv("ES_JAVA_OPTS", "-Xms2g -Xmx2g")
+ .withPassword(DEFAULT_PASSWORD)
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ return esContainer;
+ }
+
+ private ElasticsearchClient createElasticsearchClient() {
+ BasicCredentialsProvider credentialsProvider = new
BasicCredentialsProvider();
+ credentialsProvider.setCredentials(
+ AuthScope.ANY, new
UsernamePasswordCredentials(DEFAULT_USERNAME, DEFAULT_PASSWORD));
+ RestClientTransport transport =
+ new RestClientTransport(
+ RestClient.builder(
+ new HttpHost(
+
ELASTICSEARCH_CONTAINER.getHost(),
+
ELASTICSEARCH_CONTAINER.getFirstMappedPort(),
+ "http"))
+ .setHttpClientConfigCallback(
+ httpAsyncClientBuilder ->
+ httpAsyncClientBuilder
+
.setDefaultCredentialsProvider(
+
credentialsProvider))
+ .build(),
+ new JacksonJsonpMapper());
+ return new ElasticsearchClient(transport);
+ }
+
+ private void initEsData() throws IOException {
+ String dbName = inventoryDatabase.getDatabaseName();
+ createProductsIndex(dbName);
+ createCustomersIndex(dbName);
+ }
+
+ private void createProductsIndex(String dbName) throws IOException {
+ String indexName = dbName + ".products";
+ client.indices()
+ .create(
+ c ->
+ c.index(indexName)
+ .mappings(
+ m ->
+ m.properties("id", p
-> p.integer(i -> i))
+ .properties(
+ "name",
+ p ->
p.keyword(k -> k))
+ .properties(
+
"description",
+ p ->
p.text(t -> t))
+ .properties(
+
"weight",
+ p ->
p.float_(f -> f))
+ .properties(
+
"enum_c",
+ p ->
p.keyword(k -> k))
+ .properties(
+
"json_c",
+ p ->
p.keyword(k -> k))
+ .properties(
+
"point_c",
+ p ->
p.keyword(k -> k))));
+ }
+
+ private void createCustomersIndex(String dbName) throws IOException {
+ String indexName = dbName + ".customers";
+ client.indices()
+ .create(
+ c ->
+ c.index(indexName)
+ .mappings(
+ m ->
+ m.properties("id", p
-> p.integer(i -> i))
+ .properties(
+ "name",
+ p ->
p.keyword(k -> k))
+ .properties(
+
"address",
+ p ->
p.text(t -> t))
+ .properties(
+
"phone_number",
+ p ->
p.keyword(k -> k))));
+ }
+
+ private void waitForEsDocument(String indexName, String docId) throws
Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetResponse<Map> response =
+ client.get(
+ new
GetRequest.Builder().index(indexName).id(docId).build(), Map.class);
+ if (response.source() != null) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail("Timed out waiting for ES document: " + indexName +
"/" + docId);
+ }
+
+ private void waitForEsDocumentField(
+ String indexName, String docId, String field, Object
expectedValue) throws Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetResponse<Map> response =
+ client.get(
+ new
GetRequest.Builder().index(indexName).id(docId).build(), Map.class);
+ if (response.source() != null &&
expectedValue.equals(response.source().get(field))) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail(
+ "Timed out waiting for ES document field: "
+ + indexName
+ + "/"
+ + docId
+ + " "
+ + field
+ + "="
+ + expectedValue);
+ }
+
+ private void waitForEsDocumentDeleted(String indexName, String docId)
throws Exception {
+ long deadline = System.currentTimeMillis() +
EVENT_WAITING_TIMEOUT.toMillis();
+ while (System.currentTimeMillis() < deadline) {
+ GetResponse<Map> response =
+ client.get(
+ new
GetRequest.Builder().index(indexName).id(docId).build(), Map.class);
+ if (response.source() == null) {
+ return;
+ }
+ Thread.sleep(1000L);
+ }
+ Assertions.fail("Document was not deleted within timeout: " +
indexName + "/" + docId);
+ }
+}
diff --git
a/flink-cdc-flink1-compat/src/main/java/org/apache/flink/api/connector/sink2/StatefulSinkWriterAdapter.java
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/api/connector/sink2/StatefulSinkWriterAdapter.java
new file mode 100644
index 000000000..70e8034e3
--- /dev/null
+++
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/api/connector/sink2/StatefulSinkWriterAdapter.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.connector.sink2;
+
+/**
+ * Compatibility adapter for Flink 1.20. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public interface StatefulSinkWriterAdapter<InputT, WriterStateT>
+ extends StatefulSink.StatefulSinkWriter<InputT, WriterStateT> {}
diff --git
a/flink-cdc-flink1-compat/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContextAdapter.java
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContextAdapter.java
new file mode 100644
index 000000000..04d8cde07
--- /dev/null
+++
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContextAdapter.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.connector.sink2;
+
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.TaskInfo;
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.operators.ProcessingTimeService;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import java.util.OptionalLong;
+
+/**
+ * Compatibility adapter for Flink 1.20. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public class WriterInitContextAdapter implements WriterInitContext {
+
+ private final Sink.InitContext context;
+
+ public WriterInitContextAdapter(Sink.InitContext context) {
+ this.context = context;
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return this.context.getUserCodeClassLoader();
+ }
+
+ @Override
+ public MailboxExecutor getMailboxExecutor() {
+ return this.context.getMailboxExecutor();
+ }
+
+ @Override
+ public ProcessingTimeService getProcessingTimeService() {
+ return this.context.getProcessingTimeService();
+ }
+
+ @Override
+ public SinkWriterMetricGroup metricGroup() {
+ return this.context.metricGroup();
+ }
+
+ @Override
+ public SerializationSchema.InitializationContext
asSerializationSchemaInitializationContext() {
+ return this.context.asSerializationSchemaInitializationContext();
+ }
+
+ @Override
+ public boolean isObjectReuseEnabled() {
+ return this.context.isObjectReuseEnabled();
+ }
+
+ @Override
+ public <IN> TypeSerializer<IN> createInputSerializer() {
+ return this.context.createInputSerializer();
+ }
+
+ @Override
+ public OptionalLong getRestoredCheckpointId() {
+ return this.context.getRestoredCheckpointId();
+ }
+
+ @Override
+ public JobInfo getJobInfo() {
+ return this.context.getJobInfo();
+ }
+
+ @Override
+ public TaskInfo getTaskInfo() {
+ return this.context.getTaskInfo();
+ }
+}
diff --git
a/flink-cdc-flink1-compat/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseAdapter.java
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseAdapter.java
new file mode 100644
index 000000000..881a2ff39
--- /dev/null
+++
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseAdapter.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.connector.base.sink;
+
+import org.apache.flink.api.connector.sink2.StatefulSinkWriterAdapter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+
+/**
+ * Compatibility adapter for Flink 1.20. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public abstract class AsyncSinkBaseAdapter<InputT, RequestEntryT extends
Serializable>
+ extends AsyncSinkBase<InputT, RequestEntryT> {
+ protected AsyncSinkBaseAdapter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long maxBatchSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes) {
+ super(
+ elementConverter,
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInBytes);
+ }
+
+ @Override
+ public StatefulSinkWriter<InputT, BufferedRequestState<RequestEntryT>>
restoreWriter(
+ WriterInitContext context,
+ Collection<BufferedRequestState<RequestEntryT>> recoveredState)
+ throws IOException {
+ return restoreWriterAdapter(context, recoveredState);
+ }
+
+ public abstract StatefulSinkWriterAdapter<InputT,
BufferedRequestState<RequestEntryT>>
+ restoreWriterAdapter(
+ WriterInitContext context,
+ Collection<BufferedRequestState<RequestEntryT>>
recoveredState)
+ throws IOException;
+}
diff --git
a/flink-cdc-flink1-compat/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterAdapter.java
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterAdapter.java
new file mode 100644
index 000000000..6d5b2d7e1
--- /dev/null
+++
b/flink-cdc-flink1-compat/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterAdapter.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.connector.base.sink.writer;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import
org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
+
+import java.io.Serializable;
+import java.util.Collection;
+
+/**
+ * Compatibility adapter for Flink 1.20. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public abstract class AsyncSinkWriterAdapter<InputT, RequestEntryT extends
Serializable>
+ extends AsyncSinkWriter<InputT, RequestEntryT> {
+
+ public AsyncSinkWriterAdapter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ Sink.InitContext context,
+ AsyncSinkWriterConfiguration configuration,
+ Collection<BufferedRequestState<RequestEntryT>>
bufferedRequestStates) {
+ super(elementConverter, context, configuration, bufferedRequestStates);
+ }
+}
diff --git
a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/StatefulSinkWriterAdapter.java
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/StatefulSinkWriterAdapter.java
new file mode 100644
index 000000000..8d7fbcb33
--- /dev/null
+++
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/StatefulSinkWriterAdapter.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.connector.sink2;
+
+/**
+ * Compatibility adapter for Flink 2.2. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public interface StatefulSinkWriterAdapter<InputT, WriterStateT>
+ extends StatefulSinkWriter<InputT, WriterStateT> {}
diff --git
a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContextAdapter.java
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContextAdapter.java
new file mode 100644
index 000000000..1aa5588e1
--- /dev/null
+++
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContextAdapter.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.api.connector.sink2;
+
+import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.TaskInfo;
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.operators.ProcessingTimeService;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+import org.apache.flink.util.UserCodeClassLoader;
+
+import java.util.OptionalLong;
+
+/**
+ * Compatibility adapter for Flink 2.2. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public class WriterInitContextAdapter implements WriterInitContext {
+
+ private final Sink.InitContext context;
+
+ public WriterInitContextAdapter(Sink.InitContext context) {
+ this.context = context;
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return this.context.getUserCodeClassLoader();
+ }
+
+ @Override
+ public MailboxExecutor getMailboxExecutor() {
+ return this.context.getMailboxExecutor();
+ }
+
+ @Override
+ public ProcessingTimeService getProcessingTimeService() {
+ return this.context.getProcessingTimeService();
+ }
+
+ @Override
+ public SinkWriterMetricGroup metricGroup() {
+ return this.context.metricGroup();
+ }
+
+ @Override
+ public SerializationSchema.InitializationContext
asSerializationSchemaInitializationContext() {
+ return this.context.asSerializationSchemaInitializationContext();
+ }
+
+ @Override
+ public boolean isObjectReuseEnabled() {
+ return this.context.isObjectReuseEnabled();
+ }
+
+ @Override
+ public <IN> TypeSerializer<IN> createInputSerializer() {
+ return this.context.createInputSerializer();
+ }
+
+ @Override
+ public OptionalLong getRestoredCheckpointId() {
+ return this.context.getRestoredCheckpointId();
+ }
+
+ @Override
+ public JobInfo getJobInfo() {
+ return this.context.getJobInfo();
+ }
+
+ @Override
+ public TaskInfo getTaskInfo() {
+ return this.context.getTaskInfo();
+ }
+}
diff --git
a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseAdapter.java
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseAdapter.java
new file mode 100644
index 000000000..c4841ef0c
--- /dev/null
+++
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseAdapter.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.connector.base.sink;
+
+import org.apache.flink.api.connector.sink2.StatefulSinkWriter;
+import org.apache.flink.api.connector.sink2.StatefulSinkWriterAdapter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+
+/**
+ * Compatibility adapter for Flink 2.2. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public abstract class AsyncSinkBaseAdapter<InputT, RequestEntryT extends
Serializable>
+ extends AsyncSinkBase<InputT, RequestEntryT> {
+ protected AsyncSinkBaseAdapter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ int maxBatchSize,
+ int maxInFlightRequests,
+ int maxBufferedRequests,
+ long maxBatchSizeInBytes,
+ long maxTimeInBufferMS,
+ long maxRecordSizeInBytes) {
+ super(
+ elementConverter,
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInBytes);
+ }
+
+ @Override
+ public StatefulSinkWriter<InputT, BufferedRequestState<RequestEntryT>>
restoreWriter(
+ WriterInitContext context,
+ Collection<BufferedRequestState<RequestEntryT>> recoveredState)
+ throws IOException {
+ return restoreWriterAdapter(context, recoveredState);
+ }
+
+ public abstract StatefulSinkWriterAdapter<InputT,
BufferedRequestState<RequestEntryT>>
+ restoreWriterAdapter(
+ WriterInitContext context,
+ Collection<BufferedRequestState<RequestEntryT>>
recoveredState)
+ throws IOException;
+}
diff --git
a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterAdapter.java
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterAdapter.java
new file mode 100644
index 000000000..87f02eb1a
--- /dev/null
+++
b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterAdapter.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flink.connector.base.sink.writer;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.api.connector.sink2.WriterInitContextAdapter;
+import
org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration;
+
+import java.io.Serializable;
+import java.util.Collection;
+
+/**
+ * Compatibility adapter for Flink 2.2. This class is part of the
multi-version compatibility layer
+ * that allows Flink CDC to work across different Flink versions.
+ */
+public abstract class AsyncSinkWriterAdapter<InputT, RequestEntryT extends
Serializable>
+ extends AsyncSinkWriter<InputT, RequestEntryT> {
+
+ public AsyncSinkWriterAdapter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ Sink.InitContext context,
+ AsyncSinkWriterConfiguration configuration,
+ Collection<BufferedRequestState<RequestEntryT>>
bufferedRequestStates) {
+ super(
+ elementConverter,
+ new WriterInitContextAdapter(context),
+ configuration,
+ bufferedRequestStates);
+ }
+
+ public AsyncSinkWriterAdapter(
+ ElementConverter<InputT, RequestEntryT> elementConverter,
+ WriterInitContext context,
+ AsyncSinkWriterConfiguration configuration,
+ Collection<BufferedRequestState<RequestEntryT>>
bufferedRequestStates,
+ BatchCreator<RequestEntryT> batchCreator,
+ RequestBuffer<RequestEntryT> bufferedRequestEntries) {
+ super(
+ elementConverter,
+ context,
+ configuration,
+ bufferedRequestStates,
+ batchCreator,
+ bufferedRequestEntries);
+ }
+}
