You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by re...@apache.org on 2024/04/26 02:19:18 UTC
(flink-cdc) 01/02: [cdc-connector][db2] Db2 support incremental source
This is an automated email from the ASF dual-hosted git repository.
renqs pushed a commit to branch release-3.1
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit 5b6e893e94fbb9f2606efe74415cb72dce1208b6
Author: gong <ge...@gmail.com>
AuthorDate: Thu Dec 14 02:06:07 2023 +0800
[cdc-connector][db2] Db2 support incremental source
---
.../connectors/legacy-flink-cdc-sources/db2-cdc.md | 157 +++++-
.../legacy-flink-cdc-sources/overview.md | 2 +-
.../tutorials/db2-tutorial.md | 5 +-
.../connectors/legacy-flink-cdc-sources/db2-cdc.md | 149 ++++-
.../legacy-flink-cdc-sources/overview.md | 2 +-
.../tutorials/db2-tutorial.md | 5 +-
.../base/relational/JdbcSourceEventDispatcher.java | 3 +
.../handler/SchemaChangeEventHandler.java | 1 +
.../base/source/reader/external/FetchTask.java | 1 +
.../flink-connector-db2-cdc/pom.xml | 7 +-
.../io/debezium/connector/db2/Db2Connection.java | 600 +++++++++++++++++++++
.../db2/Db2StreamingChangeEventSource.java | 502 +++++++++++++++++
.../apache/flink/cdc/connectors/db2/Db2Source.java | 2 +-
.../connectors/db2/source/Db2SourceBuilder.java | 262 +++++++++
.../db2/source/config/Db2SourceConfig.java | 92 ++++
.../db2/source/config/Db2SourceConfigFactory.java | 108 ++++
.../db2/source/dialect/Db2ChunkSplitter.java | 360 +++++++++++++
.../connectors/db2/source/dialect/Db2Dialect.java | 175 ++++++
.../dialect/Db2PooledDataSourceFactory.java} | 25 +-
.../connectors/db2/source/dialect/Db2Schema.java | 80 +++
.../db2/source/fetch/Db2ScanFetchTask.java | 348 ++++++++++++
.../source/fetch/Db2SourceFetchTaskContext.java | 311 +++++++++++
.../db2/source/fetch/Db2StreamFetchTask.java | 173 ++++++
.../handler/Db2SchemaChangeEventHandler.java | 47 ++
.../connectors/db2/source/offset/LsnFactory.java | 69 +++
.../connectors/db2/source/offset/LsnOffset.java | 95 ++++
.../db2/source/utils/Db2ConnectionUtils.java | 83 +++
.../connectors/db2/source/utils/Db2TypeUtils.java | 83 +++
.../cdc/connectors/db2/source/utils/Db2Utils.java | 442 +++++++++++++++
.../cdc/connectors/db2/table/Db2TableSource.java | 148 ++++-
.../db2/table/Db2TableSourceFactory.java | 109 +++-
.../cdc/connectors/db2/table/StartupOptions.java | 3 +
.../db2/Db2ParallelSourceExampleTest.java | 62 +++
.../flink/cdc/connectors/db2/Db2SourceTest.java | 34 +-
.../flink/cdc/connectors/db2/Db2TestBase.java | 196 ++++++-
.../cdc/connectors/db2/source/Db2SourceITCase.java | 333 ++++++++++++
.../connectors/db2/table/Db2ConnectorITCase.java | 149 +++--
.../db2/table/Db2TableSourceFactoryTest.java | 71 ++-
.../connectors/db2/testutils/RecordsFormatter.java | 97 ++++
.../cdc/connectors/db2/testutils/TestTable.java | 96 ++++
.../src/test/resources/db2_server/Dockerfile | 1 -
.../src/test/resources/db2_server/asncdctables.sql | 2 +-
.../test/resources/db2_server/column_type_test.sql | 7 +-
.../src/test/resources/db2_server/customers.sql | 48 ++
.../src/test/resources/db2_server/dbsetup.sh | 6 +-
.../src/test/resources/db2_server/inventory.sql | 40 +-
.../test/resources/db2_server/startup-cdc-demo.sql | 23 -
.../src/test/resources/log4j2-test.properties | 4 +-
.../mysql/source/MySqlSourceTestBase.java | 4 +-
.../flink-sql-connector-db2-cdc/pom.xml | 2 +-
.../flink/cdc/connectors/tests/Db2E2eITCase.java | 160 +++++-
.../src/test/resources/docker/db2/Dockerfile | 1 -
.../test/resources/docker/db2/column_type_test.sql | 11 +-
.../src/test/resources/docker/db2/customers.sql | 48 ++
.../src/test/resources/docker/db2/dbsetup.sh | 11 +-
.../src/test/resources/docker/db2/inventory.sql | 40 +-
.../test/resources/docker/db2/startup-cdc-demo.sql | 23 -
57 files changed, 5592 insertions(+), 326 deletions(-)
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md
index cf69abb00..03d3a2b24 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md
@@ -26,7 +26,7 @@ under the License.
# Db2 CDC Connector
-The Db2 CDC connector allows for reading snapshot data and incremental data from Db2 database. This document
+The Db2 CDC connector allows for reading snapshot data and incremental data from Db2 database. This document
describes how to setup the db2 CDC connector to run SQL queries against Db2 databases.
@@ -39,7 +39,7 @@ describes how to setup the db2 CDC connector to run SQL queries against Db2 data
Dependencies
------------
-In order to set up the Db2 CDC connector, the following table provides dependency information for both projects
+In order to set up the Db2 CDC connector, the following table provides dependency information for both projects
using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
### Maven dependency
@@ -48,10 +48,10 @@ using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR
### SQL Client JAR
-Download [flink-sql-connector-db2-cdc-3.0.1.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-db2-cdc/3.0.1/flink-sql-connector-db2-cdc-3.0.1.jar) and
+Download [flink-sql-connector-db2-cdc-3.0.1.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-db2-cdc/3.0.1/flink-sql-connector-db2-cdc-3.0.1.jar) and
put it under `<FLINK_HOME>/lib/`.
-**Note:** Refer to
+**Note:** Refer to
[flink-sql-connector-db2-cdc](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-db2-cdc),
more released versions will be available in the Maven central warehouse.
@@ -104,7 +104,8 @@ Flink SQL> CREATE TABLE products (
ID INT NOT NULL,
NAME STRING,
DESCRIPTION STRING,
- WEIGHT DECIMAL(10,3)
+ WEIGHT DECIMAL(10,3),
+ PRIMARY KEY(ID) NOT ENFORCED
) WITH (
'connector' = 'db2-cdc',
'hostname' = 'localhost',
@@ -112,10 +113,9 @@ Flink SQL> CREATE TABLE products (
'username' = 'root',
'password' = '123456',
'database-name' = 'mydb',
- 'schema-name' = 'myschema',
- 'table-name' = 'products');
+ 'table-name' = 'myschema.products');
--- read snapshot and binlogs from products table
+-- read snapshot and redo logs from products table
Flink SQL> SELECT * FROM products;
```
@@ -168,20 +168,13 @@ Connector Options
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Database name of the Db2 server to monitor.</td>
- </tr>
- <tr>
- <td>schema-name</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>Schema name of the Db2 database to monitor.</td>
</tr>
<tr>
<td>table-name</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Table name of the Db2 database to monitor.</td>
+ <td>Table name of the Db2 database to monitor, e.g.: "db1.table1"</td>
</tr>
<tr>
<td>port</td>
@@ -210,6 +203,46 @@ for more detailed information.</td>
If not set, then ZoneId.systemDefault() is used to determine the server time zone.
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether enable parallelism snapshot.</td>
+ </tr>
+ <tr>
+ <td>chunk-meta.group.size</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Integer</td>
+ <td>The group size of chunk meta, if the meta size exceeds the group size, the meta will be divided into multiple groups.</td>
+ </tr>
+ <tr>
+ <td>chunk-key.even-distribution.factor.lower-bound</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">0.05d</td>
+ <td>Double</td>
+ <td>The lower bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distribution or not.
+ The table chunks would use evenly calculation optimization when the data distribution is even, and the query for splitting would happen when it is uneven.
+ The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.</td>
+ </tr>
+ <tr>
+ <td>chunk-key.even-distribution.factor.upper-bound</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">1000.0d</td>
+ <td>Double</td>
+ <td>The upper bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distribution or not.
+ The table chunks would use evenly calculation optimization when the data distribution is even, and the query for splitting would happen when it is uneven.
+ The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.chunk.key-column</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table.
+ By default, the chunk key is the first column of the primary key. This column must be a column of the primary key.</td>
+ </tr>
<tr>
<td>debezium.*</td>
<td>optional</td>
@@ -220,19 +253,67 @@ Db2 server.
For example: <code>'debezium.snapshot.mode' = 'never'</code>.
See more about the <a href="https://debezium.io/documentation/reference/1.9/connectors/db2.html#db2-connector-properties">Debezium's Db2 Connector properties</a></td>
</tr>
+ <tr>
+ <td>scan.incremental.close-idle-reader.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to close idle readers at the end of the snapshot phase. <br>
+ The flink version is required to be greater than or equal to 1.14 when 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' is set to true.<br>
+ If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
+ so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
+ </td>
+ </tr>
</tbody>
</table>
</div>
+Available Metadata
+----------------
+
+The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.
+
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 15%">Key</th>
+ <th class="text-left" style="width: 30%">DataType</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>table_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the table that contain the row.</td>
+ </tr>
+ <tr>
+ <td>schema_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the schema that contain the row.</td>
+ </tr>
+ <tr>
+ <td>database_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the database that contain the row.</td>
+ </tr>
+ <tr>
+ <td>op_ts</td>
+ <td>TIMESTAMP_LTZ(3) NOT NULL</td>
+ <td>It indicates the time that the change was made in the database. <br>If the record is read from snapshot of the table instead of the change stream, the value is always 0.</td>
+ </tr>
+ </tbody>
+</table>
+
Features
--------
### Startup Reading Position
The config option `scan.startup.mode` specifies the startup mode for DB2 CDC consumer. The valid enumerations are:
-- `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest binlog.
+- `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest redo logs.
- `latest-offset`: Never to perform snapshot on the monitored database tables upon first startup, just read from
- the end of the binlog which means only have the changes since the connector was started.
+ the end of the redo logs which means only have the changes since the connector was started.
_Note: the mechanism of `scan.startup.mode` option relying on Debezium's `snapshot.mode` configuration. So please do not using them together. If you speicifying both `scan.startup.mode` and `debezium.snapshot.mode` options in the table DDL, it may make `scan.startup.mode` doesn't work._
@@ -273,6 +354,46 @@ public class Db2SourceExample {
}
```
+The DB2 CDC incremental connector (after 3.1.0) can be used as the following shows:
+```java
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder.Db2IncrementalSource;
+import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
+
+public class Db2ParallelSourceExample {
+
+ public static void main(String[] args) throws Exception {
+
+ Db2IncrementalSource<String> sqlServerSource =
+ new Db2SourceBuilder()
+ .hostname("localhost")
+ .port(50000)
+ .databaseList("TESTDB")
+ .tableList("DB2INST1.CUSTOMERS")
+ .username("flink")
+ .password("flinkpw")
+ .deserializer(new JsonDebeziumDeserializationSchema())
+ .startupOptions(StartupOptions.initial())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ // enable checkpoint
+ env.enableCheckpointing(3000);
+ // set the source parallelism to 2
+ env.fromSource(sqlServerSource, WatermarkStrategy.noWatermarks(), "Db2IncrementalSource")
+ .setParallelism(2)
+ .print()
+ .setParallelism(1);
+
+ env.execute("Print DB2 Snapshot + Change Stream");
+ }
+}
+```
+
Data Type Mapping
----------------
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md
index 5fc627371..a4ca4dd14 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md
@@ -81,7 +81,7 @@ The following table shows the current features of the connector:
| [sqlserver-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/sqlserver-cdc" >}}) | ✅ | ✅ | ✅ | ✅ |
| [oceanbase-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/oceanbase-cdc" >}}) | ❌ | ❌ | ❌ | ❌ |
| [tidb-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/tidb-cdc" >}}) | ✅ | ❌ | ✅ | ❌ |
-| [db2-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/db2-cdc" >}}) | ❌ | ❌ | ✅ | ❌ |
+| [db2-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/db2-cdc" >}}) | ✅ | ✅ | ✅ | ✅ |
| [vitess-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/vitess-cdc" >}}) | ✅ | ❌ | ✅ | ❌ |
## Usage for Table/SQL API
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
index 722da0ae0..1a19e93ef 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
@@ -111,9 +111,8 @@ Flink SQL> CREATE TABLE products (
'port' = '50000',
'username' = 'db2inst1',
'password' = 'admin',
- 'database-name' = 'testdb',
- 'schema-name' = 'DB2INST1',
- 'table-name' = 'PRODUCTS'
+ 'database-name' = 'TESTDB',
+ 'table-name' = 'DB2INST1.PRODUCTS'
);
Flink SQL> CREATE TABLE es_products (
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md
index 5c1e45079..484b869fb 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/db2-cdc.md
@@ -104,7 +104,8 @@ Flink SQL> CREATE TABLE products (
ID INT NOT NULL,
NAME STRING,
DESCRIPTION STRING,
- WEIGHT DECIMAL(10,3)
+ WEIGHT DECIMAL(10,3),
+ PRIMARY KEY(ID) NOT ENFORCED
) WITH (
'connector' = 'db2-cdc',
'hostname' = 'localhost',
@@ -112,10 +113,9 @@ Flink SQL> CREATE TABLE products (
'username' = 'root',
'password' = '123456',
'database-name' = 'mydb',
- 'schema-name' = 'myschema',
- 'table-name' = 'products');
+ 'table-name' = 'myschema.products');
--- read snapshot and binlogs from products table
+-- read snapshot and redo logs from products table
Flink SQL> SELECT * FROM products;
```
@@ -168,20 +168,13 @@ Connector Options
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Database name of the Db2 server to monitor.</td>
- </tr>
- <tr>
- <td>schema-name</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>Schema name of the Db2 database to monitor.</td>
</tr>
<tr>
<td>table-name</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Table name of the Db2 database to monitor.</td>
+ <td>Table name of the Db2 database to monitor, e.g.: "db1.table1"</td>
</tr>
<tr>
<td>port</td>
@@ -210,6 +203,46 @@ for more detailed information.</td>
If not set, then ZoneId.systemDefault() is used to determine the server time zone.
</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether enable parallelism snapshot.</td>
+ </tr>
+ <tr>
+ <td>chunk-meta.group.size</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Integer</td>
+ <td>The group size of chunk meta, if the meta size exceeds the group size, the meta will be divided into multiple groups.</td>
+ </tr>
+ <tr>
+ <td>chunk-key.even-distribution.factor.lower-bound</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">0.05d</td>
+ <td>Double</td>
+ <td>The lower bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distribution or not.
+ The table chunks would use evenly calculation optimization when the data distribution is even, and the query for splitting would happen when it is uneven.
+ The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.</td>
+ </tr>
+ <tr>
+ <td>chunk-key.even-distribution.factor.upper-bound</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">1000.0d</td>
+ <td>Double</td>
+ <td>The upper bound of chunk key distribution factor. The distribution factor is used to determine whether the table is evenly distribution or not.
+ The table chunks would use evenly calculation optimization when the data distribution is even, and the query for splitting would happen when it is uneven.
+ The distribution factor could be calculated by (MAX(id) - MIN(id) + 1) / rowCount.</td>
+ </tr>
+ <tr>
+ <td>scan.incremental.snapshot.chunk.key-column</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The chunk key of table snapshot, captured tables are split into multiple chunks by a chunk key when read the snapshot of table.
+ By default, the chunk key is the first column of the primary key. This column must be a column of the primary key.</td>
+ </tr>
<tr>
<td>debezium.*</td>
<td>optional</td>
@@ -220,19 +253,67 @@ Db2 server.
For example: <code>'debezium.snapshot.mode' = 'never'</code>.
See more about the <a href="https://debezium.io/documentation/reference/1.9/connectors/db2.html#db2-connector-properties">Debezium's Db2 Connector properties</a></td>
</tr>
+ <tr>
+ <td>scan.incremental.close-idle-reader.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to close idle readers at the end of the snapshot phase. <br>
+ The flink version is required to be greater than or equal to 1.14 when 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' is set to true.<br>
+ If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
+ so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
+ </td>
+ </tr>
</tbody>
</table>
</div>
+Available Metadata
+----------------
+
+The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.
+
+<table class="colwidths-auto docutils">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 15%">Key</th>
+ <th class="text-left" style="width: 30%">DataType</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td>table_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the table that contain the row.</td>
+ </tr>
+ <tr>
+ <td>schema_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the schema that contain the row.</td>
+ </tr>
+ <tr>
+ <td>database_name</td>
+ <td>STRING NOT NULL</td>
+ <td>Name of the database that contain the row.</td>
+ </tr>
+ <tr>
+ <td>op_ts</td>
+ <td>TIMESTAMP_LTZ(3) NOT NULL</td>
+ <td>It indicates the time that the change was made in the database. <br>If the record is read from snapshot of the table instead of the change stream, the value is always 0.</td>
+ </tr>
+ </tbody>
+</table>
+
Features
--------
### Startup Reading Position
The config option `scan.startup.mode` specifies the startup mode for DB2 CDC consumer. The valid enumerations are:
-- `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest binlog.
+- `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest redo logs.
- `latest-offset`: Never to perform snapshot on the monitored database tables upon first startup, just read from
- the end of the binlog which means only have the changes since the connector was started.
+ the end of the redo logs which means only have the changes since the connector was started.
_Note: the mechanism of `scan.startup.mode` option relying on Debezium's `snapshot.mode` configuration. So please do not using them together. If you speicifying both `scan.startup.mode` and `debezium.snapshot.mode` options in the table DDL, it may make `scan.startup.mode` doesn't work._
@@ -273,6 +354,46 @@ public class Db2SourceExample {
}
```
+The DB2 CDC incremental connector (after 3.1.0) can be used as the following shows:
+```java
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder.Db2IncrementalSource;
+import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
+
+public class Db2ParallelSourceExample {
+
+ public static void main(String[] args) throws Exception {
+
+ Db2IncrementalSource<String> sqlServerSource =
+ new Db2SourceBuilder()
+ .hostname("localhost")
+ .port(50000)
+ .databaseList("TESTDB")
+ .tableList("DB2INST1.CUSTOMERS")
+ .username("flink")
+ .password("flinkpw")
+ .deserializer(new JsonDebeziumDeserializationSchema())
+ .startupOptions(StartupOptions.initial())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ // enable checkpoint
+ env.enableCheckpointing(3000);
+ // set the source parallelism to 2
+ env.fromSource(sqlServerSource, WatermarkStrategy.noWatermarks(), "Db2IncrementalSource")
+ .setParallelism(2)
+ .print()
+ .setParallelism(1);
+
+ env.execute("Print DB2 Snapshot + Change Stream");
+ }
+}
+```
+
Data Type Mapping
----------------
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md
index 20cd61135..f98800752 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md
@@ -81,7 +81,7 @@ The following table shows the current features of the connector:
| [sqlserver-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/sqlserver-cdc" >}}) | ✅ | ✅ | ✅ | ✅ |
| [oceanbase-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/oceanbase-cdc" >}}) | ❌ | ❌ | ❌ | ❌ |
| [tidb-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/tidb-cdc" >}}) | ✅ | ❌ | ✅ | ❌ |
-| [db2-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/db2-cdc" >}}) | ❌ | ❌ | ✅ | ❌ |
+| [db2-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/db2-cdc" >}}) | ✅ | ✅ | ✅ | ✅ |
| [vitess-cdc]({{< ref "docs/connectors/legacy-flink-cdc-sources/vitess-cdc" >}}) | ✅ | ❌ | ✅ | ❌ |
## Usage for Table/SQL API
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
index 54e69b7f6..464005312 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
@@ -111,9 +111,8 @@ Flink SQL> CREATE TABLE products (
'port' = '50000',
'username' = 'db2inst1',
'password' = 'admin',
- 'database-name' = 'testdb',
- 'schema-name' = 'DB2INST1',
- 'table-name' = 'PRODUCTS'
+ 'database-name' = 'TESTDB',
+ 'table-name' = 'DB2INST1.PRODUCTS'
);
Flink SQL> CREATE TABLE es_products (
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/JdbcSourceEventDispatcher.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/JdbcSourceEventDispatcher.java
index 7886bf8c8..deb6794ac 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/JdbcSourceEventDispatcher.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/JdbcSourceEventDispatcher.java
@@ -209,6 +209,9 @@ public class JdbcSourceEventDispatcher<P extends Partition> extends EventDispatc
@Override
public void schemaChangeEvent(SchemaChangeEvent event) throws InterruptedException {
+ if (SchemaChangeEvent.SchemaChangeEventType.DROP.equals(event.getType())) {
+ LOG.info("Received drop table event " + event + " at offset: " + event.getOffset());
+ }
historizedSchema.applySchemaChange(event);
if (connectorConfig.isSchemaChangesHistoryEnabled()) {
try {
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/handler/SchemaChangeEventHandler.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/handler/SchemaChangeEventHandler.java
index bdc9f39ff..ede51c8c3 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/handler/SchemaChangeEventHandler.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/relational/handler/SchemaChangeEventHandler.java
@@ -26,5 +26,6 @@ import java.util.Map;
/** This handler helps to parse the source struct in SchemaChangeEvent and generate source info. */
@Experimental
public interface SchemaChangeEventHandler {
+
Map<String, Object> parseSource(SchemaChangeEvent event);
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/FetchTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/FetchTask.java
index beab207e4..fbf86b598 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/FetchTask.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/FetchTask.java
@@ -52,6 +52,7 @@ public interface FetchTask<Split> {
/** Base context used in the execution of fetch task. */
interface Context {
+
void configure(SourceSplitBase sourceSplitBase);
ChangeEventQueue<DataChangeEvent> getQueue();
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/pom.xml b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/pom.xml
index 85cb27321..1efec2aef 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/pom.xml
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/pom.xml
@@ -30,6 +30,12 @@ limitations under the License.
<packaging>jar</packaging>
<dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<!-- Debezium dependencies -->
<dependency>
<groupId>org.apache.flink</groupId>
@@ -68,7 +74,6 @@ limitations under the License.
<groupId>com.ibm.db2.jcc</groupId>
<artifactId>db2jcc</artifactId>
<version>db2jcc4</version>
- <scope>test</scope>
</dependency>
<dependency>
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java
new file mode 100644
index 000000000..a916808bf
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2Connection.java
@@ -0,0 +1,600 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.debezium.connector.db2;
+
+import com.ibm.db2.jcc.DB2Driver;
+import io.debezium.config.Configuration;
+import io.debezium.jdbc.JdbcConfiguration;
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.Column;
+import io.debezium.relational.ColumnEditor;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.util.BoundedConcurrentHashMap;
+import io.debezium.util.Collect;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Copied from Debezium 1.9.8.Final. {@link JdbcConnection} extension to be used with IBM Db2
+ *
+ * @author Horia Chiorean (hchiorea@redhat.com), Jiri Pechanec, Peter Urbanetz
+ */
+public class Db2Connection extends JdbcConnection {
+
+ private static final String GET_DATABASE_NAME =
+ "SELECT CURRENT SERVER FROM SYSIBM.SYSDUMMY1"; // DB2
+
+ private static Logger LOGGER = LoggerFactory.getLogger(Db2Connection.class);
+
+ private static final String CDC_SCHEMA = "ASNCDC";
+
+ private static final String STATEMENTS_PLACEHOLDER = "#";
+ private static final String GET_MAX_LSN =
+ "SELECT max(t.SYNCHPOINT) FROM ( SELECT CD_NEW_SYNCHPOINT AS SYNCHPOINT FROM "
+ + CDC_SCHEMA
+ + ".IBMSNAP_REGISTER UNION ALL SELECT SYNCHPOINT AS SYNCHPOINT FROM "
+ + CDC_SCHEMA
+ + ".IBMSNAP_REGISTER) t";
+
+ private static final String LOCK_TABLE = "SELECT * FROM # WITH CS"; // DB2
+
+ private static final String LSN_TO_TIMESTAMP =
+ "SELECT CURRENT TIMEstamp FROM sysibm.sysdummy1 WHERE ? > X'00000000000000000000000000000000'";
+
+ private static final String GET_ALL_CHANGES_FOR_TABLE =
+ "SELECT "
+ + "CASE "
+ + "WHEN IBMSNAP_OPERATION = 'D' AND (LEAD(cdc.IBMSNAP_OPERATION,1,'X') OVER (PARTITION BY cdc.IBMSNAP_COMMITSEQ ORDER BY cdc.IBMSNAP_INTENTSEQ)) ='I' THEN 3 "
+ + "WHEN IBMSNAP_OPERATION = 'I' AND (LAG(cdc.IBMSNAP_OPERATION,1,'X') OVER (PARTITION BY cdc.IBMSNAP_COMMITSEQ ORDER BY cdc.IBMSNAP_INTENTSEQ)) ='D' THEN 4 "
+ + "WHEN IBMSNAP_OPERATION = 'D' THEN 1 "
+ + "WHEN IBMSNAP_OPERATION = 'I' THEN 2 "
+ + "END "
+ + "OPCODE,"
+ + "cdc.* "
+ + "FROM ASNCDC.# cdc WHERE IBMSNAP_COMMITSEQ >= ? AND IBMSNAP_COMMITSEQ <= ? "
+ + "order by IBMSNAP_COMMITSEQ, IBMSNAP_INTENTSEQ";
+
+ private static final String GET_LIST_OF_CDC_ENABLED_TABLES =
+ "select r.SOURCE_OWNER, r.SOURCE_TABLE, r.CD_OWNER, r.CD_TABLE, r.CD_NEW_SYNCHPOINT, r.CD_OLD_SYNCHPOINT, t.TBSPACEID, t.TABLEID , CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER )from "
+ + CDC_SCHEMA
+ + ".IBMSNAP_REGISTER r left JOIN SYSCAT.TABLES t ON r.SOURCE_OWNER = t.TABSCHEMA AND r.SOURCE_TABLE = t.TABNAME WHERE r.SOURCE_OWNER <> ''";
+
+ // No new Tabels 1=0
+ private static final String GET_LIST_OF_NEW_CDC_ENABLED_TABLES =
+ "select CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER ) AS OBJECTID, "
+ + " CD_OWNER CONCAT '.' CONCAT CD_TABLE, "
+ + " CD_NEW_SYNCHPOINT, "
+ + " CD_OLD_SYNCHPOINT "
+ + "from ASNCDC.IBMSNAP_REGISTER r left JOIN SYSCAT.TABLES t ON r.SOURCE_OWNER = t.TABSCHEMA AND r.SOURCE_TABLE = t.TABNAME "
+ + "WHERE r.SOURCE_OWNER <> '' AND 1=0 AND CD_NEW_SYNCHPOINT > ? AND CD_OLD_SYNCHPOINT < ? ";
+
+ private static final String GET_LIST_OF_KEY_COLUMNS =
+ "SELECT "
+ + "CAST((t.TBSPACEID * 65536 + t.TABLEID )AS INTEGER ) as objectid, "
+ + "c.colname,c.colno,c.keyseq "
+ + "FROM syscat.tables as t "
+ + "inner join syscat.columns as c on t.tabname = c.tabname and t.tabschema = c.tabschema and c.KEYSEQ > 0 AND "
+ + "t.tbspaceid = CAST(BITAND( ? , 4294901760) / 65536 AS SMALLINT) AND t.tableid= CAST(BITAND( ? , 65535) AS SMALLINT)";
+
+ private static final int CHANGE_TABLE_DATA_COLUMN_OFFSET = 4;
+
+ private static final String QUOTED_CHARACTER = "\"";
+
+ private static final String URL_PATTERN =
+ "jdbc:db2://${"
+ + JdbcConfiguration.HOSTNAME
+ + "}:${"
+ + JdbcConfiguration.PORT
+ + "}/${"
+ + JdbcConfiguration.DATABASE
+ + "}";
+
+ private static final ConnectionFactory FACTORY =
+ JdbcConnection.patternBasedFactory(
+ URL_PATTERN,
+ DB2Driver.class.getName(),
+ Db2Connection.class.getClassLoader(),
+ JdbcConfiguration.PORT.withDefault(
+ Db2ConnectorConfig.PORT.defaultValueAsString()));
+
+ /**
+ * actual name of the database, which could differ in casing from the database name given in the
+ * connector config.
+ */
+ private final String realDatabaseName;
+
+ private final BoundedConcurrentHashMap<Lsn, Instant> lsnToInstantCache;
+
+ /**
+ * Creates a new connection using the supplied configuration.
+ *
+ * @param config {@link Configuration} instance, may not be null.
+ */
+ public Db2Connection(JdbcConfiguration config) {
+ super(config, FACTORY, QUOTED_CHARACTER, QUOTED_CHARACTER);
+ lsnToInstantCache = new BoundedConcurrentHashMap<>(100);
+ realDatabaseName = retrieveRealDatabaseName();
+ }
+
+ /** @return the current largest log sequence number */
+ public Lsn getMaxLsn() throws SQLException {
+ return queryAndMap(
+ GET_MAX_LSN,
+ singleResultMapper(
+ rs -> {
+ final Lsn ret = Lsn.valueOf(rs.getBytes(1));
+ LOGGER.trace("Current maximum lsn is {}", ret);
+ return ret;
+ },
+ "Maximum LSN query must return exactly one value"));
+ }
+
+ /**
+ * Provides all changes recorded by the DB2 CDC capture process for a given table.
+ *
+ * @param tableId - the requested table changes
+ * @param fromLsn - closed lower bound of interval of changes to be provided
+ * @param toLsn - closed upper bound of interval of changes to be provided
+ * @param consumer - the change processor
+ * @throws SQLException
+ */
+ public void getChangesForTable(
+ TableId tableId, Lsn fromLsn, Lsn toLsn, ResultSetConsumer consumer)
+ throws SQLException {
+ final String query =
+ GET_ALL_CHANGES_FOR_TABLE.replace(STATEMENTS_PLACEHOLDER, cdcNameForTable(tableId));
+ prepareQuery(
+ query,
+ statement -> {
+ statement.setBytes(1, fromLsn.getBinary());
+ statement.setBytes(2, toLsn.getBinary());
+ },
+ consumer);
+ }
+
+ /**
+ * Provides all changes recorder by the DB2 CDC capture process for a set of tables.
+ *
+ * @param changeTables - the requested tables to obtain changes for
+ * @param intervalFromLsn - closed lower bound of interval of changes to be provided
+ * @param intervalToLsn - closed upper bound of interval of changes to be provided
+ * @param consumer - the change processor
+ * @throws SQLException
+ */
+ public void getChangesForTables(
+ Db2ChangeTable[] changeTables,
+ Lsn intervalFromLsn,
+ Lsn intervalToLsn,
+ BlockingMultiResultSetConsumer consumer)
+ throws SQLException, InterruptedException {
+ final String[] queries = new String[changeTables.length];
+ final StatementPreparer[] preparers = new StatementPreparer[changeTables.length];
+
+ int idx = 0;
+ for (Db2ChangeTable changeTable : changeTables) {
+ final String query =
+ GET_ALL_CHANGES_FOR_TABLE.replace(
+ STATEMENTS_PLACEHOLDER, changeTable.getCaptureInstance());
+ queries[idx] = query;
+ // If the table was added in the middle of queried buffer we need
+ // to adjust from to the first LSN available
+ LOGGER.trace(
+ "Getting changes for table {} in range[{}, {}]",
+ changeTable,
+ intervalFromLsn,
+ intervalToLsn);
+ preparers[idx] =
+ statement -> {
+ statement.setBytes(1, intervalFromLsn.getBinary());
+ statement.setBytes(2, intervalToLsn.getBinary());
+ };
+
+ idx++;
+ }
+ prepareQuery(queries, preparers, consumer);
+ }
+
+ /**
+ * Obtain the next available position in the database log.
+ *
+ * @param lsn - LSN of the current position
+ * @return LSN of the next position in the database
+ * @throws SQLException
+ */
+ public Lsn incrementLsn(Lsn lsn) throws SQLException {
+ return lsn.increment();
+ }
+
+ /**
+ * Map a commit LSN to a point in time when the commit happened.
+ *
+ * @param lsn - LSN of the commit
+ * @return time when the commit was recorded into the database log
+ * @throws SQLException
+ */
+ public Instant timestampOfLsn(Lsn lsn) throws SQLException {
+ final String query = LSN_TO_TIMESTAMP;
+
+ if (lsn.getBinary() == null) {
+ return null;
+ }
+
+ Instant cachedInstant = lsnToInstantCache.get(lsn);
+ if (cachedInstant != null) {
+ return cachedInstant;
+ }
+
+ return prepareQueryAndMap(
+ query,
+ statement -> {
+ statement.setBytes(1, lsn.getBinary());
+ },
+ singleResultMapper(
+ rs -> {
+ final Timestamp ts = rs.getTimestamp(1);
+ final Instant ret = (ts == null) ? null : ts.toInstant();
+ LOGGER.trace("Timestamp of lsn {} is {}", lsn, ret);
+ if (ret != null) {
+ lsnToInstantCache.put(lsn, ret);
+ }
+ return ret;
+ },
+ "LSN to timestamp query must return exactly one value"));
+ }
+
+ @Override
+ public Optional<Timestamp> getCurrentTimestamp() throws SQLException {
+ return queryAndMap(
+ "SELECT CURRENT_TIMESTAMP result FROM sysibm.sysdummy1",
+ rs -> rs.next() ? Optional.of(rs.getTimestamp(1)) : Optional.empty());
+ }
+
+ /**
+ * Creates an exclusive lock for a given table.
+ *
+ * @param tableId to be locked
+ * @throws SQLException
+ */
+ public void lockTable(TableId tableId) throws SQLException {
+ final String lockTableStmt = LOCK_TABLE.replace(STATEMENTS_PLACEHOLDER, tableId.table());
+ execute(lockTableStmt);
+ }
+
+ private String cdcNameForTable(TableId tableId) {
+ return Db2ObjectNameQuoter.quoteNameIfNecessary(tableId.schema() + '_' + tableId.table());
+ }
+
+ public static class CdcEnabledTable {
+ private final String tableId;
+ private final String captureName;
+ private final Lsn fromLsn;
+
+ private CdcEnabledTable(String tableId, String captureName, Lsn fromLsn) {
+ this.tableId = tableId;
+ this.captureName = captureName;
+ this.fromLsn = fromLsn;
+ }
+
+ public String getTableId() {
+ return tableId;
+ }
+
+ public String getCaptureName() {
+ return captureName;
+ }
+
+ public Lsn getFromLsn() {
+ return fromLsn;
+ }
+ }
+
+ public Set<Db2ChangeTable> listOfChangeTables() throws SQLException {
+ final String query = GET_LIST_OF_CDC_ENABLED_TABLES;
+
+ return queryAndMap(
+ query,
+ rs -> {
+ final Set<Db2ChangeTable> changeTables = new HashSet<>();
+ while (rs.next()) {
+ /**
+ * changeTables.add( new ChangeTable( new TableId(realDatabaseName,
+ * rs.getString(1), rs.getString(2)), rs.getString(3), rs.getInt(4),
+ * Lsn.valueOf(rs.getBytes(6)), Lsn.valueOf(rs.getBytes(7))
+ *
+ * <p>)
+ */
+ changeTables.add(
+ new Db2ChangeTable(
+ new TableId("", rs.getString(1), rs.getString(2)),
+ rs.getString(4),
+ rs.getInt(9),
+ Lsn.valueOf(rs.getBytes(5)),
+ Lsn.valueOf(rs.getBytes(6))));
+ }
+ return changeTables;
+ });
+ }
+
+ public Set<Db2ChangeTable> listOfNewChangeTables(Lsn fromLsn, Lsn toLsn) throws SQLException {
+ final String query = GET_LIST_OF_NEW_CDC_ENABLED_TABLES;
+
+ return prepareQueryAndMap(
+ query,
+ ps -> {
+ ps.setBytes(1, fromLsn.getBinary());
+ ps.setBytes(2, toLsn.getBinary());
+ },
+ rs -> {
+ final Set<Db2ChangeTable> changeTables = new HashSet<>();
+ while (rs.next()) {
+ changeTables.add(
+ new Db2ChangeTable(
+ rs.getString(2),
+ rs.getInt(1),
+ Lsn.valueOf(rs.getBytes(3)),
+ Lsn.valueOf(rs.getBytes(4))));
+ }
+ return changeTables;
+ });
+ }
+
+ public Table getTableSchemaFromTable(Db2ChangeTable changeTable) throws SQLException {
+ final DatabaseMetaData metadata = connection().getMetaData();
+
+ List<Column> columns = new ArrayList<>();
+ try (ResultSet rs =
+ metadata.getColumns(
+ null,
+ changeTable.getSourceTableId().schema(),
+ changeTable.getSourceTableId().table(),
+ null)) {
+ while (rs.next()) {
+ readTableColumn(rs, changeTable.getSourceTableId(), null)
+ .ifPresent(ce -> columns.add(ce.create()));
+ }
+ }
+
+ final List<String> pkColumnNames =
+ readPrimaryKeyNames(metadata, changeTable.getSourceTableId());
+ Collections.sort(columns);
+ return Table.editor()
+ .tableId(changeTable.getSourceTableId())
+ .addColumns(columns)
+ .setPrimaryKeyNames(pkColumnNames)
+ .create();
+ }
+
+ public Table getTableSchemaFromChangeTable(Db2ChangeTable changeTable) throws SQLException {
+ final DatabaseMetaData metadata = connection().getMetaData();
+ final TableId changeTableId = changeTable.getChangeTableId();
+
+ List<ColumnEditor> columnEditors = new ArrayList<>();
+ try (ResultSet rs =
+ metadata.getColumns(null, changeTableId.schema(), changeTableId.table(), null)) {
+ while (rs.next()) {
+ readTableColumn(rs, changeTableId, null).ifPresent(columnEditors::add);
+ }
+ }
+
+ // The first 5 columns and the last column of the change table are CDC metadata
+ // final List<Column> columns = columnEditors.subList(CHANGE_TABLE_DATA_COLUMN_OFFSET,
+ // columnEditors.size() - 1).stream()
+ final List<Column> columns =
+ columnEditors.subList(CHANGE_TABLE_DATA_COLUMN_OFFSET, columnEditors.size())
+ .stream()
+ .map(
+ c ->
+ c.position(c.position() - CHANGE_TABLE_DATA_COLUMN_OFFSET)
+ .create())
+ .collect(Collectors.toList());
+
+ final List<String> pkColumnNames = new ArrayList<>();
+ /**
+ * URB prepareQuery(GET_LIST_OF_KEY_COLUMNS, ps -> ps.setInt(1,
+ * changeTable.getChangeTableObjectId()), rs -> { while (rs.next()) {
+ * pkColumnNames.add(rs.getString(2)); } });
+ */
+ prepareQuery(
+ GET_LIST_OF_KEY_COLUMNS,
+ ps -> {
+ ps.setInt(1, changeTable.getChangeTableObjectId());
+ ps.setInt(1, changeTable.getChangeTableObjectId());
+ },
+ rs -> {
+ while (rs.next()) {
+ pkColumnNames.add(rs.getString(2));
+ }
+ });
+ Collections.sort(columns);
+ return Table.editor()
+ .tableId(changeTable.getSourceTableId())
+ .addColumns(columns)
+ .setPrimaryKeyNames(pkColumnNames)
+ .create();
+ }
+
+ public String getNameOfChangeTable(String captureName) {
+ return captureName + "_CT";
+ }
+
+ public String getRealDatabaseName() {
+ return realDatabaseName;
+ }
+
+ @Override
+ protected boolean isTableUniqueIndexIncluded(String indexName, String columnName) {
+ // ignore indices with no name;
+ return indexName != null;
+ }
+
+ private String retrieveRealDatabaseName() {
+ try {
+ return queryAndMap(
+ GET_DATABASE_NAME,
+ singleResultMapper(rs -> rs.getString(1), "Could not retrieve database name"));
+ } catch (SQLException e) {
+ throw new RuntimeException("Couldn't obtain database name", e);
+ }
+ }
+
+ /**
+ * Returns a JDBC connection string for the current configuration.
+ *
+ * @return a {@code String} where the variables in {@code urlPattern} are replaced with values
+ * from the configuration
+ */
+ public String connectionString() {
+ return connectionString(URL_PATTERN);
+ }
+
+ @Override
+ public String quotedTableIdString(TableId tableId) {
+ StringBuilder quoted = new StringBuilder();
+ if (tableId.schema() != null && !tableId.schema().isEmpty()) {
+ quoted.append(Db2ObjectNameQuoter.quoteNameIfNecessary(tableId.schema())).append(".");
+ }
+ quoted.append(Db2ObjectNameQuoter.quoteNameIfNecessary(tableId.table()));
+ return quoted.toString();
+ }
+
+ protected String resolveCatalogName(String catalogName) {
+ return realDatabaseName;
+ }
+
+ public List<String> readPrimaryKeyNames(DatabaseMetaData metadata, TableId id)
+ throws SQLException {
+ final List<String> pkColumnNames = new ArrayList<>();
+ try (ResultSet rs = metadata.getPrimaryKeys(null, id.schema(), id.table())) {
+ while (rs.next()) {
+ String columnName = rs.getString(4);
+ int columnIndex = rs.getInt(5);
+ Collect.set(pkColumnNames, columnIndex - 1, columnName, null);
+ }
+ }
+ return pkColumnNames;
+ }
+
+ public List<String> readTableUniqueIndices(DatabaseMetaData metadata, TableId id)
+ throws SQLException {
+ final List<String> uniqueIndexColumnNames = new ArrayList<>();
+ final Set<String> excludedIndexNames = new HashSet<>();
+ try (ResultSet rs = metadata.getIndexInfo(null, id.schema(), id.table(), true, true)) {
+ String firstIndexName = null;
+ while (rs.next()) {
+ final String indexName = rs.getString(6);
+ final String columnName = rs.getString(9);
+ final int columnIndex = rs.getInt(8);
+
+ // Some databases return a null index name record, often as the first row.
+ // This index should be ignored, as should any row with an index that has been
+ // marked excluded
+ if (indexName == null || excludedIndexNames.contains(indexName)) {
+ continue;
+ }
+
+ // Check whether the index and/or its column is included by the connector
+ boolean indexIncluded = isTableUniqueIndexIncluded(indexName, columnName);
+ if (!indexIncluded) {
+ // The connector considered the index and/or its column to be excluded.
+ // Register the index as an excluded index.
+ excludedIndexNames.add(indexName);
+
+ if (firstIndexName == null || indexName.equals(firstIndexName)) {
+ // We either have not yet found a valid first index or the index is the same
+ // as the
+ // current index we have processed a column for. The later can happen when
+ // any
+ // column after the first is seen as an excluded pattern, and in this case
+ // the
+ // entire index state should be discarded and any future rows related to it
+ // will
+ // also be discarded.
+ firstIndexName = null;
+ uniqueIndexColumnNames.clear();
+ continue;
+ }
+ }
+
+ if (firstIndexName == null) {
+ firstIndexName = indexName;
+ }
+
+ if (!indexName.equals(firstIndexName)) {
+ // This means we've reached a point in the result set where we've processed two
+ // index
+ // mappings and both are included by the connector, so we return the first index
+ // we
+ // completely mapped.
+ return uniqueIndexColumnNames;
+ }
+
+ if (columnName != null) {
+ // The returned columnIndex is 0 when columnName is null. These are related
+ // to table statistics that get returned as part of the index descriptors
+ // and should be ignored.
+ Collect.set(uniqueIndexColumnNames, columnIndex - 1, columnName, null);
+ }
+ }
+ }
+ return uniqueIndexColumnNames;
+ }
+
+ /**
+ * Get the identifiers of the tables.
+ *
+ * @param databaseCatalog the name of the catalog, which is typically the database name; may be
+ * an empty string for tables that have no catalog, or {@code null} if the catalog name
+ * should not be used to narrow the list of table identifiers
+ * @param schemaNamePattern the pattern used to match database schema names, which may be "" to
+ * match only those tables with no schema or {@code null} if the schema name should not be
+ * used to narrow the list of table identifiers
+ * @param tableNamePattern the pattern used to match database table names, which may be null to
+ * match all table names
+ * @param tableTypes the set of table types to include in the results, which may be null for all
+ * table types
+ * @return the set of {@link TableId}s; never null but possibly empty
+ * @throws SQLException if an error occurs while accessing the database metadata
+ */
+ public Set<TableId> readTableNames(
+ String databaseCatalog,
+ String schemaNamePattern,
+ String tableNamePattern,
+ String[] tableTypes)
+ throws SQLException {
+ if (tableNamePattern == null) {
+ tableNamePattern = "%";
+ }
+ Set<TableId> tableIds = new HashSet<>();
+ DatabaseMetaData metadata = connection().getMetaData();
+ try (ResultSet rs =
+ metadata.getTables(
+ databaseCatalog, schemaNamePattern, tableNamePattern, tableTypes)) {
+ while (rs.next()) {
+ String catalogName = resolveCatalogName(rs.getString(1));
+ String schemaName = rs.getString(2);
+ String tableName = rs.getString(3);
+ TableId tableId = new TableId(catalogName, schemaName, tableName);
+ tableIds.add(tableId);
+ }
+ }
+ return tableIds;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java
new file mode 100644
index 000000000..261a59fb2
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/io/debezium/connector/db2/Db2StreamingChangeEventSource.java
@@ -0,0 +1,502 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.debezium.connector.db2;
+
+import io.debezium.pipeline.ErrorHandler;
+import io.debezium.pipeline.EventDispatcher;
+import io.debezium.pipeline.source.spi.ChangeTableResultSet;
+import io.debezium.pipeline.source.spi.StreamingChangeEventSource;
+import io.debezium.relational.TableId;
+import io.debezium.schema.DatabaseSchema;
+import io.debezium.schema.SchemaChangeEvent.SchemaChangeEventType;
+import io.debezium.util.Clock;
+import io.debezium.util.Metronome;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/**
+ * Copied from Debezium project(1.9.8.final)
+ *
+ * <p>A {@link StreamingChangeEventSource} based on DB2 change data capture functionality. A main
+ * loop polls database DDL change and change data tables and turns them into change events.
+ *
+ * <p>The connector uses CDC functionality of DB2 that is implemented as as a process that monitors
+ * source table and write changes from the table into the change table.
+ *
+ * <p>The main loop keeps a pointer to the LSN of changes that were already processed. It queries
+ * all change tables and get result set of changes. It always finds the smallest LSN across all
+ * tables and the change is converted into the event message and sent downstream. The process
+ * repeats until all result sets are empty. The LSN is marked and the procedure repeats.
+ *
+ * <p>The schema changes detection follows the procedure recommended by DB2 CDC documentation. The
+ * database operator should create one more capture process (and table) when a table schema is
+ * updated. The code detects presence of two change tables for a single source table. It decides
+ * which table is the new one depending on LSNs stored in them. The loop streams changes from the
+ * older table till there are events in new table with the LSN larger than in the old one. Then the
+ * change table is switched and streaming is executed from the new one.
+ *
+ * @author Jiri Pechanec, Peter Urbanetz
+ */
+public class Db2StreamingChangeEventSource
+ implements StreamingChangeEventSource<Db2Partition, Db2OffsetContext> {
+
+ private static final int COL_COMMIT_LSN = 2;
+ private static final int COL_ROW_LSN = 3;
+ private static final int COL_OPERATION = 1;
+ private static final int COL_DATA = 5;
+
+ private static final Pattern MISSING_CDC_FUNCTION_CHANGES_ERROR =
+ Pattern.compile("Invalid object name 'cdc.fn_cdc_get_all_changes_(.*)'\\.");
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(Db2StreamingChangeEventSource.class);
+
+ /** Connection used for reading CDC tables. */
+ private final Db2Connection dataConnection;
+
+ /**
+ * A separate connection for retrieving timestamps; without it, adaptive buffering will not
+ * work.
+ */
+ private final Db2Connection metadataConnection;
+
+ private final EventDispatcher<Db2Partition, TableId> dispatcher;
+ private final ErrorHandler errorHandler;
+ private final Clock clock;
+ private final Db2DatabaseSchema schema;
+ private final Duration pollInterval;
+ private final Db2ConnectorConfig connectorConfig;
+
+ public Db2StreamingChangeEventSource(
+ Db2ConnectorConfig connectorConfig,
+ Db2Connection dataConnection,
+ Db2Connection metadataConnection,
+ EventDispatcher<Db2Partition, TableId> dispatcher,
+ ErrorHandler errorHandler,
+ Clock clock,
+ Db2DatabaseSchema schema) {
+ this.connectorConfig = connectorConfig;
+ this.dataConnection = dataConnection;
+ this.metadataConnection = metadataConnection;
+ this.dispatcher = dispatcher;
+ this.errorHandler = errorHandler;
+ this.clock = clock;
+ this.schema = schema;
+ this.pollInterval = connectorConfig.getPollInterval();
+ }
+
+ @Override
+ public void execute(
+ ChangeEventSourceContext context,
+ Db2Partition partition,
+ Db2OffsetContext offsetContext)
+ throws InterruptedException {
+ if (!connectorConfig.getSnapshotMode().shouldStream()) {
+ LOGGER.info("Streaming is not enabled in current configuration");
+ return;
+ }
+
+ final Metronome metronome = Metronome.sleeper(pollInterval, clock);
+ final Queue<Db2ChangeTable> schemaChangeCheckpoints =
+ new PriorityQueue<>((x, y) -> x.getStopLsn().compareTo(y.getStopLsn()));
+ try {
+ final AtomicReference<Db2ChangeTable[]> tablesSlot =
+ new AtomicReference<>(getCdcTablesToQuery(partition, offsetContext));
+
+ final TxLogPosition lastProcessedPositionOnStart = offsetContext.getChangePosition();
+ final long lastProcessedEventSerialNoOnStart = offsetContext.getEventSerialNo();
+ LOGGER.info(
+ "Last position recorded in offsets is {}[{}]",
+ lastProcessedPositionOnStart,
+ lastProcessedEventSerialNoOnStart);
+
+ TxLogPosition lastProcessedPosition = lastProcessedPositionOnStart;
+
+ // LSN should be increased for the first run only immediately after snapshot completion
+ // otherwise we might skip an incomplete transaction after restart
+ boolean shouldIncreaseFromLsn = offsetContext.isSnapshotCompleted();
+ while (context.isRunning()) {
+ final Lsn currentMaxLsn = dataConnection.getMaxLsn();
+
+ // Shouldn't happen if the agent is running, but it is better to guard against such
+ // situation
+ if (!currentMaxLsn.isAvailable()) {
+ LOGGER.warn(
+ "No maximum LSN recorded in the database; please ensure that the DB2 Agent is running");
+ metronome.pause();
+ continue;
+ }
+ // There is no change in the database
+ if (currentMaxLsn.equals(lastProcessedPosition.getCommitLsn())
+ && shouldIncreaseFromLsn) {
+ LOGGER.debug("No change in the database");
+ metronome.pause();
+ continue;
+ }
+
+ // Reading interval is inclusive so we need to move LSN forward but not for first
+ // run as TX might not be streamed completely
+ final Lsn fromLsn =
+ lastProcessedPosition.getCommitLsn().isAvailable() && shouldIncreaseFromLsn
+ ? dataConnection.incrementLsn(lastProcessedPosition.getCommitLsn())
+ : lastProcessedPosition.getCommitLsn();
+ shouldIncreaseFromLsn = true;
+
+ while (!schemaChangeCheckpoints.isEmpty()) {
+ migrateTable(partition, offsetContext, schemaChangeCheckpoints);
+ }
+ if (!dataConnection.listOfNewChangeTables(fromLsn, currentMaxLsn).isEmpty()) {
+ final Db2ChangeTable[] tables = getCdcTablesToQuery(partition, offsetContext);
+ tablesSlot.set(tables);
+ for (Db2ChangeTable table : tables) {
+ if (table.getStartLsn().isBetween(fromLsn, currentMaxLsn)) {
+ LOGGER.info("Schema will be changed for {}", table);
+ schemaChangeCheckpoints.add(table);
+ }
+ }
+ }
+ try {
+ dataConnection.getChangesForTables(
+ tablesSlot.get(),
+ fromLsn,
+ currentMaxLsn,
+ resultSets -> {
+ long eventSerialNoInInitialTx = 1;
+ final int tableCount = resultSets.length;
+ final ChangeTablePointer[] changeTables =
+ new ChangeTablePointer[tableCount];
+ final Db2ChangeTable[] tables = tablesSlot.get();
+
+ for (int i = 0; i < tableCount; i++) {
+ changeTables[i] =
+ new ChangeTablePointer(tables[i], resultSets[i]);
+ changeTables[i].next();
+ }
+
+ for (; ; ) {
+ ChangeTablePointer tableWithSmallestLsn = null;
+ for (ChangeTablePointer changeTable : changeTables) {
+ if (changeTable.isCompleted()) {
+ continue;
+ }
+ if (tableWithSmallestLsn == null
+ || changeTable.compareTo(tableWithSmallestLsn)
+ < 0) {
+ tableWithSmallestLsn = changeTable;
+ }
+ }
+ if (tableWithSmallestLsn == null) {
+ // No more LSNs available
+ break;
+ }
+
+ if (!(tableWithSmallestLsn.getChangePosition().isAvailable()
+ && tableWithSmallestLsn
+ .getChangePosition()
+ .getInTxLsn()
+ .isAvailable())) {
+ LOGGER.error(
+ "Skipping change {} as its LSN is NULL which is not expected",
+ tableWithSmallestLsn);
+ tableWithSmallestLsn.next();
+ continue;
+ }
+ // After restart for changes that were executed before the last
+ // committed offset
+ if (tableWithSmallestLsn
+ .getChangePosition()
+ .compareTo(lastProcessedPositionOnStart)
+ < 0) {
+ LOGGER.info(
+ "Skipping change {} as its position is smaller than the last recorded position {}",
+ tableWithSmallestLsn,
+ lastProcessedPositionOnStart);
+ tableWithSmallestLsn.next();
+ continue;
+ }
+ // After restart for change that was the last committed and
+ // operations in it before the last committed offset
+ if (tableWithSmallestLsn
+ .getChangePosition()
+ .compareTo(lastProcessedPositionOnStart)
+ == 0
+ && eventSerialNoInInitialTx
+ <= lastProcessedEventSerialNoOnStart) {
+ LOGGER.info(
+ "Skipping change {} as its order in the transaction {} is smaller than or equal to the last recorded operation {}[{}]",
+ tableWithSmallestLsn,
+ eventSerialNoInInitialTx,
+ lastProcessedPositionOnStart,
+ lastProcessedEventSerialNoOnStart);
+ eventSerialNoInInitialTx++;
+ tableWithSmallestLsn.next();
+ continue;
+ }
+ if (tableWithSmallestLsn
+ .getChangeTable()
+ .getStopLsn()
+ .isAvailable()
+ && tableWithSmallestLsn
+ .getChangeTable()
+ .getStopLsn()
+ .compareTo(
+ tableWithSmallestLsn
+ .getChangePosition()
+ .getCommitLsn())
+ <= 0) {
+ LOGGER.debug(
+ "Skipping table change {} as its stop LSN is smaller than the last recorded LSN {}",
+ tableWithSmallestLsn,
+ tableWithSmallestLsn.getChangePosition());
+ tableWithSmallestLsn.next();
+ continue;
+ }
+ LOGGER.trace("Processing change {}", tableWithSmallestLsn);
+ if (!schemaChangeCheckpoints.isEmpty()) {
+ if (tableWithSmallestLsn
+ .getChangePosition()
+ .getCommitLsn()
+ .compareTo(
+ schemaChangeCheckpoints
+ .peek()
+ .getStopLsn())
+ >= 0) {
+ migrateTable(
+ partition,
+ offsetContext,
+ schemaChangeCheckpoints);
+ }
+ }
+ final TableId tableId =
+ tableWithSmallestLsn
+ .getChangeTable()
+ .getSourceTableId();
+ final int operation = tableWithSmallestLsn.getOperation();
+ final Object[] data = tableWithSmallestLsn.getData();
+
+ // UPDATE consists of two consecutive events, first event
+ // contains
+ // the row before it was updated and the second the row after
+ // it was updated
+ int eventCount = 1;
+ if (operation == Db2ChangeRecordEmitter.OP_UPDATE_BEFORE) {
+ if (!tableWithSmallestLsn.next()
+ || tableWithSmallestLsn.getOperation()
+ != Db2ChangeRecordEmitter.OP_UPDATE_AFTER) {
+ throw new IllegalStateException(
+ "The update before event at "
+ + tableWithSmallestLsn
+ .getChangePosition()
+ + " for table "
+ + tableId
+ + " was not followed by after event.\n Please report this as a bug together with a events around given LSN.");
+ }
+ eventCount = 2;
+ }
+ final Object[] dataNext =
+ (operation == Db2ChangeRecordEmitter.OP_UPDATE_BEFORE)
+ ? tableWithSmallestLsn.getData()
+ : null;
+
+ offsetContext.setChangePosition(
+ tableWithSmallestLsn.getChangePosition(), eventCount);
+ offsetContext.event(
+ tableWithSmallestLsn
+ .getChangeTable()
+ .getSourceTableId(),
+ metadataConnection.timestampOfLsn(
+ tableWithSmallestLsn
+ .getChangePosition()
+ .getCommitLsn()));
+
+ dispatcher.dispatchDataChangeEvent(
+ partition,
+ tableId,
+ new Db2ChangeRecordEmitter(
+ partition,
+ offsetContext,
+ operation,
+ data,
+ dataNext,
+ clock));
+ tableWithSmallestLsn.next();
+ }
+ });
+ lastProcessedPosition = TxLogPosition.valueOf(currentMaxLsn);
+ // Terminate the transaction otherwise CDC could not be disabled for tables
+ dataConnection.rollback();
+ // Determine whether to continue streaming in db2 cdc snapshot phase
+ afterHandleLsn(partition, currentMaxLsn);
+ } catch (SQLException e) {
+ tablesSlot.set(processErrorFromChangeTableQuery(e, tablesSlot.get()));
+ }
+ }
+ } catch (Exception e) {
+ errorHandler.setProducerThrowable(e);
+ }
+ }
+
+ private void migrateTable(
+ Db2Partition partition,
+ Db2OffsetContext offsetContext,
+ final Queue<Db2ChangeTable> schemaChangeCheckpoints)
+ throws InterruptedException, SQLException {
+ final Db2ChangeTable newTable = schemaChangeCheckpoints.poll();
+ LOGGER.info("Migrating schema to {}", newTable);
+ dispatcher.dispatchSchemaChangeEvent(
+ partition,
+ newTable.getSourceTableId(),
+ new Db2SchemaChangeEventEmitter(
+ partition,
+ offsetContext,
+ newTable,
+ metadataConnection.getTableSchemaFromTable(newTable),
+ SchemaChangeEventType.ALTER));
+ }
+
+ private Db2ChangeTable[] processErrorFromChangeTableQuery(
+ SQLException exception, Db2ChangeTable[] currentChangeTables) throws Exception {
+ final Matcher m = MISSING_CDC_FUNCTION_CHANGES_ERROR.matcher(exception.getMessage());
+ if (m.matches()) {
+ final String captureName = m.group(1);
+ LOGGER.info("Table is no longer captured with capture instance {}", captureName);
+ return Arrays.asList(currentChangeTables).stream()
+ .filter(x -> !x.getCaptureInstance().equals(captureName))
+ .collect(Collectors.toList())
+ .toArray(new Db2ChangeTable[0]);
+ }
+ throw exception;
+ }
+
+ private Db2ChangeTable[] getCdcTablesToQuery(
+ Db2Partition partition, Db2OffsetContext offsetContext)
+ throws SQLException, InterruptedException {
+ final Set<Db2ChangeTable> cdcEnabledTables = dataConnection.listOfChangeTables();
+
+ if (cdcEnabledTables.isEmpty()) {
+ LOGGER.warn(
+ "No table has enabled CDC or security constraints prevents getting the list of change tables");
+ }
+
+ final Map<TableId, List<Db2ChangeTable>> includedAndCdcEnabledTables =
+ cdcEnabledTables.stream()
+ .filter(
+ changeTable -> {
+ if (connectorConfig
+ .getTableFilters()
+ .dataCollectionFilter()
+ .isIncluded(changeTable.getSourceTableId())) {
+ return true;
+ } else {
+ LOGGER.info(
+ "CDC is enabled for table {} but the table is not included by connector",
+ changeTable);
+ return false;
+ }
+ })
+ .collect(Collectors.groupingBy(x -> x.getSourceTableId()));
+
+ if (includedAndCdcEnabledTables.isEmpty()) {
+ LOGGER.warn(DatabaseSchema.NO_CAPTURED_DATA_COLLECTIONS_WARNING);
+ }
+
+ final List<Db2ChangeTable> tables = new ArrayList<>();
+ for (List<Db2ChangeTable> captures : includedAndCdcEnabledTables.values()) {
+ Db2ChangeTable currentTable = captures.get(0);
+ if (captures.size() > 1) {
+ Db2ChangeTable futureTable;
+ if (captures.get(0).getStartLsn().compareTo(captures.get(1).getStartLsn()) < 0) {
+ futureTable = captures.get(1);
+ } else {
+ currentTable = captures.get(1);
+ futureTable = captures.get(0);
+ }
+ currentTable.setStopLsn(futureTable.getStartLsn());
+ tables.add(futureTable);
+ LOGGER.info(
+ "Multiple capture instances present for the same table: {} and {}",
+ currentTable,
+ futureTable);
+ }
+ if (schema.tableFor(currentTable.getSourceTableId()) == null) {
+ LOGGER.info(
+ "Table {} is new to be monitored by capture instance {}",
+ currentTable.getSourceTableId(),
+ currentTable.getCaptureInstance());
+ // this prevents potential NPE if there is no TableId information in the source info
+ offsetContext.event(currentTable.getSourceTableId(), Instant.now());
+
+ // We need to read the source table schema - nullability information cannot be
+ // obtained from change table
+ dispatcher.dispatchSchemaChangeEvent(
+ partition,
+ currentTable.getSourceTableId(),
+ new Db2SchemaChangeEventEmitter(
+ partition,
+ offsetContext,
+ currentTable,
+ dataConnection.getTableSchemaFromTable(currentTable),
+ SchemaChangeEventType.CREATE));
+ }
+ tables.add(currentTable);
+ }
+
+ return tables.toArray(new Db2ChangeTable[tables.size()]);
+ }
+
+ /**
+ * The logical representation of a position for the change in the transaction log. During each
+ * sourcing cycle it is necessary to query all change tables and then make a total order of
+ * changes across all tables.<br>
+ * This class represents an open database cursor over the change table that is able to move the
+ * cursor forward and report the LSN for the change to which the cursor now points.
+ *
+ * @author Jiri Pechanec
+ */
+ private static class ChangeTablePointer
+ extends ChangeTableResultSet<Db2ChangeTable, TxLogPosition> {
+
+ public ChangeTablePointer(Db2ChangeTable changeTable, ResultSet resultSet) {
+ super(changeTable, resultSet, COL_DATA);
+ }
+
+ @Override
+ protected int getOperation(ResultSet resultSet) throws SQLException {
+ return resultSet.getInt(COL_OPERATION);
+ }
+
+ @Override
+ protected TxLogPosition getNextChangePosition(ResultSet resultSet) throws SQLException {
+ return isCompleted()
+ ? TxLogPosition.NULL
+ : TxLogPosition.valueOf(
+ Lsn.valueOf(resultSet.getBytes(COL_COMMIT_LSN)),
+ Lsn.valueOf(resultSet.getBytes(COL_ROW_LSN)));
+ }
+ }
+
+ /** expose control to the user to stop the connector. */
+ protected void afterHandleLsn(Db2Partition partition, Lsn toLsn) {
+ // do nothing
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java
index 39479c3fe..2b704562d 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/Db2Source.java
@@ -17,7 +17,7 @@
package org.apache.flink.cdc.connectors.db2;
-import org.apache.flink.cdc.connectors.db2.table.StartupOptions;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.cdc.debezium.Validator;
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java
new file mode 100644
index 000000000..884aaabdb
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceBuilder.java
@@ -0,0 +1,262 @@
+/*
+ * 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.connectors.db2.source;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
+import org.apache.flink.cdc.connectors.db2.source.config.Db2SourceConfigFactory;
+import org.apache.flink.cdc.connectors.db2.source.dialect.Db2Dialect;
+import org.apache.flink.cdc.connectors.db2.source.offset.LsnFactory;
+import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
+
+import java.time.Duration;
+import java.util.Properties;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The builder class for {@link Db2IncrementalSource} to make it easier for the users to construct a
+ * {@link Db2IncrementalSource}.
+ *
+ * <p>Check the Java docs of each individual method to learn more about the settings to build a
+ * {@link Db2IncrementalSource}.
+ */
+@PublicEvolving
+public class Db2SourceBuilder<T> {
+
+ private final Db2SourceConfigFactory configFactory = new Db2SourceConfigFactory();
+
+ private LsnFactory offsetFactory;
+
+ private Db2Dialect dialect;
+
+ private DebeziumDeserializationSchema<T> deserializer;
+
+ /** Hostname of the Db2 database server. */
+ public Db2SourceBuilder<T> hostname(String hostname) {
+ this.configFactory.hostname(hostname);
+ return this;
+ }
+
+ /** Integer port number of the Db2 database server. */
+ public Db2SourceBuilder<T> port(int port) {
+ this.configFactory.port(port);
+ return this;
+ }
+
+ /**
+ * A required list of regular expressions that match database names to be monitored; any
+ * database name not included in the whitelist will be excluded from monitoring.
+ */
+ public Db2SourceBuilder<T> databaseList(String... databaseList) {
+ this.configFactory.databaseList(databaseList);
+ return this;
+ }
+
+ /**
+ * A required list of regular expressions that match fully-qualified table identifiers for
+ * tables to be monitored; any table not included in the list will be excluded from monitoring.
+ * Each identifier is of the form {@code <schemaName>.<tableName>}.
+ */
+ public Db2SourceBuilder<T> tableList(String... tableList) {
+ this.configFactory.tableList(tableList);
+ return this;
+ }
+
+ /** Name of the Db2 database to use when connecting to the Db2 database server. */
+ public Db2SourceBuilder<T> username(String username) {
+ this.configFactory.username(username);
+ return this;
+ }
+
+ /** Password to use when connecting to the Db2 database server. */
+ public Db2SourceBuilder<T> password(String password) {
+ this.configFactory.password(password);
+ return this;
+ }
+
+ /**
+ * The session time zone in database server, e.g. "America/Los_Angeles". It controls how the
+ * TIMESTAMP type in Db2 converted to STRING. See more
+ * https://debezium.io/documentation/reference/1.9/connectors/db2.html#db2-temporal-types
+ */
+ public Db2SourceBuilder<T> serverTimeZone(String timeZone) {
+ this.configFactory.serverTimeZone(timeZone);
+ return this;
+ }
+
+ /**
+ * The split size (number of rows) of table snapshot, captured tables are split into multiple
+ * splits when read the snapshot of table.
+ */
+ public Db2SourceBuilder<T> splitSize(int splitSize) {
+ this.configFactory.splitSize(splitSize);
+ return this;
+ }
+
+ /**
+ * The group size of split meta, if the meta size exceeds the group size, the meta will be will
+ * be divided into multiple groups.
+ */
+ public Db2SourceBuilder<T> splitMetaGroupSize(int splitMetaGroupSize) {
+ this.configFactory.splitMetaGroupSize(splitMetaGroupSize);
+ return this;
+ }
+
+ /**
+ * The upper bound of split key evenly distribution factor, the factor is used to determine
+ * whether the table is evenly distribution or not.
+ */
+ public Db2SourceBuilder<T> distributionFactorUpper(double distributionFactorUpper) {
+ this.configFactory.distributionFactorUpper(distributionFactorUpper);
+ return this;
+ }
+
+ /**
+ * The lower bound of split key evenly distribution factor, the factor is used to determine
+ * whether the table is evenly distribution or not.
+ */
+ public Db2SourceBuilder<T> distributionFactorLower(double distributionFactorLower) {
+ this.configFactory.distributionFactorLower(distributionFactorLower);
+ return this;
+ }
+
+ /** The maximum fetch size for per poll when read table snapshot. */
+ public Db2SourceBuilder<T> fetchSize(int fetchSize) {
+ this.configFactory.fetchSize(fetchSize);
+ return this;
+ }
+
+ /**
+ * The maximum time that the connector should wait after trying to connect to the Db2 database
+ * server before timing out.
+ */
+ public Db2SourceBuilder<T> connectTimeout(Duration connectTimeout) {
+ this.configFactory.connectTimeout(connectTimeout);
+ return this;
+ }
+
+ /** The max retry times to get connection. */
+ public Db2SourceBuilder<T> connectMaxRetries(int connectMaxRetries) {
+ this.configFactory.connectMaxRetries(connectMaxRetries);
+ return this;
+ }
+
+ /** The connection pool size. */
+ public Db2SourceBuilder<T> connectionPoolSize(int connectionPoolSize) {
+ this.configFactory.connectionPoolSize(connectionPoolSize);
+ return this;
+ }
+
+ /** Whether the {@link Db2IncrementalSource} should output the schema changes or not. */
+ public Db2SourceBuilder<T> includeSchemaChanges(boolean includeSchemaChanges) {
+ this.configFactory.includeSchemaChanges(includeSchemaChanges);
+ return this;
+ }
+
+ /** Specifies the startup options. */
+ public Db2SourceBuilder<T> startupOptions(StartupOptions startupOptions) {
+ this.configFactory.startupOptions(startupOptions);
+ return this;
+ }
+
+ /**
+ * The chunk key of table snapshot, captured tables are split into multiple chunks by the chunk
+ * key column when read the snapshot of table.
+ */
+ public Db2SourceBuilder<T> chunkKeyColumn(String chunkKeyColumn) {
+ this.configFactory.chunkKeyColumn(chunkKeyColumn);
+ return this;
+ }
+
+ /** The Debezium Db2 connector properties. For example, "snapshot.mode". */
+ public Db2SourceBuilder<T> debeziumProperties(Properties properties) {
+ this.configFactory.debeziumProperties(properties);
+ return this;
+ }
+
+ /**
+ * The deserializer used to convert from consumed {@link
+ * org.apache.kafka.connect.source.SourceRecord}.
+ */
+ public Db2SourceBuilder<T> deserializer(DebeziumDeserializationSchema<T> deserializer) {
+ this.deserializer = deserializer;
+ return this;
+ }
+
+ /**
+ * Whether to close idle readers at the end of the snapshot phase. This feature depends on
+ * FLIP-147: Support Checkpoints After Tasks Finished. The flink version is required to be
+ * greater than or equal to 1.14, and the configuration <code>
+ * 'execution.checkpointing.checkpoints-after-tasks-finish.enabled'</code> needs to be set to
+ * true.
+ *
+ * <p>See more <a
+ * href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-147%3A+Support+Checkpoints+After+Tasks+Finished">FLIP-147:
+ * Support Checkpoints After Tasks Finished</a>.
+ */
+ public Db2SourceBuilder<T> closeIdleReaders(boolean closeIdleReaders) {
+ this.configFactory.closeIdleReaders(closeIdleReaders);
+ return this;
+ }
+
+ /**
+ * Whether to skip backfill in snapshot reading phase.
+ *
+ * <p>If backfill is skipped, changes on captured tables during snapshot phase will be consumed
+ * later in redo logs reading phase instead of being merged into the snapshot.
+ *
+ * <p>WARNING: Skipping backfill might lead to data inconsistency because some redo logs events
+ * happened within the snapshot phase might be replayed (only at-least-once semantic is
+ * promised). For example updating an already updated value in snapshot, or deleting an already
+ * deleted entry in snapshot. These replayed redo logs events should be handled specially.
+ */
+ public Db2SourceBuilder<T> skipSnapshotBackfill(boolean skipSnapshotBackfill) {
+ this.configFactory.skipSnapshotBackfill(skipSnapshotBackfill);
+ return this;
+ }
+
+ /**
+ * Build the {@link Db2IncrementalSource}.
+ *
+ * @return a Db2ParallelSource with the settings made for this builder.
+ */
+ public Db2IncrementalSource<T> build() {
+ this.offsetFactory = new LsnFactory();
+ this.dialect = new Db2Dialect(configFactory.create(0));
+ return new Db2IncrementalSource<T>(
+ configFactory, checkNotNull(deserializer), offsetFactory, dialect);
+ }
+
+ /** The {@link JdbcIncrementalSource} implementation for Db2. */
+ public static class Db2IncrementalSource<T> extends JdbcIncrementalSource<T> {
+
+ public Db2IncrementalSource(
+ Db2SourceConfigFactory configFactory,
+ DebeziumDeserializationSchema<T> deserializationSchema,
+ LsnFactory offsetFactory,
+ Db2Dialect dataSourceDialect) {
+ super(configFactory, deserializationSchema, offsetFactory, dataSourceDialect);
+ }
+
+ public static <T> Db2SourceBuilder<T> builder() {
+ return new Db2SourceBuilder<>();
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
new file mode 100644
index 000000000..f63dcdda4
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfig.java
@@ -0,0 +1,92 @@
+/*
+ * 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.connectors.db2.source.config;
+
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+
+import io.debezium.config.Configuration;
+import io.debezium.connector.db2.Db2ConnectorConfig;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Describes the connection information of the Db2 database and the configuration information for
+ * performing snapshotting and streaming reading, such as splitSize.
+ */
+public class Db2SourceConfig extends JdbcSourceConfig {
+
+ public Db2SourceConfig(
+ StartupOptions startupOptions,
+ List<String> databaseList,
+ List<String> tableList,
+ int splitSize,
+ int splitMetaGroupSize,
+ double distributionFactorUpper,
+ double distributionFactorLower,
+ boolean includeSchemaChanges,
+ boolean closeIdleReaders,
+ Properties dbzProperties,
+ Configuration dbzConfiguration,
+ String driverClassName,
+ String hostname,
+ int port,
+ String username,
+ String password,
+ int fetchSize,
+ String serverTimeZone,
+ Duration connectTimeout,
+ int connectMaxRetries,
+ int connectionPoolSize,
+ String chunkKeyColumn,
+ boolean skipSnapshotBackfill) {
+ super(
+ startupOptions,
+ databaseList,
+ null,
+ tableList,
+ splitSize,
+ splitMetaGroupSize,
+ distributionFactorUpper,
+ distributionFactorLower,
+ includeSchemaChanges,
+ closeIdleReaders,
+ dbzProperties,
+ dbzConfiguration,
+ driverClassName,
+ hostname,
+ port,
+ username,
+ password,
+ fetchSize,
+ serverTimeZone,
+ connectTimeout,
+ connectMaxRetries,
+ connectionPoolSize,
+ chunkKeyColumn,
+ skipSnapshotBackfill,
+ false);
+ }
+
+ @Override
+ public Db2ConnectorConfig getDbzConnectorConfig() {
+ return new Db2ConnectorConfig(getDbzConfiguration());
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
new file mode 100644
index 000000000..8b792876e
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java
@@ -0,0 +1,108 @@
+/*
+ * 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.connectors.db2.source.config;
+
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfigFactory;
+import org.apache.flink.cdc.connectors.base.source.EmbeddedFlinkDatabaseHistory;
+
+import io.debezium.config.Configuration;
+import io.debezium.connector.db2.Db2Connector;
+
+import java.util.Properties;
+import java.util.UUID;
+
+import static org.apache.flink.cdc.connectors.base.utils.EnvironmentUtils.checkSupportCheckpointsAfterTasksFinished;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Factory for creating {@link Db2SourceConfig}. */
+public class Db2SourceConfigFactory extends JdbcSourceConfigFactory {
+
+ private static final String DATABASE_SERVER_NAME = "Db2_transaction_log_source";
+ private static final String DRIVER_ClASS_NAME = "com.ibm.db2.jcc.DB2Driver";
+
+ @Override
+ public Db2SourceConfig create(int subtask) {
+ checkSupportCheckpointsAfterTasksFinished(closeIdleReaders);
+ Properties props = new Properties();
+ props.setProperty("connector.class", Db2Connector.class.getCanonicalName());
+
+ // set database history impl to flink database history
+ props.setProperty(
+ "database.history", EmbeddedFlinkDatabaseHistory.class.getCanonicalName());
+ props.setProperty("database.history.instance.name", UUID.randomUUID() + "_" + subtask);
+
+ // hard code server name, because we don't need to distinguish it, docs:
+ // Logical name that identifies and provides a namespace for the SQL Server database
+ // server that you want Debezium to capture. The logical name should be unique across
+ // all other connectors, since it is used as a prefix for all Kafka topic names
+ // emanating from this connector. Only alphanumeric characters and underscores should be
+ // used.
+ props.setProperty("database.server.name", DATABASE_SERVER_NAME);
+ props.setProperty("database.hostname", checkNotNull(hostname));
+ props.setProperty("database.user", checkNotNull(username));
+ props.setProperty("database.password", checkNotNull(password));
+ props.setProperty("database.port", String.valueOf(port));
+ props.setProperty("database.history.skip.unparseable.ddl", String.valueOf(true));
+ props.setProperty("database.dbname", checkNotNull(databaseList.get(0)));
+
+ if (tableList != null) {
+ props.setProperty("table.include.list", String.join(",", tableList));
+ }
+
+ switch (startupOptions.startupMode) {
+ case INITIAL:
+ props.setProperty("snapshot.mode", "initial");
+ break;
+ case LATEST_OFFSET:
+ props.setProperty("snapshot.mode", "schema_only");
+ break;
+ default:
+ throw new UnsupportedOperationException();
+ }
+
+ if (dbzProperties != null) {
+ props.putAll(dbzProperties);
+ }
+
+ Configuration dbzConfiguration = Configuration.from(props);
+ return new Db2SourceConfig(
+ startupOptions,
+ databaseList,
+ tableList,
+ splitSize,
+ splitMetaGroupSize,
+ distributionFactorUpper,
+ distributionFactorLower,
+ includeSchemaChanges,
+ closeIdleReaders,
+ props,
+ dbzConfiguration,
+ DRIVER_ClASS_NAME,
+ hostname,
+ port,
+ username,
+ password,
+ fetchSize,
+ serverTimeZone,
+ connectTimeout,
+ connectMaxRetries,
+ connectionPoolSize,
+ chunkKeyColumn,
+ skipSnapshotBackfill);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2ChunkSplitter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2ChunkSplitter.java
new file mode 100644
index 000000000..7cc6ddbf9
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2ChunkSplitter.java
@@ -0,0 +1,360 @@
+/*
+ * 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.connectors.db2.source.dialect;
+
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
+import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect;
+import org.apache.flink.cdc.connectors.base.source.assigner.splitter.ChunkRange;
+import org.apache.flink.cdc.connectors.base.source.assigner.splitter.JdbcSourceChunkSplitter;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
+import org.apache.flink.cdc.connectors.base.utils.ObjectUtils;
+import org.apache.flink.cdc.connectors.db2.source.utils.Db2TypeUtils;
+import org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges.TableChange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare;
+
+/**
+ * The splitter to split the table into chunks using primary-key (by default) or a given split key.
+ */
+public class Db2ChunkSplitter implements JdbcSourceChunkSplitter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Db2ChunkSplitter.class);
+
+ private final JdbcSourceConfig sourceConfig;
+ private final JdbcDataSourceDialect dialect;
+
+ public Db2ChunkSplitter(JdbcSourceConfig sourceConfig, JdbcDataSourceDialect dialect) {
+ this.sourceConfig = sourceConfig;
+ this.dialect = dialect;
+ }
+
+ private static String splitId(TableId tableId, int chunkId) {
+ return tableId.toString() + ":" + chunkId;
+ }
+
+ private static void maySleep(int count, TableId tableId) {
+ // every 10 queries to sleep 0.1s
+ if (count % 10 == 0) {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ // nothing to do
+ }
+ LOG.info("JdbcSourceChunkSplitter has split {} chunks for table {}", count, tableId);
+ }
+ }
+
+ @Override
+ public Collection<SnapshotSplit> generateSplits(TableId tableId) {
+ try (JdbcConnection jdbc = dialect.openJdbcConnection(sourceConfig)) {
+
+ LOG.info("Start splitting table {} into chunks...", tableId);
+ long start = System.currentTimeMillis();
+
+ Table table = dialect.queryTableSchema(jdbc, tableId).getTable();
+ Column splitColumn = Db2Utils.getSplitColumn(table, sourceConfig.getChunkKeyColumn());
+ final List<ChunkRange> chunks;
+ try {
+ chunks = splitTableIntoChunks(jdbc, tableId, splitColumn);
+ } catch (SQLException e) {
+ throw new FlinkRuntimeException("Failed to split chunks for table " + tableId, e);
+ }
+
+ // convert chunks into splits
+ List<SnapshotSplit> splits = new ArrayList<>();
+ RowType splitType = getSplitType(splitColumn);
+ for (int i = 0; i < chunks.size(); i++) {
+ ChunkRange chunk = chunks.get(i);
+ SnapshotSplit split =
+ createSnapshotSplit(
+ jdbc,
+ tableId,
+ i,
+ splitType,
+ chunk.getChunkStart(),
+ chunk.getChunkEnd());
+ splits.add(split);
+ }
+
+ long end = System.currentTimeMillis();
+ LOG.info(
+ "Split table {} into {} chunks, time cost: {}ms.",
+ tableId,
+ splits.size(),
+ end - start);
+ return splits;
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(
+ String.format("Generate Splits for table %s error", tableId), e);
+ }
+ }
+
+ @Override
+ public Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
+ throws SQLException {
+ return Db2Utils.queryMinMax(jdbc, tableId, columnName);
+ }
+
+ @Override
+ public Object queryMin(
+ JdbcConnection jdbc, TableId tableId, String columnName, Object excludedLowerBound)
+ throws SQLException {
+ return Db2Utils.queryMin(jdbc, tableId, columnName, excludedLowerBound);
+ }
+
+ @Override
+ public DataType fromDbzColumn(Column splitColumn) {
+ return Db2TypeUtils.fromDbzColumn(splitColumn);
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // Utilities
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public Object queryNextChunkMax(
+ JdbcConnection jdbc,
+ TableId tableId,
+ String columnName,
+ int chunkSize,
+ Object includedLowerBound)
+ throws SQLException {
+ return Db2Utils.queryNextChunkMax(jdbc, tableId, columnName, chunkSize, includedLowerBound);
+ }
+
+ @Override
+ public Long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId) throws SQLException {
+ return Db2Utils.queryApproximateRowCnt(jdbc, tableId);
+ }
+
+ @Override
+ public String buildSplitScanQuery(
+ TableId tableId, RowType splitKeyType, boolean isFirstSplit, boolean isLastSplit) {
+ return Db2Utils.buildSplitScanQuery(tableId, splitKeyType, isFirstSplit, isLastSplit);
+ }
+
+ /**
+ * We can use evenly-sized chunks or unevenly-sized chunks when split table into chunks, using
+ * evenly-sized chunks which is much efficient, using unevenly-sized chunks which will request
+ * many queries and is not efficient.
+ */
+ private List<ChunkRange> splitTableIntoChunks(
+ JdbcConnection jdbc, TableId tableId, Column splitColumn) throws SQLException {
+ final String splitColumnName = splitColumn.name();
+ final Object[] minMax = queryMinMax(jdbc, tableId, splitColumnName);
+ final Object min = minMax[0];
+ final Object max = minMax[1];
+ if (min == null || max == null || min.equals(max)) {
+ // empty table, or only one row, return full table scan as a chunk
+ return Collections.singletonList(ChunkRange.all());
+ }
+
+ final int chunkSize = sourceConfig.getSplitSize();
+ final double distributionFactorUpper = sourceConfig.getDistributionFactorUpper();
+ final double distributionFactorLower = sourceConfig.getDistributionFactorLower();
+
+ if (isEvenlySplitColumn(splitColumn)) {
+ long approximateRowCnt = queryApproximateRowCnt(jdbc, tableId);
+ double distributionFactor =
+ calculateDistributionFactor(tableId, min, max, approximateRowCnt);
+
+ boolean dataIsEvenlyDistributed =
+ doubleCompare(distributionFactor, distributionFactorLower) >= 0
+ && doubleCompare(distributionFactor, distributionFactorUpper) <= 0;
+
+ if (dataIsEvenlyDistributed) {
+ // the minimum dynamic chunk size is at least 1
+ final int dynamicChunkSize = Math.max((int) (distributionFactor * chunkSize), 1);
+ return splitEvenlySizedChunks(
+ tableId, min, max, approximateRowCnt, chunkSize, dynamicChunkSize);
+ } else {
+ return splitUnevenlySizedChunks(
+ jdbc, tableId, splitColumnName, min, max, chunkSize);
+ }
+ } else {
+ return splitUnevenlySizedChunks(jdbc, tableId, splitColumnName, min, max, chunkSize);
+ }
+ }
+
+ /**
+ * Split table into evenly sized chunks based on the numeric min and max value of split column,
+ * and tumble chunks in step size.
+ */
+ private List<ChunkRange> splitEvenlySizedChunks(
+ TableId tableId,
+ Object min,
+ Object max,
+ long approximateRowCnt,
+ int chunkSize,
+ int dynamicChunkSize) {
+ LOG.info(
+ "Use evenly-sized chunk optimization for table {}, the approximate row count is {}, the chunk size is {}, the dynamic chunk size is {}",
+ tableId,
+ approximateRowCnt,
+ chunkSize,
+ dynamicChunkSize);
+ if (approximateRowCnt <= chunkSize) {
+ // there is no more than one chunk, return full table as a chunk
+ return Collections.singletonList(ChunkRange.all());
+ }
+
+ final List<ChunkRange> splits = new ArrayList<>();
+ Object chunkStart = null;
+ Object chunkEnd = ObjectUtils.plus(min, dynamicChunkSize);
+ while (ObjectUtils.compare(chunkEnd, max) <= 0) {
+ splits.add(ChunkRange.of(chunkStart, chunkEnd));
+ chunkStart = chunkEnd;
+ try {
+ chunkEnd = ObjectUtils.plus(chunkEnd, dynamicChunkSize);
+ } catch (ArithmeticException e) {
+ // Stop chunk split to avoid dead loop when number overflows.
+ break;
+ }
+ }
+ // add the ending split
+ splits.add(ChunkRange.of(chunkStart, null));
+ return splits;
+ }
+
+ // ------------------------------------------------------------------------------------------
+
+ /** Split table into unevenly sized chunks by continuously calculating next chunk max value. */
+ private List<ChunkRange> splitUnevenlySizedChunks(
+ JdbcConnection jdbc,
+ TableId tableId,
+ String splitColumnName,
+ Object min,
+ Object max,
+ int chunkSize)
+ throws SQLException {
+ LOG.info(
+ "Use unevenly-sized chunks for table {}, the chunk size is {}", tableId, chunkSize);
+ final List<ChunkRange> splits = new ArrayList<>();
+ Object chunkStart = null;
+ Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumnName, max, chunkSize);
+ int count = 0;
+ while (chunkEnd != null && ObjectUtils.compare(chunkEnd, max) <= 0) {
+ // we start from [null, min + chunk_size) and avoid [null, min)
+ splits.add(ChunkRange.of(chunkStart, chunkEnd));
+ // may sleep awhile to avoid DDOS on Db2 server
+ maySleep(count++, tableId);
+ chunkStart = chunkEnd;
+ chunkEnd = nextChunkEnd(jdbc, chunkEnd, tableId, splitColumnName, max, chunkSize);
+ }
+ // add the ending split
+ splits.add(ChunkRange.of(chunkStart, null));
+ return splits;
+ }
+
+ private Object nextChunkEnd(
+ JdbcConnection jdbc,
+ Object previousChunkEnd,
+ TableId tableId,
+ String splitColumnName,
+ Object max,
+ int chunkSize)
+ throws SQLException {
+ // chunk end might be null when max values are removed
+ Object chunkEnd =
+ queryNextChunkMax(jdbc, tableId, splitColumnName, chunkSize, previousChunkEnd);
+ if (Objects.equals(previousChunkEnd, chunkEnd)) {
+ // we don't allow equal chunk start and end,
+ // should query the next one larger than chunkEnd
+ chunkEnd = queryMin(jdbc, tableId, splitColumnName, chunkEnd);
+ }
+ if (ObjectUtils.compare(chunkEnd, max) >= 0) {
+ return null;
+ } else {
+ return chunkEnd;
+ }
+ }
+
+ private SnapshotSplit createSnapshotSplit(
+ JdbcConnection jdbc,
+ TableId tableId,
+ int chunkId,
+ RowType splitKeyType,
+ Object chunkStart,
+ Object chunkEnd) {
+ // currently, we only support single split column
+ Object[] splitStart = chunkStart == null ? null : new Object[] {chunkStart};
+ Object[] splitEnd = chunkEnd == null ? null : new Object[] {chunkEnd};
+ Map<TableId, TableChange> schema = new HashMap<>();
+ schema.put(tableId, dialect.queryTableSchema(jdbc, tableId));
+ return new SnapshotSplit(
+ tableId,
+ splitId(tableId, chunkId),
+ splitKeyType,
+ splitStart,
+ splitEnd,
+ null,
+ schema);
+ }
+
+ /** Returns the distribution factor of the given table. */
+ private double calculateDistributionFactor(
+ TableId tableId, Object min, Object max, long approximateRowCnt) {
+
+ if (!min.getClass().equals(max.getClass())) {
+ throw new IllegalStateException(
+ String.format(
+ "Unsupported operation type, the MIN value type %s is different with MAX value type %s.",
+ min.getClass().getSimpleName(), max.getClass().getSimpleName()));
+ }
+ if (approximateRowCnt == 0) {
+ return Double.MAX_VALUE;
+ }
+ BigDecimal difference = ObjectUtils.minus(max, min);
+ // factor = (max - min + 1) / rowCount
+ final BigDecimal subRowCnt = difference.add(BigDecimal.valueOf(1));
+ double distributionFactor =
+ subRowCnt
+ .divide(new BigDecimal(approximateRowCnt), 4, RoundingMode.CEILING)
+ .doubleValue();
+ LOG.info(
+ "The distribution factor of table {} is {} according to the min split key {}, max split key {} and approximate row count {}",
+ tableId,
+ distributionFactor,
+ min,
+ max,
+ approximateRowCnt);
+ return distributionFactor;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2Dialect.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2Dialect.java
new file mode 100644
index 000000000..7de4265fb
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2Dialect.java
@@ -0,0 +1,175 @@
+/*
+ * 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.connectors.db2.source.dialect;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
+import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect;
+import org.apache.flink.cdc.connectors.base.relational.connection.JdbcConnectionPoolFactory;
+import org.apache.flink.cdc.connectors.base.source.assigner.splitter.ChunkSplitter;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask.Context;
+import org.apache.flink.cdc.connectors.db2.source.config.Db2SourceConfig;
+import org.apache.flink.cdc.connectors.db2.source.fetch.Db2ScanFetchTask;
+import org.apache.flink.cdc.connectors.db2.source.fetch.Db2SourceFetchTaskContext;
+import org.apache.flink.cdc.connectors.db2.source.fetch.Db2StreamFetchTask;
+import org.apache.flink.cdc.connectors.db2.source.utils.Db2ConnectionUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.TableId;
+import io.debezium.relational.Tables;
+import io.debezium.relational.history.TableChanges.TableChange;
+
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.cdc.connectors.db2.source.utils.Db2ConnectionUtils.createDb2Connection;
+import static org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils.currentLsn;
+
+/** The {@link JdbcDataSourceDialect} implementation for Db2 datasource. */
+@Experimental
+public class Db2Dialect implements JdbcDataSourceDialect {
+
+ private static final long serialVersionUID = 1L;
+ private final Db2SourceConfig sourceConfig;
+ private transient Db2Schema db2Schema;
+ private transient Tables.TableFilter filters;
+
+ public Db2Dialect(Db2SourceConfig sourceConfig) {
+ this.sourceConfig = sourceConfig;
+ }
+
+ @Override
+ public String getName() {
+ return "Db2";
+ }
+
+ @Override
+ public Offset displayCurrentOffset(JdbcSourceConfig sourceConfig) {
+ try (JdbcConnection jdbcConnection = openJdbcConnection(sourceConfig)) {
+ return currentLsn((Db2Connection) jdbcConnection);
+ } catch (Exception e) {
+ throw new FlinkRuntimeException("Read the redoLog offset error", e);
+ }
+ }
+
+ @Override
+ public boolean isDataCollectionIdCaseSensitive(JdbcSourceConfig sourceConfig) {
+ return true;
+ }
+
+ @Override
+ public JdbcConnection openJdbcConnection(JdbcSourceConfig sourceConfig) {
+ return createDb2Connection(sourceConfig.getDbzConnectorConfig());
+ }
+
+ @Override
+ public ChunkSplitter createChunkSplitter(JdbcSourceConfig sourceConfig) {
+ return new Db2ChunkSplitter(sourceConfig, this);
+ }
+
+ @Override
+ public JdbcConnectionPoolFactory getPooledDataSourceFactory() {
+ return new Db2PooledDataSourceFactory();
+ }
+
+ @Override
+ public List<TableId> discoverDataCollections(JdbcSourceConfig sourceConfig) {
+ Db2SourceConfig db2SourceConfig = (Db2SourceConfig) sourceConfig;
+ try (JdbcConnection jdbcConnection = openJdbcConnection(sourceConfig)) {
+ return Db2ConnectionUtils.listTables(
+ jdbcConnection, db2SourceConfig.getDbzConnectorConfig().getTableFilters());
+ } catch (SQLException e) {
+ throw new FlinkRuntimeException("Error to discover tables: " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public Map<TableId, TableChange> discoverDataCollectionSchemas(JdbcSourceConfig sourceConfig) {
+ final List<TableId> capturedTableIds = discoverDataCollections(sourceConfig);
+
+ try (Db2Connection jdbc = createDb2Connection(sourceConfig.getDbzConnectorConfig())) {
+ // fetch table schemas
+ Map<TableId, TableChange> tableSchemas = new HashMap<>();
+ for (TableId tableId : capturedTableIds) {
+ TableChange tableSchema = queryTableSchema(jdbc, tableId);
+ tableSchemas.put(tableId, tableSchema);
+ }
+ return tableSchemas;
+ } catch (Exception e) {
+ throw new FlinkRuntimeException(
+ "Error to discover table schemas: " + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public TableChange queryTableSchema(JdbcConnection jdbc, TableId tableId) {
+ if (db2Schema == null) {
+ db2Schema = new Db2Schema();
+ }
+ return db2Schema.getTableSchema(
+ jdbc,
+ tableId,
+ sourceConfig.getDbzConnectorConfig().getTableFilters().dataCollectionFilter());
+ }
+
+ @Override
+ public FetchTask<SourceSplitBase> createFetchTask(SourceSplitBase sourceSplitBase) {
+ if (sourceSplitBase.isSnapshotSplit()) {
+ return new Db2ScanFetchTask(sourceSplitBase.asSnapshotSplit());
+ } else {
+ return new Db2StreamFetchTask(sourceSplitBase.asStreamSplit());
+ }
+ }
+
+ /**
+ * The task context used for fetch task to fetch data from external systems.
+ *
+ * @param sourceConfig
+ */
+ @Override
+ public Context createFetchTaskContext(JdbcSourceConfig sourceConfig) {
+ final Db2Connection jdbcConnection =
+ createDb2Connection(sourceConfig.getDbzConnectorConfig());
+ final Db2Connection metaDataConnection =
+ createDb2Connection(sourceConfig.getDbzConnectorConfig());
+ return new Db2SourceFetchTaskContext(
+ sourceConfig, this, jdbcConnection, metaDataConnection);
+ }
+
+ /**
+ * Check if the tableId is included in SourceConfig.
+ *
+ * @param sourceConfig
+ * @param tableId
+ */
+ @Override
+ public boolean isIncludeDataCollection(JdbcSourceConfig sourceConfig, TableId tableId) {
+ if (filters == null) {
+ this.filters = sourceConfig.getTableFilters().dataCollectionFilter();
+ }
+
+ return filters.isIncluded(tableId);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupMode.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2PooledDataSourceFactory.java
similarity index 51%
rename from flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupMode.java
rename to flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2PooledDataSourceFactory.java
index 43b97c3d1..2647fcff2 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupMode.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2PooledDataSourceFactory.java
@@ -15,14 +15,21 @@
* limitations under the License.
*/
-package org.apache.flink.cdc.connectors.db2.table;
+package org.apache.flink.cdc.connectors.db2.source.dialect;
-/**
- * Startup modes for the Db2 CDC Consumer.
- *
- * @see StartupOptions
- */
-public enum StartupMode {
- INITIAL,
- LATEST_OFFSET
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
+import org.apache.flink.cdc.connectors.base.relational.connection.JdbcConnectionPoolFactory;
+
+/** Factory to create {@link JdbcConnectionPoolFactory} for SQL Server. */
+public class Db2PooledDataSourceFactory extends JdbcConnectionPoolFactory {
+
+ private static final String URL_PATTERN = "jdbc:db2://%s:%s/%s";
+
+ @Override
+ public String getJdbcUrl(JdbcSourceConfig sourceConfig) {
+ String hostName = sourceConfig.getHostname();
+ int port = sourceConfig.getPort();
+ String database = sourceConfig.getDatabaseList().get(0);
+ return String.format(URL_PATTERN, hostName, port, database);
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2Schema.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2Schema.java
new file mode 100644
index 000000000..a8dc79d81
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/dialect/Db2Schema.java
@@ -0,0 +1,80 @@
+/*
+ * 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.connectors.db2.source.dialect;
+
+import org.apache.flink.util.FlinkRuntimeException;
+
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.relational.Tables;
+import io.debezium.relational.history.TableChanges;
+import io.debezium.relational.history.TableChanges.TableChange;
+
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** A component used to get schema by table path. */
+public class Db2Schema {
+
+ private final Map<TableId, TableChange> schemasByTableId;
+
+ public Db2Schema() {
+ this.schemasByTableId = new ConcurrentHashMap<>();
+ }
+
+ public TableChange getTableSchema(
+ JdbcConnection jdbc, TableId tableId, Tables.TableFilter tableFilters) {
+ // read schema from cache first
+ TableChange schema = schemasByTableId.get(tableId);
+ if (schema == null) {
+ schema = readTableSchema(jdbc, tableId, tableFilters);
+ schemasByTableId.put(tableId, schema);
+ }
+ return schema;
+ }
+
+ private TableChange readTableSchema(
+ JdbcConnection jdbc, TableId tableId, Tables.TableFilter tableFilters) {
+ Db2Connection db2Connection = (Db2Connection) jdbc;
+
+ final Map<TableId, TableChange> tableChangeMap = new HashMap<>();
+ Tables tables = new Tables();
+ tables.overwriteTable(tables.editOrCreateTable(tableId).create());
+
+ try {
+ db2Connection.readSchema(tables, null, tableId.schema(), tableFilters, null, false);
+ Table table = tables.forTable(tableId);
+ TableChange tableChange = new TableChange(TableChanges.TableChangeType.CREATE, table);
+ tableChangeMap.put(tableId, tableChange);
+ } catch (SQLException e) {
+ throw new FlinkRuntimeException(
+ String.format("Failed to read schema for table %s ", tableId), e);
+ }
+
+ if (!tableChangeMap.containsKey(tableId)) {
+ throw new FlinkRuntimeException(
+ String.format("Can't obtain schema for table %s ", tableId));
+ }
+
+ return tableChangeMap.get(tableId);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java
new file mode 100644
index 000000000..dba9fe84a
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2ScanFetchTask.java
@@ -0,0 +1,348 @@
+/*
+ * 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.connectors.db2.source.fetch;
+
+import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+import org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask;
+import org.apache.flink.cdc.connectors.db2.source.fetch.Db2StreamFetchTask.StreamSplitReadTask;
+
+import io.debezium.DebeziumException;
+import io.debezium.config.Configuration;
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.connector.db2.Db2ConnectorConfig;
+import io.debezium.connector.db2.Db2DatabaseSchema;
+import io.debezium.connector.db2.Db2OffsetContext;
+import io.debezium.connector.db2.Db2Partition;
+import io.debezium.heartbeat.Heartbeat;
+import io.debezium.pipeline.EventDispatcher;
+import io.debezium.pipeline.source.AbstractSnapshotChangeEventSource;
+import io.debezium.pipeline.source.spi.ChangeEventSource;
+import io.debezium.pipeline.source.spi.SnapshotProgressListener;
+import io.debezium.pipeline.spi.ChangeRecordEmitter;
+import io.debezium.pipeline.spi.SnapshotResult;
+import io.debezium.relational.RelationalSnapshotChangeEventSource;
+import io.debezium.relational.SnapshotChangeRecordEmitter;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.util.Clock;
+import io.debezium.util.ColumnUtils;
+import io.debezium.util.Strings;
+import io.debezium.util.Threads;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Duration;
+
+import static org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils.buildSplitScanQuery;
+import static org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils.readTableSplitDataStatement;
+
+/** The task to work for fetching data of Db2 table snapshot split. */
+public class Db2ScanFetchTask extends AbstractScanFetchTask {
+
+ public Db2ScanFetchTask(SnapshotSplit split) {
+ super(split);
+ }
+
+ @Override
+ protected void executeDataSnapshot(Context context) throws Exception {
+ Db2SourceFetchTaskContext sourceFetchContext = (Db2SourceFetchTaskContext) context;
+ Db2SnapshotSplitReadTask snapshotSplitReadTask =
+ new Db2SnapshotSplitReadTask(
+ sourceFetchContext.getDbzConnectorConfig(),
+ sourceFetchContext.getOffsetContext(),
+ sourceFetchContext.getSnapshotChangeEventSourceMetrics(),
+ sourceFetchContext.getDatabaseSchema(),
+ sourceFetchContext.getConnection(),
+ sourceFetchContext.getDispatcher(),
+ sourceFetchContext.getSnapshotReceiver(),
+ snapshotSplit);
+ Db2SnapshotSplitChangeEventSourceContext changeEventSourceContext =
+ new Db2SnapshotSplitChangeEventSourceContext();
+ SnapshotResult<Db2OffsetContext> snapshotResult =
+ snapshotSplitReadTask.execute(
+ changeEventSourceContext,
+ sourceFetchContext.getPartition(),
+ sourceFetchContext.getOffsetContext());
+ // execute stream read task
+ if (!snapshotResult.isCompletedOrSkipped()) {
+ taskRunning = false;
+ throw new IllegalStateException(
+ String.format("Read snapshot for Db2 split %s fail", snapshotSplit));
+ }
+ }
+
+ @Override
+ protected void executeBackfillTask(Context context, StreamSplit backfillStreamSplit)
+ throws Exception {
+ Db2SourceFetchTaskContext sourceFetchContext = (Db2SourceFetchTaskContext) context;
+ final Db2OffsetContext.Loader loader =
+ new Db2OffsetContext.Loader(sourceFetchContext.getDbzConnectorConfig());
+ final Db2OffsetContext streamOffsetContext =
+ loader.load(backfillStreamSplit.getStartingOffset().getOffset());
+
+ final StreamSplitReadTask backfillBinlogReadTask =
+ createBackFillLsnSplitReadTask(backfillStreamSplit, sourceFetchContext);
+ backfillBinlogReadTask.execute(
+ new Db2SnapshotSplitChangeEventSourceContext(),
+ sourceFetchContext.getPartition(),
+ streamOffsetContext);
+ }
+
+ private StreamSplitReadTask createBackFillLsnSplitReadTask(
+ StreamSplit backfillBinlogSplit, Db2SourceFetchTaskContext context) {
+ // we should only capture events for the current table,
+ // otherwise, we may can't find corresponding schema
+ Configuration dezConf =
+ context.getDbzConnectorConfig()
+ .getConfig()
+ .edit()
+ // table.include.list is schema.table format
+ .with(
+ "table.include.list",
+ new TableId(
+ null,
+ snapshotSplit.getTableId().schema(),
+ snapshotSplit.getTableId().table()))
+ // Disable heartbeat event in snapshot split fetcher
+ .with(Heartbeat.HEARTBEAT_INTERVAL, 0)
+ .build();
+ // task to read wal and backfill for current split
+ return new StreamSplitReadTask(
+ new Db2ConnectorConfig(dezConf),
+ context.getConnection(),
+ context.getMetaDataConnection(),
+ context.getDispatcher(),
+ context.getErrorHandler(),
+ context.getDatabaseSchema(),
+ backfillBinlogSplit);
+ }
+
+ /** A wrapped task to fetch snapshot split of table. */
+ public static class Db2SnapshotSplitReadTask
+ extends AbstractSnapshotChangeEventSource<Db2Partition, Db2OffsetContext> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Db2SnapshotSplitReadTask.class);
+
+ /** Interval for showing a log statement with the progress while scanning a single table. */
+ private static final Duration LOG_INTERVAL = Duration.ofMillis(10_000);
+
+ private final Db2ConnectorConfig connectorConfig;
+ private final Db2DatabaseSchema databaseSchema;
+ private final Db2Connection jdbcConnection;
+ private final JdbcSourceEventDispatcher<Db2Partition> dispatcher;
+ private final Clock clock;
+ private final SnapshotSplit snapshotSplit;
+ private final Db2OffsetContext offsetContext;
+ private final SnapshotProgressListener<Db2Partition> snapshotProgressListener;
+ private final EventDispatcher.SnapshotReceiver<Db2Partition> snapshotReceiver;
+
+ public Db2SnapshotSplitReadTask(
+ Db2ConnectorConfig connectorConfig,
+ Db2OffsetContext previousOffset,
+ SnapshotProgressListener<Db2Partition> snapshotProgressListener,
+ Db2DatabaseSchema databaseSchema,
+ Db2Connection jdbcConnection,
+ JdbcSourceEventDispatcher<Db2Partition> dispatcher,
+ EventDispatcher.SnapshotReceiver<Db2Partition> snapshotReceiver,
+ SnapshotSplit snapshotSplit) {
+ super(connectorConfig, snapshotProgressListener);
+ this.offsetContext = previousOffset;
+ this.connectorConfig = connectorConfig;
+ this.databaseSchema = databaseSchema;
+ this.jdbcConnection = jdbcConnection;
+ this.dispatcher = dispatcher;
+ this.clock = Clock.SYSTEM;
+ this.snapshotSplit = snapshotSplit;
+ this.snapshotProgressListener = snapshotProgressListener;
+ this.snapshotReceiver = snapshotReceiver;
+ }
+
+ @Override
+ public SnapshotResult<Db2OffsetContext> execute(
+ ChangeEventSourceContext context,
+ Db2Partition partition,
+ Db2OffsetContext previousOffset)
+ throws InterruptedException {
+ SnapshottingTask snapshottingTask = getSnapshottingTask(partition, previousOffset);
+ final Db2SnapshotContext ctx;
+ try {
+ ctx = prepare(partition);
+ } catch (Exception e) {
+ LOG.error("Failed to initialize snapshot context.", e);
+ throw new RuntimeException(e);
+ }
+ try {
+ return doExecute(context, previousOffset, ctx, snapshottingTask);
+ } catch (InterruptedException e) {
+ LOG.warn("Snapshot was interrupted before completion");
+ throw e;
+ } catch (Exception t) {
+ throw new DebeziumException(t);
+ }
+ }
+
+ @Override
+ protected SnapshotResult<Db2OffsetContext> doExecute(
+ ChangeEventSourceContext context,
+ Db2OffsetContext previousOffset,
+ SnapshotContext<Db2Partition, Db2OffsetContext> snapshotContext,
+ SnapshottingTask snapshottingTask)
+ throws Exception {
+ final Db2SnapshotContext ctx = (Db2SnapshotContext) snapshotContext;
+ ctx.offset = offsetContext;
+
+ createDataEvents(ctx, snapshotSplit.getTableId());
+
+ return SnapshotResult.completed(ctx.offset);
+ }
+
+ @Override
+ protected SnapshottingTask getSnapshottingTask(
+ Db2Partition partition, Db2OffsetContext previousOffset) {
+ return new SnapshottingTask(false, true);
+ }
+
+ @Override
+ protected Db2SnapshotContext prepare(Db2Partition partition) throws Exception {
+ return new Db2SnapshotContext(partition);
+ }
+
+ private void createDataEvents(Db2SnapshotContext snapshotContext, TableId tableId)
+ throws Exception {
+ LOG.debug("Snapshotting table {}", tableId);
+ createDataEventsForTable(
+ snapshotContext, snapshotReceiver, databaseSchema.tableFor(tableId));
+ snapshotReceiver.completeSnapshot();
+ }
+
+ /** Dispatches the data change events for the records of a single table. */
+ private void createDataEventsForTable(
+ Db2SnapshotContext snapshotContext,
+ EventDispatcher.SnapshotReceiver<Db2Partition> snapshotReceiver,
+ Table table)
+ throws InterruptedException {
+
+ long exportStart = clock.currentTimeInMillis();
+ LOG.info(
+ "Exporting data from split '{}' of table {}",
+ snapshotSplit.splitId(),
+ table.id());
+
+ final String selectSql =
+ buildSplitScanQuery(
+ snapshotSplit.getTableId(),
+ snapshotSplit.getSplitKeyType(),
+ snapshotSplit.getSplitStart() == null,
+ snapshotSplit.getSplitEnd() == null);
+ LOG.info(
+ "For split '{}' of table {} using select statement: '{}'",
+ snapshotSplit.splitId(),
+ table.id(),
+ selectSql);
+
+ try (PreparedStatement selectStatement =
+ readTableSplitDataStatement(
+ jdbcConnection,
+ selectSql,
+ snapshotSplit.getSplitStart() == null,
+ snapshotSplit.getSplitEnd() == null,
+ snapshotSplit.getSplitStart(),
+ snapshotSplit.getSplitEnd(),
+ snapshotSplit.getSplitKeyType().getFieldCount(),
+ connectorConfig.getQueryFetchSize());
+ ResultSet rs = selectStatement.executeQuery()) {
+
+ ColumnUtils.ColumnArray columnArray = ColumnUtils.toArray(rs, table);
+ long rows = 0;
+ Threads.Timer logTimer = getTableScanLogTimer();
+
+ while (rs.next()) {
+ rows++;
+ final Object[] row =
+ jdbcConnection.rowToArray(table, databaseSchema, rs, columnArray);
+ if (logTimer.expired()) {
+ long stop = clock.currentTimeInMillis();
+ LOG.info(
+ "Exported {} records for split '{}' after {}",
+ rows,
+ snapshotSplit.splitId(),
+ Strings.duration(stop - exportStart));
+ snapshotProgressListener.rowsScanned(
+ snapshotContext.partition, table.id(), rows);
+ logTimer = getTableScanLogTimer();
+ }
+ dispatcher.dispatchSnapshotEvent(
+ snapshotContext.partition,
+ table.id(),
+ getChangeRecordEmitter(snapshotContext, table.id(), row),
+ snapshotReceiver);
+ }
+ LOG.info(
+ "Finished exporting {} records for split '{}', total duration '{}'",
+ rows,
+ snapshotSplit.splitId(),
+ Strings.duration(clock.currentTimeInMillis() - exportStart));
+ } catch (SQLException e) {
+ throw new ConnectException("Snapshotting of table " + table.id() + " failed", e);
+ }
+ }
+
+ protected ChangeRecordEmitter<Db2Partition> getChangeRecordEmitter(
+ Db2SnapshotContext snapshotContext, TableId tableId, Object[] row) {
+ snapshotContext.offset.event(tableId, clock.currentTime());
+ return new SnapshotChangeRecordEmitter<>(
+ snapshotContext.partition, snapshotContext.offset, row, clock);
+ }
+
+ private Threads.Timer getTableScanLogTimer() {
+ return Threads.timer(clock, LOG_INTERVAL);
+ }
+
+ private static class Db2SnapshotContext
+ extends RelationalSnapshotChangeEventSource.RelationalSnapshotContext<
+ Db2Partition, Db2OffsetContext> {
+
+ public Db2SnapshotContext(Db2Partition partition) throws SQLException {
+ super(partition, "");
+ }
+ }
+ }
+
+ /**
+ * The {@link ChangeEventSource.ChangeEventSourceContext} implementation for bounded stream task
+ * of a snapshot split task.
+ */
+ public class Db2SnapshotSplitChangeEventSourceContext
+ implements ChangeEventSource.ChangeEventSourceContext {
+
+ public void finished() {
+ taskRunning = false;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return taskRunning;
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java
new file mode 100644
index 000000000..45bf132ea
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2SourceFetchTaskContext.java
@@ -0,0 +1,311 @@
+/*
+ * 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.connectors.db2.source.fetch;
+
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
+import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher;
+import org.apache.flink.cdc.connectors.base.source.EmbeddedFlinkDatabaseHistory;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+import org.apache.flink.cdc.connectors.base.source.reader.external.JdbcSourceFetchTaskContext;
+import org.apache.flink.cdc.connectors.db2.source.config.Db2SourceConfig;
+import org.apache.flink.cdc.connectors.db2.source.dialect.Db2Dialect;
+import org.apache.flink.cdc.connectors.db2.source.handler.Db2SchemaChangeEventHandler;
+import org.apache.flink.cdc.connectors.db2.source.offset.LsnOffset;
+import org.apache.flink.cdc.connectors.db2.source.utils.Db2Utils;
+import org.apache.flink.table.types.logical.RowType;
+
+import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.connector.base.ChangeEventQueue.Builder;
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.connector.db2.Db2Connector;
+import io.debezium.connector.db2.Db2ConnectorConfig;
+import io.debezium.connector.db2.Db2DatabaseSchema;
+import io.debezium.connector.db2.Db2OffsetContext;
+import io.debezium.connector.db2.Db2OffsetContext.Loader;
+import io.debezium.connector.db2.Db2Partition;
+import io.debezium.connector.db2.Db2TaskContext;
+import io.debezium.connector.db2.Db2TopicSelector;
+import io.debezium.connector.db2.SourceInfo;
+import io.debezium.data.Envelope.FieldName;
+import io.debezium.pipeline.DataChangeEvent;
+import io.debezium.pipeline.ErrorHandler;
+import io.debezium.pipeline.EventDispatcher;
+import io.debezium.pipeline.metrics.DefaultChangeEventSourceMetricsFactory;
+import io.debezium.pipeline.metrics.SnapshotChangeEventSourceMetrics;
+import io.debezium.pipeline.metrics.StreamingChangeEventSourceMetrics;
+import io.debezium.pipeline.source.spi.EventMetadataProvider;
+import io.debezium.pipeline.spi.OffsetContext;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.relational.Tables.TableFilter;
+import io.debezium.schema.DataCollectionId;
+import io.debezium.schema.TopicSelector;
+import io.debezium.util.Collect;
+import io.debezium.util.SchemaNameAdjuster;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import java.time.Instant;
+import java.util.Map;
+
+/** The context for fetch task that fetching data of snapshot split from Db2 data source. */
+public class Db2SourceFetchTaskContext extends JdbcSourceFetchTaskContext {
+
+ /** Connection used for reading CDC tables. */
+ private final Db2Connection connection;
+
+ private final Db2Connection metaDataConnection;
+
+ private final Db2EventMetadataProvider metadataProvider;
+ private Db2OffsetContext offsetContext;
+ private Db2Partition partition;
+ private Db2DatabaseSchema databaseSchema;
+ private JdbcSourceEventDispatcher<Db2Partition> dispatcher;
+ private ErrorHandler errorHandler;
+ private ChangeEventQueue<DataChangeEvent> queue;
+ private Db2TaskContext taskContext;
+ private TopicSelector<TableId> topicSelector;
+ private EventDispatcher.SnapshotReceiver<Db2Partition> snapshotReceiver;
+ private SnapshotChangeEventSourceMetrics<Db2Partition> snapshotChangeEventSourceMetrics;
+ private StreamingChangeEventSourceMetrics<Db2Partition> streamingChangeEventSourceMetrics;
+
+ public Db2SourceFetchTaskContext(
+ JdbcSourceConfig sourceConfig,
+ Db2Dialect dataSourceDialect,
+ Db2Connection connection,
+ Db2Connection metaDataConnection) {
+ super(sourceConfig, dataSourceDialect);
+ this.connection = connection;
+ this.metadataProvider = new Db2EventMetadataProvider();
+ this.metaDataConnection = metaDataConnection;
+ }
+
+ @Override
+ public void configure(SourceSplitBase sourceSplitBase) {
+ // initial stateful objects
+ final Db2ConnectorConfig connectorConfig = getDbzConnectorConfig();
+ this.topicSelector = Db2TopicSelector.defaultSelector(connectorConfig);
+ EmbeddedFlinkDatabaseHistory.registerHistory(
+ sourceConfig
+ .getDbzConfiguration()
+ .getString(EmbeddedFlinkDatabaseHistory.DATABASE_HISTORY_INSTANCE_NAME),
+ sourceSplitBase.getTableSchemas().values());
+
+ this.databaseSchema = Db2Utils.createDb2DatabaseSchema(connectorConfig, connection);
+ this.offsetContext = loadStartingOffsetState(new Loader(connectorConfig), sourceSplitBase);
+
+ String serverName = connectorConfig.getLogicalName();
+ this.partition = new Db2Partition(serverName);
+
+ validateAndLoadDatabaseHistory(offsetContext, databaseSchema);
+
+ this.taskContext = new Db2TaskContext(connectorConfig, databaseSchema);
+
+ final int queueSize =
+ sourceSplitBase.isSnapshotSplit()
+ ? getSourceConfig().getSplitSize()
+ : getSourceConfig().getDbzConnectorConfig().getMaxQueueSize();
+ this.queue =
+ new Builder<DataChangeEvent>()
+ .pollInterval(connectorConfig.getPollInterval())
+ .maxBatchSize(connectorConfig.getMaxBatchSize())
+ .maxQueueSize(queueSize)
+ .maxQueueSizeInBytes(connectorConfig.getMaxQueueSizeInBytes())
+ .loggingContextSupplier(
+ () -> taskContext.configureLoggingContext("Db2-cdc-connector-task"))
+ // do not buffer any element, we use signal event
+ // .buffering()
+ .build();
+ this.dispatcher =
+ new JdbcSourceEventDispatcher<>(
+ connectorConfig,
+ topicSelector,
+ databaseSchema,
+ queue,
+ connectorConfig.getTableFilters().dataCollectionFilter(),
+ DataChangeEvent::new,
+ metadataProvider,
+ schemaNameAdjuster,
+ new Db2SchemaChangeEventHandler());
+
+ this.snapshotReceiver = dispatcher.getSnapshotChangeEventReceiver();
+
+ final DefaultChangeEventSourceMetricsFactory<Db2Partition> changeEventSourceMetricsFactory =
+ new DefaultChangeEventSourceMetricsFactory<>();
+ this.snapshotChangeEventSourceMetrics =
+ changeEventSourceMetricsFactory.getSnapshotMetrics(
+ taskContext, queue, metadataProvider);
+ this.streamingChangeEventSourceMetrics =
+ changeEventSourceMetricsFactory.getStreamingMetrics(
+ taskContext, queue, metadataProvider);
+ this.errorHandler = new ErrorHandler(Db2Connector.class, connectorConfig, queue);
+ }
+
+ /** Loads the connector's persistent offset (if present) via the given loader. */
+ private Db2OffsetContext loadStartingOffsetState(
+ Db2OffsetContext.Loader loader, SourceSplitBase sourceSplitBase) {
+ Offset offset =
+ sourceSplitBase.isSnapshotSplit()
+ ? LsnOffset.INITIAL_OFFSET
+ : sourceSplitBase.asStreamSplit().getStartingOffset();
+
+ return loader.load(offset.getOffset());
+ }
+
+ private void validateAndLoadDatabaseHistory(Db2OffsetContext offset, Db2DatabaseSchema schema) {
+ schema.initializeStorage();
+ schema.recover(partition, offset);
+ }
+
+ @Override
+ public ChangeEventQueue<DataChangeEvent> getQueue() {
+ return queue;
+ }
+
+ @Override
+ public TableFilter getTableFilter() {
+ return getDbzConnectorConfig().getTableFilters().dataCollectionFilter();
+ }
+
+ @Override
+ public Offset getStreamOffset(SourceRecord record) {
+ return Db2Utils.getLsn(record);
+ }
+
+ @Override
+ public Db2DatabaseSchema getDatabaseSchema() {
+ return databaseSchema;
+ }
+
+ @Override
+ public RowType getSplitType(Table table) {
+ Column splitColumn = Db2Utils.getSplitColumn(table, sourceConfig.getChunkKeyColumn());
+ return Db2Utils.getSplitType(splitColumn);
+ }
+
+ @Override
+ public ErrorHandler getErrorHandler() {
+ return errorHandler;
+ }
+
+ @Override
+ public Db2ConnectorConfig getDbzConnectorConfig() {
+ return (Db2ConnectorConfig) super.getDbzConnectorConfig();
+ }
+
+ @Override
+ public Db2SourceConfig getSourceConfig() {
+ return (Db2SourceConfig) sourceConfig;
+ }
+
+ @Override
+ public JdbcSourceEventDispatcher<Db2Partition> getDispatcher() {
+ return dispatcher;
+ }
+
+ public EventDispatcher.SnapshotReceiver<Db2Partition> getSnapshotReceiver() {
+ return snapshotReceiver;
+ }
+
+ @Override
+ public Db2OffsetContext getOffsetContext() {
+ return offsetContext;
+ }
+
+ @Override
+ public Db2Partition getPartition() {
+ return partition;
+ }
+
+ public Db2Connection getConnection() {
+ return connection;
+ }
+
+ public Db2Connection getMetaDataConnection() {
+ return metaDataConnection;
+ }
+
+ public SnapshotChangeEventSourceMetrics<Db2Partition> getSnapshotChangeEventSourceMetrics() {
+ return snapshotChangeEventSourceMetrics;
+ }
+
+ public StreamingChangeEventSourceMetrics<Db2Partition> getStreamingChangeEventSourceMetrics() {
+ return streamingChangeEventSourceMetrics;
+ }
+
+ @Override
+ public void close() throws Exception {
+ metaDataConnection.commit();
+ connection.commit();
+ metaDataConnection.close();
+ connection.close();
+ }
+
+ @Override
+ public SchemaNameAdjuster getSchemaNameAdjuster() {
+ return schemaNameAdjuster;
+ }
+
+ /** Copied from debezium for accessing here. */
+ public static class Db2EventMetadataProvider implements EventMetadataProvider {
+
+ @Override
+ public Instant getEventTimestamp(
+ DataCollectionId source, OffsetContext offset, Object key, Struct value) {
+ if (value == null) {
+ return null;
+ }
+ final Struct sourceInfo = value.getStruct(FieldName.SOURCE);
+ if (source == null) {
+ return null;
+ }
+ final Long timestamp = sourceInfo.getInt64(SourceInfo.TIMESTAMP_KEY);
+ return timestamp == null ? null : Instant.ofEpochMilli(timestamp);
+ }
+
+ @Override
+ public Map<String, String> getEventSourcePosition(
+ DataCollectionId source, OffsetContext offset, Object key, Struct value) {
+ if (value == null) {
+ return null;
+ }
+ final Struct sourceInfo = value.getStruct(FieldName.SOURCE);
+ if (source == null) {
+ return null;
+ }
+ return Collect.hashMapOf(
+ SourceInfo.COMMIT_LSN_KEY, sourceInfo.getString(SourceInfo.COMMIT_LSN_KEY),
+ SourceInfo.CHANGE_LSN_KEY, sourceInfo.getString(SourceInfo.CHANGE_LSN_KEY));
+ }
+
+ @Override
+ public String getTransactionId(
+ DataCollectionId source, OffsetContext offset, Object key, Struct value) {
+ if (value == null) {
+ return null;
+ }
+ final Struct sourceInfo = value.getStruct(FieldName.SOURCE);
+ if (source == null) {
+ return null;
+ }
+ return sourceInfo.getString(SourceInfo.COMMIT_LSN_KEY);
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java
new file mode 100644
index 000000000..36d7fe03e
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/fetch/Db2StreamFetchTask.java
@@ -0,0 +1,173 @@
+/*
+ * 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.connectors.db2.source.fetch;
+
+import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+import org.apache.flink.cdc.connectors.base.source.meta.wartermark.WatermarkKind;
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.db2.source.offset.LsnOffset;
+
+import io.debezium.DebeziumException;
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.connector.db2.Db2ConnectorConfig;
+import io.debezium.connector.db2.Db2DatabaseSchema;
+import io.debezium.connector.db2.Db2OffsetContext;
+import io.debezium.connector.db2.Db2Partition;
+import io.debezium.connector.db2.Db2StreamingChangeEventSource;
+import io.debezium.connector.db2.Lsn;
+import io.debezium.pipeline.ErrorHandler;
+import io.debezium.pipeline.source.spi.ChangeEventSource.ChangeEventSourceContext;
+import io.debezium.util.Clock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.flink.cdc.connectors.db2.source.offset.LsnOffset.NO_STOPPING_OFFSET;
+
+/** The task to work for fetching data of Db2 table stream split . */
+public class Db2StreamFetchTask implements FetchTask<SourceSplitBase> {
+
+ private final StreamSplit split;
+ private volatile boolean taskRunning = false;
+
+ public Db2StreamFetchTask(StreamSplit split) {
+ this.split = split;
+ }
+
+ @Override
+ public void execute(Context context) throws Exception {
+ Db2SourceFetchTaskContext sourceFetchContext = (Db2SourceFetchTaskContext) context;
+ sourceFetchContext.getOffsetContext().preSnapshotCompletion();
+ taskRunning = true;
+ StreamSplitReadTask streamSplitReadTask =
+ new StreamSplitReadTask(
+ sourceFetchContext.getDbzConnectorConfig(),
+ sourceFetchContext.getConnection(),
+ sourceFetchContext.getMetaDataConnection(),
+ sourceFetchContext.getDispatcher(),
+ sourceFetchContext.getErrorHandler(),
+ sourceFetchContext.getDatabaseSchema(),
+ split);
+ RedoLogSplitChangeEventSourceContext changeEventSourceContext =
+ new RedoLogSplitChangeEventSourceContext();
+ streamSplitReadTask.execute(
+ changeEventSourceContext,
+ sourceFetchContext.getPartition(),
+ sourceFetchContext.getOffsetContext());
+ }
+
+ @Override
+ public boolean isRunning() {
+ return taskRunning;
+ }
+
+ @Override
+ public StreamSplit getSplit() {
+ return split;
+ }
+
+ @Override
+ public void close() {
+ taskRunning = false;
+ }
+
+ /**
+ * A wrapped task to read all redo logs for table and also supports read bounded (from
+ * lowWatermark to highWatermark) redo logs.
+ */
+ public static class StreamSplitReadTask extends Db2StreamingChangeEventSource {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StreamSplitReadTask.class);
+ private final StreamSplit lsnSplit;
+ private final JdbcSourceEventDispatcher<Db2Partition> dispatcher;
+ private final ErrorHandler errorHandler;
+ private ChangeEventSourceContext context;
+
+ public StreamSplitReadTask(
+ Db2ConnectorConfig connectorConfig,
+ Db2Connection connection,
+ Db2Connection metadataConnection,
+ JdbcSourceEventDispatcher<Db2Partition> dispatcher,
+ ErrorHandler errorHandler,
+ Db2DatabaseSchema schema,
+ StreamSplit lsnSplit) {
+ super(
+ connectorConfig,
+ connection,
+ metadataConnection,
+ dispatcher,
+ errorHandler,
+ Clock.system(),
+ schema);
+ this.lsnSplit = lsnSplit;
+ this.dispatcher = dispatcher;
+ this.errorHandler = errorHandler;
+ }
+
+ @Override
+ public void afterHandleLsn(Db2Partition partition, Lsn toLsn) {
+ // check do we need to stop for fetch redo logs for snapshot split.
+ if (isBoundedRead()) {
+ LsnOffset currentLsnOffset = new LsnOffset(null, toLsn, null);
+ Offset endingOffset = lsnSplit.getEndingOffset();
+ if (currentLsnOffset.isAtOrAfter(endingOffset)) {
+ // send streaming end event
+ try {
+ dispatcher.dispatchWatermarkEvent(
+ partition.getSourcePartition(),
+ lsnSplit,
+ currentLsnOffset,
+ WatermarkKind.END);
+ } catch (InterruptedException e) {
+ LOG.error("Send signal event error.", e);
+ errorHandler.setProducerThrowable(
+ new DebeziumException(
+ "Error processing redo logs signal event", e));
+ }
+ // tell fetcher the streaming task finished
+ ((Db2ScanFetchTask.Db2SnapshotSplitChangeEventSourceContext) context)
+ .finished();
+ }
+ }
+ }
+
+ private boolean isBoundedRead() {
+ return !NO_STOPPING_OFFSET.equals(lsnSplit.getEndingOffset());
+ }
+
+ @Override
+ public void execute(
+ ChangeEventSourceContext context,
+ Db2Partition partition,
+ Db2OffsetContext offsetContext)
+ throws InterruptedException {
+ this.context = context;
+ super.execute(context, partition, offsetContext);
+ }
+ }
+
+ /** The {@link ChangeEventSourceContext} implementation for redo logs split task. */
+ private class RedoLogSplitChangeEventSourceContext implements ChangeEventSourceContext {
+ @Override
+ public boolean isRunning() {
+ return taskRunning;
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/handler/Db2SchemaChangeEventHandler.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/handler/Db2SchemaChangeEventHandler.java
new file mode 100644
index 000000000..968aac567
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/handler/Db2SchemaChangeEventHandler.java
@@ -0,0 +1,47 @@
+/*
+ * 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.connectors.db2.source.handler;
+
+import org.apache.flink.cdc.connectors.base.relational.handler.SchemaChangeEventHandler;
+
+import io.debezium.schema.SchemaChangeEvent;
+import org.apache.kafka.connect.data.Struct;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static io.debezium.connector.db2.SourceInfo.CHANGE_LSN_KEY;
+import static io.debezium.connector.db2.SourceInfo.COMMIT_LSN_KEY;
+
+/**
+ * This Db2SchemaChangeEventHandler helps to parse the source struct in SchemaChangeEvent and
+ * generate source info.
+ */
+public class Db2SchemaChangeEventHandler implements SchemaChangeEventHandler {
+
+ @Override
+ public Map<String, Object> parseSource(SchemaChangeEvent event) {
+ Map<String, Object> source = new HashMap<>();
+ Struct sourceInfo = event.getSource();
+ String changeLsn = sourceInfo.getString(CHANGE_LSN_KEY);
+ String commitLsn = sourceInfo.getString(COMMIT_LSN_KEY);
+ source.put(CHANGE_LSN_KEY, changeLsn);
+ source.put(COMMIT_LSN_KEY, commitLsn);
+ return source;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/offset/LsnFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/offset/LsnFactory.java
new file mode 100644
index 000000000..1c2944c80
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/offset/LsnFactory.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.cdc.connectors.db2.source.offset;
+
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+
+import io.debezium.connector.db2.Lsn;
+import io.debezium.connector.db2.SourceInfo;
+
+import java.util.Map;
+
+/** A factory to create {@link LsnOffset}. */
+public class LsnFactory extends OffsetFactory {
+ private static final String EVENT_SERIAL_NO_KEY = "event_serial_no";
+
+ @Override
+ public Offset newOffset(Map<String, String> offset) {
+ Lsn changeLsn = Lsn.valueOf(offset.get(SourceInfo.CHANGE_LSN_KEY));
+ Lsn commitLsn = Lsn.valueOf(offset.get(SourceInfo.COMMIT_LSN_KEY));
+ Long eventSerialNo = null;
+ if (offset.get(EVENT_SERIAL_NO_KEY) != null) {
+ eventSerialNo = Long.valueOf(offset.get(EVENT_SERIAL_NO_KEY));
+ }
+ return new LsnOffset(changeLsn, commitLsn, eventSerialNo);
+ }
+
+ @Override
+ public Offset newOffset(String filename, Long position) {
+ throw new UnsupportedOperationException(
+ "not supported create new Offset by filename and position.");
+ }
+
+ @Override
+ public Offset newOffset(Long position) {
+ throw new UnsupportedOperationException(
+ "not supported create new Offset by filename and position.");
+ }
+
+ @Override
+ public Offset createTimestampOffset(long timestampMillis) {
+ throw new UnsupportedOperationException("not supported create new Offset by timestamp.");
+ }
+
+ @Override
+ public Offset createInitialOffset() {
+ return LsnOffset.INITIAL_OFFSET;
+ }
+
+ @Override
+ public Offset createNoStoppingOffset() {
+ return LsnOffset.NO_STOPPING_OFFSET;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/offset/LsnOffset.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/offset/LsnOffset.java
new file mode 100644
index 000000000..72fffabf6
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/offset/LsnOffset.java
@@ -0,0 +1,95 @@
+/*
+ * 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.connectors.db2.source.offset;
+
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+
+import io.debezium.connector.db2.Lsn;
+import io.debezium.connector.db2.SourceInfo;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** A structure describes an offset in a Lsn event. */
+public class LsnOffset extends Offset {
+
+ public static final LsnOffset INITIAL_OFFSET =
+ new LsnOffset(Lsn.valueOf(new byte[] {Byte.MIN_VALUE}));
+ public static final LsnOffset NO_STOPPING_OFFSET =
+ new LsnOffset(Lsn.valueOf(new byte[] {Byte.MAX_VALUE}));
+ private static final String EVENT_SERIAL_NO_KEY = "event_serial_no";
+
+ public LsnOffset(Lsn changeLsn, Lsn commitLsn, Long eventSerialNo) {
+ Map<String, String> offsetMap = new HashMap<>();
+
+ if (changeLsn != null && changeLsn.isAvailable()) {
+ offsetMap.put(SourceInfo.CHANGE_LSN_KEY, changeLsn.toString());
+ }
+ if (commitLsn != null && commitLsn.isAvailable()) {
+ offsetMap.put(SourceInfo.COMMIT_LSN_KEY, commitLsn.toString());
+ }
+ if (eventSerialNo != null) {
+ offsetMap.put(EVENT_SERIAL_NO_KEY, String.valueOf(eventSerialNo));
+ }
+
+ this.offset = offsetMap;
+ }
+
+ public LsnOffset(Lsn changeLsn) {
+ this(changeLsn, null, null);
+ }
+
+ @Override
+ public int compareTo(Offset offset) {
+ LsnOffset that = (LsnOffset) offset;
+ // the NO_STOPPING_OFFSET is the max offset
+ if (NO_STOPPING_OFFSET.equals(that) && NO_STOPPING_OFFSET.equals(this)) {
+ return 0;
+ }
+ if (NO_STOPPING_OFFSET.equals(this)) {
+ return 1;
+ }
+ if (NO_STOPPING_OFFSET.equals(that)) {
+ return -1;
+ }
+
+ Lsn lsn = Lsn.valueOf(this.offset.get(SourceInfo.COMMIT_LSN_KEY));
+ Lsn targetLsn = Lsn.valueOf(that.offset.get(SourceInfo.COMMIT_LSN_KEY));
+ if (targetLsn.isAvailable()) {
+ if (lsn.isAvailable()) {
+ return lsn.compareTo(targetLsn);
+ }
+ return -1;
+ } else if (lsn.isAvailable()) {
+ return 1;
+ }
+ return 0;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof LsnOffset)) {
+ return false;
+ }
+ LsnOffset that = (LsnOffset) o;
+ return offset.equals(that.offset);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java
new file mode 100644
index 000000000..58bffa63c
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2ConnectionUtils.java
@@ -0,0 +1,83 @@
+/*
+ * 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.connectors.db2.source.utils;
+
+import io.debezium.config.Configuration;
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.jdbc.JdbcConfiguration;
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.RelationalDatabaseConnectorConfig;
+import io.debezium.relational.RelationalTableFilters;
+import io.debezium.relational.TableId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/** Utils for Db2 connection. */
+public class Db2ConnectionUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Db2ConnectionUtils.class);
+
+ public static Db2Connection createDb2Connection(
+ RelationalDatabaseConnectorConfig connectorConfig) {
+ Configuration dbzConnectorConfig = connectorConfig.getJdbcConfig();
+ return new Db2Connection(JdbcConfiguration.adapt(dbzConnectorConfig));
+ }
+
+ public static List<TableId> listTables(JdbcConnection jdbc, RelationalTableFilters tableFilters)
+ throws SQLException {
+ final List<TableId> capturedTableIds = new ArrayList<>();
+ Set<TableId> tableIdSet = new HashSet<>();
+ String queryTablesSql = "SELECT TABSCHEMA, TABNAME FROM SYSCAT.TABLES WHERE TYPE = 'T'";
+ Db2Connection db2Connection = (Db2Connection) jdbc;
+ try {
+ jdbc.query(
+ queryTablesSql,
+ rs -> {
+ while (rs.next()) {
+ String schemaName = rs.getString(1);
+ String tableName = rs.getString(2);
+ TableId tableId =
+ new TableId(
+ db2Connection.getRealDatabaseName(),
+ schemaName,
+ tableName);
+ tableIdSet.add(tableId);
+ }
+ });
+ } catch (SQLException e) {
+ LOG.warn(" SQL execute error, sql:{}", queryTablesSql, e);
+ }
+
+ for (TableId tableId : tableIdSet) {
+ if (tableFilters.dataCollectionFilter().isIncluded(tableId)) {
+ capturedTableIds.add(tableId);
+ LOG.info("\t including '{}' for further processing", tableId);
+ } else {
+ LOG.debug("\t '{}' is filtered out of capturing", tableId);
+ }
+ }
+
+ return capturedTableIds;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2TypeUtils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2TypeUtils.java
new file mode 100644
index 000000000..2c0f6a104
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2TypeUtils.java
@@ -0,0 +1,83 @@
+/*
+ * 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.connectors.db2.source.utils;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.types.DataType;
+
+import io.debezium.relational.Column;
+
+import java.sql.Types;
+
+/** Utilities for converting from Db2 types to Flink SQL types. */
+public class Db2TypeUtils {
+
+ /** Returns a corresponding Flink data type from a debezium {@link Column}. */
+ public static DataType fromDbzColumn(Column column) {
+ DataType dataType = convertFromColumn(column);
+ if (column.isOptional()) {
+ return dataType;
+ } else {
+ return dataType.notNull();
+ }
+ }
+
+ /**
+ * Returns a corresponding Flink data type from a debezium {@link Column} with nullable always
+ * be true.copy from io.debezium.connector.db2.Db2ValueConverters#converte.
+ */
+ private static DataType convertFromColumn(Column column) {
+ switch (column.jdbcType()) {
+ case Types.CHAR:
+ case Types.VARCHAR:
+ case Types.SQLXML:
+ case Types.CLOB:
+ return DataTypes.STRING();
+ case Types.BLOB:
+ case Types.BINARY:
+ case Types.VARBINARY:
+ return DataTypes.BYTES();
+ case Types.TINYINT:
+ case Types.SMALLINT:
+ return DataTypes.SMALLINT();
+ case Types.INTEGER:
+ return DataTypes.INT();
+ case Types.BIGINT:
+ return DataTypes.BIGINT();
+ case Types.FLOAT:
+ case Types.REAL:
+ return DataTypes.FLOAT();
+ case Types.DOUBLE:
+ return DataTypes.DOUBLE();
+ case Types.DECIMAL:
+ case Types.NUMERIC:
+ return DataTypes.DECIMAL(column.length(), column.scale().orElse(0));
+ case Types.DATE:
+ return DataTypes.DATE();
+ case Types.TIMESTAMP:
+ return column.length() >= 0
+ ? DataTypes.TIMESTAMP(column.length())
+ : DataTypes.TIMESTAMP();
+ default:
+ throw new UnsupportedOperationException(
+ String.format(
+ "Don't support DB2 type '%s' yet, jdbcType:'%s'.",
+ column.typeName(), column.jdbcType()));
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java
new file mode 100644
index 000000000..06c4e741f
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/utils/Db2Utils.java
@@ -0,0 +1,442 @@
+/*
+ * 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.connectors.db2.source.utils;
+
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.db2.source.offset.LsnOffset;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import io.debezium.connector.db2.Db2Connection;
+import io.debezium.connector.db2.Db2ConnectorConfig;
+import io.debezium.connector.db2.Db2DatabaseSchema;
+import io.debezium.connector.db2.Db2TopicSelector;
+import io.debezium.connector.db2.Lsn;
+import io.debezium.connector.db2.SourceInfo;
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableId;
+import io.debezium.schema.TopicSelector;
+import io.debezium.util.SchemaNameAdjuster;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import javax.annotation.Nullable;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.rowToArray;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.ROW;
+
+/** The utils for Db2 data source. */
+public class Db2Utils {
+
+ private static final String EVENT_SERIAL_NO_KEY = "event_serial_no";
+
+ public Db2Utils() {}
+
+ public static Object[] queryMinMax(JdbcConnection jdbc, TableId tableId, String columnName)
+ throws SQLException {
+ final String minMaxQuery =
+ String.format(
+ "SELECT MIN(%s), MAX(%s) FROM %s",
+ quote(columnName), quote(columnName), quoteSchemaAndTable(tableId));
+ return jdbc.queryAndMap(
+ minMaxQuery,
+ rs -> {
+ if (!rs.next()) {
+ // this should never happen
+ throw new SQLException(
+ String.format(
+ "No result returned after running query [%s]",
+ minMaxQuery));
+ }
+ return rowToArray(rs, 2);
+ });
+ }
+
+ public static long queryApproximateRowCnt(JdbcConnection jdbc, TableId tableId)
+ throws SQLException {
+ // The statement used to get approximate row count which is less
+ // accurate than COUNT(*), but is more efficient for large table.
+ final String rowCountQuery =
+ String.format(
+ "SELECT CARD\n"
+ + "FROM SYSCAT.TABLES\n"
+ + "WHERE TABSCHEMA='%s' and TABNAME='%s';",
+ tableId.schema(), tableId.table());
+ return jdbc.queryAndMap(
+ rowCountQuery,
+ rs -> {
+ if (!rs.next()) {
+ throw new SQLException(
+ String.format(
+ "No result returned after running query [%s]",
+ rowCountQuery));
+ }
+ return rs.getLong(1);
+ });
+ }
+
+ public static Object queryMin(
+ JdbcConnection jdbc, TableId tableId, String columnName, Object excludedLowerBound)
+ throws SQLException {
+ final String minQuery =
+ String.format(
+ "SELECT MIN(%s) FROM %s WHERE %s > ?",
+ columnName, quote(tableId), columnName);
+ return jdbc.prepareQueryAndMap(
+ minQuery,
+ ps -> ps.setObject(1, excludedLowerBound),
+ rs -> {
+ if (!rs.next()) {
+ // this should never happen
+ throw new SQLException(
+ String.format(
+ "No result returned after running query [%s]", minQuery));
+ }
+ return rs.getObject(1);
+ });
+ }
+
+ /**
+ * Returns the next LSN to be read from the database. This is the LSN of the last record that
+ * was read from the database.
+ */
+ public static Object queryNextChunkMax(
+ JdbcConnection jdbc,
+ TableId tableId,
+ String splitColumnName,
+ int chunkSize,
+ Object includedLowerBound)
+ throws SQLException {
+ String quotedColumn = quote(splitColumnName);
+ String query =
+ String.format(
+ "SELECT MAX(%s) FROM ("
+ + "SELECT %s FROM %s WHERE %s >= ? ORDER BY %s ASC "
+ + " FETCH FIRST %S ROWS ONLY ) AS T",
+ quotedColumn,
+ quotedColumn,
+ quote(tableId),
+ quotedColumn,
+ quotedColumn,
+ chunkSize);
+ return jdbc.prepareQueryAndMap(
+ query,
+ ps -> ps.setObject(1, includedLowerBound),
+ rs -> {
+ if (!rs.next()) {
+ // this should never happen
+ throw new SQLException(
+ String.format(
+ "No result returned after running query [%s]", query));
+ }
+ return rs.getObject(1);
+ });
+ }
+
+ public static Column getSplitColumn(Table table, @Nullable String chunkKeyColumn) {
+ List<Column> primaryKeys = table.primaryKeyColumns();
+ if (primaryKeys.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "Incremental snapshot for tables requires primary key,"
+ + " but table %s doesn't have primary key.",
+ table.id()));
+ }
+
+ if (chunkKeyColumn != null) {
+ Optional<Column> targetPkColumn =
+ primaryKeys.stream()
+ .filter(col -> chunkKeyColumn.equals(col.name()))
+ .findFirst();
+ if (targetPkColumn.isPresent()) {
+ return targetPkColumn.get();
+ }
+ throw new ValidationException(
+ String.format(
+ "Chunk key column '%s' doesn't exist in the primary key [%s] of the table %s.",
+ chunkKeyColumn,
+ primaryKeys.stream().map(Column::name).collect(Collectors.joining(",")),
+ table.id()));
+ }
+
+ // use first field in primary key as the split key
+ return primaryKeys.get(0);
+ }
+
+ public static RowType getSplitType(Column splitColumn) {
+ return (RowType)
+ ROW(FIELD(splitColumn.name(), Db2TypeUtils.fromDbzColumn(splitColumn)))
+ .getLogicalType();
+ }
+
+ public static Offset getLsn(SourceRecord record) {
+ return getLsnPosition(record.sourceOffset());
+ }
+
+ public static LsnOffset getLsnPosition(Map<String, ?> offset) {
+ Map<String, String> offsetStrMap = new HashMap<>();
+ for (Map.Entry<String, ?> entry : offset.entrySet()) {
+ offsetStrMap.put(
+ entry.getKey(), entry.getValue() == null ? null : entry.getValue().toString());
+ }
+ Lsn changeLsn = Lsn.valueOf(offsetStrMap.get(SourceInfo.CHANGE_LSN_KEY));
+ Lsn commitLsn = Lsn.valueOf(offsetStrMap.get(SourceInfo.COMMIT_LSN_KEY));
+ Long eventSerialNo = null;
+ if (offsetStrMap.get(EVENT_SERIAL_NO_KEY) != null) {
+ eventSerialNo = Long.parseLong(offsetStrMap.get(EVENT_SERIAL_NO_KEY));
+ }
+ return new LsnOffset(changeLsn, commitLsn, eventSerialNo);
+ }
+
+ /** Fetch current largest log sequence number (LSN) of the database. */
+ public static LsnOffset currentLsn(Db2Connection connection) {
+ try {
+ Lsn maxLsn = connection.getMaxLsn();
+ return new LsnOffset(maxLsn, maxLsn, null);
+ } catch (SQLException e) {
+ throw new FlinkRuntimeException(e.getMessage(), e);
+ }
+ }
+
+ /** Get split scan query for the given table. */
+ public static String buildSplitScanQuery(
+ TableId tableId, RowType pkRowType, boolean isFirstSplit, boolean isLastSplit) {
+ return buildSplitQuery(tableId, pkRowType, isFirstSplit, isLastSplit, -1, true);
+ }
+
+ /** Get table split data PreparedStatement. */
+ public static PreparedStatement readTableSplitDataStatement(
+ JdbcConnection jdbc,
+ String sql,
+ boolean isFirstSplit,
+ boolean isLastSplit,
+ Object[] splitStart,
+ Object[] splitEnd,
+ int primaryKeyNum,
+ int fetchSize) {
+ try {
+ final PreparedStatement statement = initStatement(jdbc, sql, fetchSize);
+ if (isFirstSplit && isLastSplit) {
+ return statement;
+ }
+ if (isFirstSplit) {
+ for (int i = 0; i < primaryKeyNum; i++) {
+ statement.setObject(i + 1, splitEnd[i]);
+ statement.setObject(i + 1 + primaryKeyNum, splitEnd[i]);
+ }
+ } else if (isLastSplit) {
+ for (int i = 0; i < primaryKeyNum; i++) {
+ statement.setObject(i + 1, splitStart[i]);
+ }
+ } else {
+ for (int i = 0; i < primaryKeyNum; i++) {
+ statement.setObject(i + 1, splitStart[i]);
+ statement.setObject(i + 1 + primaryKeyNum, splitEnd[i]);
+ statement.setObject(i + 1 + 2 * primaryKeyNum, splitEnd[i]);
+ }
+ }
+ return statement;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to build the split data read statement.", e);
+ }
+ }
+
+ public static Db2DatabaseSchema createDb2DatabaseSchema(
+ Db2ConnectorConfig connectorConfig, Db2Connection connection) {
+ TopicSelector<TableId> topicSelector = Db2TopicSelector.defaultSelector(connectorConfig);
+ SchemaNameAdjuster schemaNameAdjuster = SchemaNameAdjuster.create();
+
+ return new Db2DatabaseSchema(
+ connectorConfig, schemaNameAdjuster, topicSelector, connection);
+ }
+
+ // --------------------------private method-------------------------------
+
+ private static String getPrimaryKeyColumnsProjection(RowType pkRowType) {
+ StringBuilder sql = new StringBuilder();
+ for (Iterator<String> fieldNamesIt = pkRowType.getFieldNames().iterator();
+ fieldNamesIt.hasNext(); ) {
+ sql.append(fieldNamesIt.next());
+ if (fieldNamesIt.hasNext()) {
+ sql.append(" , ");
+ }
+ }
+ return sql.toString();
+ }
+
+ private static String buildSplitQuery(
+ TableId tableId,
+ RowType pkRowType,
+ boolean isFirstSplit,
+ boolean isLastSplit,
+ int limitSize,
+ boolean isScanningData) {
+ final String condition;
+
+ if (isFirstSplit && isLastSplit) {
+ condition = null;
+ } else if (isFirstSplit) {
+ final StringBuilder sql = new StringBuilder();
+ addPrimaryKeyColumnsToCondition(pkRowType, sql, " <= ?");
+ if (isScanningData) {
+ sql.append(" AND NOT (");
+ addPrimaryKeyColumnsToCondition(pkRowType, sql, " = ?");
+ sql.append(")");
+ }
+ condition = sql.toString();
+ } else if (isLastSplit) {
+ final StringBuilder sql = new StringBuilder();
+ addPrimaryKeyColumnsToCondition(pkRowType, sql, " >= ?");
+ condition = sql.toString();
+ } else {
+ final StringBuilder sql = new StringBuilder();
+ addPrimaryKeyColumnsToCondition(pkRowType, sql, " >= ?");
+ if (isScanningData) {
+ sql.append(" AND NOT (");
+ addPrimaryKeyColumnsToCondition(pkRowType, sql, " = ?");
+ sql.append(")");
+ }
+ sql.append(" AND ");
+ addPrimaryKeyColumnsToCondition(pkRowType, sql, " <= ?");
+ condition = sql.toString();
+ }
+
+ if (isScanningData) {
+ return buildSelectWithRowLimits(
+ tableId, limitSize, "*", Optional.ofNullable(condition), Optional.empty());
+ } else {
+ final String orderBy =
+ pkRowType.getFieldNames().stream().collect(Collectors.joining(", "));
+ return buildSelectWithBoundaryRowLimits(
+ tableId,
+ limitSize,
+ getPrimaryKeyColumnsProjection(pkRowType),
+ getMaxPrimaryKeyColumnsProjection(pkRowType),
+ Optional.ofNullable(condition),
+ orderBy);
+ }
+ }
+
+ private static PreparedStatement initStatement(JdbcConnection jdbc, String sql, int fetchSize)
+ throws SQLException {
+ final Connection connection = jdbc.connection();
+ connection.setAutoCommit(false);
+ final PreparedStatement statement = connection.prepareStatement(sql);
+ statement.setFetchSize(fetchSize);
+ return statement;
+ }
+
+ private static String getMaxPrimaryKeyColumnsProjection(RowType pkRowType) {
+ StringBuilder sql = new StringBuilder();
+ for (Iterator<String> fieldNamesIt = pkRowType.getFieldNames().iterator();
+ fieldNamesIt.hasNext(); ) {
+ sql.append("MAX(" + fieldNamesIt.next() + ")");
+ if (fieldNamesIt.hasNext()) {
+ sql.append(" , ");
+ }
+ }
+ return sql.toString();
+ }
+
+ private static String buildSelectWithRowLimits(
+ TableId tableId,
+ int limit,
+ String projection,
+ Optional<String> condition,
+ Optional<String> orderBy) {
+ final StringBuilder sql = new StringBuilder("SELECT ");
+ sql.append(projection).append(" FROM ");
+ sql.append(quoteSchemaAndTable(tableId));
+ if (condition.isPresent()) {
+ sql.append(" WHERE ").append(condition.get());
+ }
+ if (orderBy.isPresent()) {
+ sql.append(" ORDER BY ").append(orderBy.get());
+ }
+ if (limit > 0) {
+ sql.append(" FETCH FIRST ").append(limit).append(" ROWS ONLY ");
+ }
+ return sql.toString();
+ }
+
+ private static String quoteSchemaAndTable(TableId tableId) {
+ StringBuilder quoted = new StringBuilder();
+
+ if (tableId.schema() != null && !tableId.schema().isEmpty()) {
+ quoted.append(quote(tableId.schema())).append(".");
+ }
+
+ quoted.append(quote(tableId.table()));
+ return quoted.toString();
+ }
+
+ public static String quote(String dbOrTableName) {
+ return "\"" + dbOrTableName + "\"";
+ }
+
+ public static String quote(TableId tableId) {
+ return "\"" + tableId.schema() + "\".\"" + tableId.table() + "\"";
+ }
+
+ private static void addPrimaryKeyColumnsToCondition(
+ RowType pkRowType, StringBuilder sql, String predicate) {
+ for (Iterator<String> fieldNamesIt = pkRowType.getFieldNames().iterator();
+ fieldNamesIt.hasNext(); ) {
+ sql.append(fieldNamesIt.next()).append(predicate);
+ if (fieldNamesIt.hasNext()) {
+ sql.append(" AND ");
+ }
+ }
+ }
+
+ private static String buildSelectWithBoundaryRowLimits(
+ TableId tableId,
+ int limit,
+ String projection,
+ String maxColumnProjection,
+ Optional<String> condition,
+ String orderBy) {
+ final StringBuilder sql = new StringBuilder("SELECT ");
+ sql.append(maxColumnProjection);
+ sql.append(" FROM (");
+ sql.append("SELECT ");
+ sql.append(projection);
+ sql.append(" FROM ");
+ sql.append(quoteSchemaAndTable(tableId));
+ if (condition.isPresent()) {
+ sql.append(" WHERE ").append(condition.get());
+ }
+ sql.append(" ORDER BY ").append(orderBy);
+ sql.append("FETCH FIRST ").append(limit).append(" ROWS ONLY ");
+ sql.append(") T");
+ return sql.toString();
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java
index 7411dfa97..5a5d6eb17 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSource.java
@@ -18,7 +18,10 @@
package org.apache.flink.cdc.connectors.db2.table;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
import org.apache.flink.cdc.connectors.db2.Db2Source;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder;
import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.cdc.debezium.table.MetadataConverter;
@@ -28,11 +31,15 @@ import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.SourceProvider;
import org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
+import javax.annotation.Nullable;
+
+import java.time.Duration;
import java.time.ZoneId;
import java.util.Collections;
import java.util.List;
@@ -52,13 +59,24 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
private final int port;
private final String hostname;
private final String database;
- private final String schemaName;
private final String tableName;
private final String username;
private final String password;
private final ZoneId serverTimeZone;
private final StartupOptions startupOptions;
private final Properties dbzProperties;
+ private final boolean enableParallelRead;
+ private final int splitSize;
+ private final int splitMetaGroupSize;
+ private final int fetchSize;
+ private final Duration connectTimeout;
+ private final int connectionPoolSize;
+ private final int connectMaxRetries;
+ private final double distributionFactorUpper;
+ private final double distributionFactorLower;
+ private final String chunkKeyColumn;
+ private final boolean closeIdleReaders;
+ private final boolean skipSnapshotBackfill;
/** Metadata that is appended at the end of a physical source row. */
protected List<String> metadataKeys;
@@ -68,18 +86,28 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
int port,
String hostname,
String database,
- String schemaName,
String tableName,
String username,
String password,
ZoneId serverTimeZone,
Properties dbzProperties,
- StartupOptions startupOptions) {
+ StartupOptions startupOptions,
+ boolean enableParallelRead,
+ int splitSize,
+ int splitMetaGroupSize,
+ int fetchSize,
+ Duration connectTimeout,
+ int connectMaxRetries,
+ int connectionPoolSize,
+ double distributionFactorUpper,
+ double distributionFactorLower,
+ @Nullable String chunkKeyColumn,
+ boolean closeIdleReaders,
+ boolean skipSnapshotBackfill) {
this.physicalSchema = physicalSchema;
this.port = port;
this.hostname = hostname;
this.database = database;
- this.schemaName = schemaName;
this.tableName = tableName;
this.username = username;
this.password = password;
@@ -88,6 +116,18 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
this.startupOptions = startupOptions;
this.producedDataType = physicalSchema.toPhysicalRowDataType();
this.metadataKeys = Collections.emptyList();
+ this.enableParallelRead = enableParallelRead;
+ this.splitSize = splitSize;
+ this.splitMetaGroupSize = splitMetaGroupSize;
+ this.fetchSize = fetchSize;
+ this.connectTimeout = connectTimeout;
+ this.connectionPoolSize = connectionPoolSize;
+ this.connectMaxRetries = connectMaxRetries;
+ this.distributionFactorUpper = distributionFactorUpper;
+ this.distributionFactorLower = distributionFactorLower;
+ this.chunkKeyColumn = chunkKeyColumn;
+ this.closeIdleReaders = closeIdleReaders;
+ this.skipSnapshotBackfill = skipSnapshotBackfill;
}
@Override
@@ -109,19 +149,48 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
.setResultTypeInfo(typeInfo)
.setServerTimeZone(serverTimeZone)
.build();
- Db2Source.Builder<RowData> builder =
- Db2Source.<RowData>builder()
- .hostname(hostname)
- .port(port)
- .database(database)
- .tableList(schemaName + "." + tableName)
- .username(username)
- .password(password)
- .debeziumProperties(dbzProperties)
- .deserializer(deserializer)
- .startupOptions(startupOptions);
- DebeziumSourceFunction<RowData> sourceFunction = builder.build();
- return SourceFunctionProvider.of(sourceFunction, false);
+
+ if (enableParallelRead) {
+ JdbcIncrementalSource<RowData> db2ChangeEventSource =
+ Db2SourceBuilder.Db2IncrementalSource.<RowData>builder()
+ .hostname(hostname)
+ .port(port)
+ .databaseList(database)
+ .tableList(tableName)
+ .serverTimeZone(serverTimeZone.toString())
+ .username(username)
+ .password(password)
+ .startupOptions(startupOptions)
+ .deserializer(deserializer)
+ .debeziumProperties(dbzProperties)
+ .splitSize(splitSize)
+ .splitMetaGroupSize(splitMetaGroupSize)
+ .fetchSize(fetchSize)
+ .connectTimeout(connectTimeout)
+ .connectionPoolSize(connectionPoolSize)
+ .connectMaxRetries(connectMaxRetries)
+ .distributionFactorUpper(distributionFactorUpper)
+ .distributionFactorLower(distributionFactorLower)
+ .chunkKeyColumn(chunkKeyColumn)
+ .closeIdleReaders(closeIdleReaders)
+ .skipSnapshotBackfill(skipSnapshotBackfill)
+ .build();
+ return SourceProvider.of(db2ChangeEventSource);
+ } else {
+ Db2Source.Builder<RowData> builder =
+ Db2Source.<RowData>builder()
+ .hostname(hostname)
+ .port(port)
+ .database(database)
+ .tableList(tableName)
+ .username(username)
+ .password(password)
+ .debeziumProperties(dbzProperties)
+ .deserializer(deserializer)
+ .startupOptions(startupOptions);
+ DebeziumSourceFunction<RowData> sourceFunction = builder.build();
+ return SourceFunctionProvider.of(sourceFunction, false);
+ }
}
private MetadataConverter[] getMetadataConverters() {
@@ -148,13 +217,24 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
port,
hostname,
database,
- schemaName,
tableName,
username,
password,
serverTimeZone,
dbzProperties,
- startupOptions);
+ startupOptions,
+ enableParallelRead,
+ splitSize,
+ splitMetaGroupSize,
+ fetchSize,
+ connectTimeout,
+ connectMaxRetries,
+ connectionPoolSize,
+ distributionFactorUpper,
+ distributionFactorLower,
+ chunkKeyColumn,
+ closeIdleReaders,
+ skipSnapshotBackfill);
source.metadataKeys = metadataKeys;
source.producedDataType = producedDataType;
return source;
@@ -173,13 +253,24 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
&& Objects.equals(physicalSchema, that.physicalSchema)
&& Objects.equals(hostname, that.hostname)
&& Objects.equals(database, that.database)
- && Objects.equals(schemaName, that.schemaName)
&& Objects.equals(tableName, that.tableName)
&& Objects.equals(username, that.username)
&& Objects.equals(password, that.password)
&& Objects.equals(serverTimeZone, that.serverTimeZone)
&& Objects.equals(dbzProperties, that.dbzProperties)
- && Objects.equals(metadataKeys, that.metadataKeys);
+ && Objects.equals(metadataKeys, that.metadataKeys)
+ && Objects.equals(enableParallelRead, that.enableParallelRead)
+ && Objects.equals(splitSize, that.splitSize)
+ && Objects.equals(splitMetaGroupSize, that.splitMetaGroupSize)
+ && Objects.equals(fetchSize, that.fetchSize)
+ && Objects.equals(connectTimeout, that.connectTimeout)
+ && Objects.equals(connectMaxRetries, that.connectMaxRetries)
+ && Objects.equals(connectionPoolSize, that.connectionPoolSize)
+ && Objects.equals(distributionFactorUpper, that.distributionFactorUpper)
+ && Objects.equals(distributionFactorLower, that.distributionFactorLower)
+ && Objects.equals(chunkKeyColumn, that.chunkKeyColumn)
+ && Objects.equals(closeIdleReaders, that.closeIdleReaders)
+ && Objects.equals(skipSnapshotBackfill, that.skipSnapshotBackfill);
}
@Override
@@ -189,13 +280,24 @@ public class Db2TableSource implements ScanTableSource, SupportsReadingMetadata
port,
hostname,
database,
- schemaName,
tableName,
username,
password,
serverTimeZone,
dbzProperties,
- metadataKeys);
+ metadataKeys,
+ enableParallelRead,
+ splitSize,
+ splitMetaGroupSize,
+ fetchSize,
+ connectTimeout,
+ connectMaxRetries,
+ connectionPoolSize,
+ distributionFactorUpper,
+ distributionFactorLower,
+ chunkKeyColumn,
+ closeIdleReaders,
+ skipSnapshotBackfill);
}
@Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java
index 7ad2238c4..2c1444927 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactory.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.connectors.db2.table;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.cdc.connectors.db2.utils.OptionUtils;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
@@ -28,12 +29,27 @@ import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
+import java.time.Duration;
import java.time.ZoneId;
import java.util.HashSet;
import java.util.Set;
+import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CONNECTION_POOL_SIZE;
+import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CONNECT_MAX_RETRIES;
+import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.CONNECT_TIMEOUT;
+import static org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.CHUNK_META_GROUP_SIZE;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND;
+import static org.apache.flink.cdc.connectors.base.options.SourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND;
+import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare;
import static org.apache.flink.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX;
import static org.apache.flink.cdc.debezium.table.DebeziumOptions.getDebeziumProperties;
+import static org.apache.flink.util.Preconditions.checkState;
/** Table source factory for DB2 CDC connector. */
public class Db2TableSourceFactory implements DynamicTableSourceFactory {
@@ -71,12 +87,6 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
.noDefaultValue()
.withDescription("Database name of the DB2 server to monitor.");
- private static final ConfigOption<String> SCHEMA_NAME =
- ConfigOptions.key("schema-name")
- .stringType()
- .noDefaultValue()
- .withDescription("Schema name of the DB2 database to monitor.");
-
private static final ConfigOption<String> TABLE_NAME =
ConfigOptions.key("table-name")
.stringType()
@@ -109,13 +119,36 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
String username = config.get(USERNAME);
String password = config.get(PASSWORD);
String databaseName = config.get(DATABASE_NAME);
- String schemaName = config.get(SCHEMA_NAME);
String tableName = config.get(TABLE_NAME);
int port = config.get(PORT);
ZoneId serverTimeZone = ZoneId.of(config.get(SERVER_TIME_ZONE));
ResolvedSchema physicalSchema = context.getCatalogTable().getResolvedSchema();
StartupOptions startupOptions = getStartupOptions(config);
+ boolean enableParallelRead = config.get(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
+ int splitSize = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE);
+ int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE);
+ int fetchSize = config.get(SCAN_SNAPSHOT_FETCH_SIZE);
+ Duration connectTimeout = config.get(CONNECT_TIMEOUT);
+ int connectMaxRetries = config.get(CONNECT_MAX_RETRIES);
+ int connectionPoolSize = config.get(CONNECTION_POOL_SIZE);
+ double distributionFactorUpper = config.get(SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
+ double distributionFactorLower = config.get(SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
+ String chunkKeyColumn =
+ config.getOptional(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN).orElse(null);
+ boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
+ boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+
+ if (enableParallelRead) {
+ validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
+ validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1);
+ validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
+ validateIntegerOption(CONNECTION_POOL_SIZE, connectionPoolSize, 1);
+ validateIntegerOption(CONNECT_MAX_RETRIES, connectMaxRetries, 0);
+ validateDistributionFactorUpper(distributionFactorUpper);
+ validateDistributionFactorLower(distributionFactorLower);
+ }
+
OptionUtils.printOptions(IDENTIFIER, ((Configuration) config).toMap());
return new Db2TableSource(
@@ -123,13 +156,24 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
port,
hostname,
databaseName,
- schemaName,
tableName,
username,
password,
serverTimeZone,
getDebeziumProperties(context.getCatalogTable().getOptions()),
- startupOptions);
+ startupOptions,
+ enableParallelRead,
+ splitSize,
+ splitMetaGroupSize,
+ fetchSize,
+ connectTimeout,
+ connectMaxRetries,
+ connectionPoolSize,
+ distributionFactorUpper,
+ distributionFactorLower,
+ chunkKeyColumn,
+ closeIdleReaders,
+ skipSnapshotBackfill);
}
@Override
@@ -142,7 +186,6 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
Set<ConfigOption<?>> options = new HashSet<>();
options.add(HOSTNAME);
options.add(DATABASE_NAME);
- options.add(SCHEMA_NAME);
options.add(TABLE_NAME);
options.add(USERNAME);
options.add(PASSWORD);
@@ -155,6 +198,18 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
options.add(PORT);
options.add(SERVER_TIME_ZONE);
options.add(SCAN_STARTUP_MODE);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE);
+ options.add(CHUNK_META_GROUP_SIZE);
+ options.add(SCAN_SNAPSHOT_FETCH_SIZE);
+ options.add(CONNECT_TIMEOUT);
+ options.add(CONNECT_MAX_RETRIES);
+ options.add(CONNECTION_POOL_SIZE);
+ options.add(SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
+ options.add(SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
+ options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
+ options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
return options;
}
@@ -181,4 +236,38 @@ public class Db2TableSourceFactory implements DynamicTableSourceFactory {
modeString));
}
}
+
+ /** Checks the value of given integer option is valid. */
+ private void validateIntegerOption(
+ ConfigOption<Integer> option, int optionValue, int exclusiveMin) {
+ checkState(
+ optionValue > exclusiveMin,
+ String.format(
+ "The value of option '%s' must larger than %d, but is %d",
+ option.key(), exclusiveMin, optionValue));
+ }
+
+ /** Checks the value of given evenly distribution factor upper bound is valid. */
+ private void validateDistributionFactorUpper(double distributionFactorUpper) {
+ checkState(
+ doubleCompare(distributionFactorUpper, 1.0d) >= 0,
+ String.format(
+ "The value of option '%s' must larger than or equals %s, but is %s",
+ SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(),
+ 1.0d,
+ distributionFactorUpper));
+ }
+
+ /** Checks the value of given evenly distribution factor lower bound is valid. */
+ private void validateDistributionFactorLower(double distributionFactorLower) {
+ checkState(
+ doubleCompare(distributionFactorLower, 0.0d) >= 0
+ && doubleCompare(distributionFactorLower, 1.0d) <= 0,
+ String.format(
+ "The value of option '%s' must between %s and %s inclusively, but is %s",
+ SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(),
+ 0.0d,
+ 1.0d,
+ distributionFactorLower));
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupOptions.java
index a06140889..bac47c6b1 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupOptions.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/table/StartupOptions.java
@@ -17,10 +17,13 @@
package org.apache.flink.cdc.connectors.db2.table;
+import org.apache.flink.cdc.connectors.base.options.StartupMode;
+
import java.util.Objects;
/** Debezium startup options. */
public final class StartupOptions {
+
public final StartupMode startupMode;
/**
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2ParallelSourceExampleTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2ParallelSourceExampleTest.java
new file mode 100644
index 000000000..daca31a72
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2ParallelSourceExampleTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.connectors.db2;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder.Db2IncrementalSource;
+import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import static org.testcontainers.containers.Db2Container.DB2_PORT;
+
+/** Example Tests for {@link Db2IncrementalSource}. */
+public class Db2ParallelSourceExampleTest extends Db2TestBase {
+
+ @Test
+ @Ignore("Test ignored because it won't stop and is used for manual test")
+ public void testDb2ExampleSource() throws Exception {
+ initializeDb2Table("customers", "CUSTOMERS");
+ Db2IncrementalSource<String> sqlServerSource =
+ new Db2SourceBuilder()
+ .hostname(DB2_CONTAINER.getHost())
+ .port(DB2_CONTAINER.getMappedPort(DB2_PORT))
+ .databaseList(DB2_CONTAINER.getDatabaseName())
+ .tableList("DB2INST1.CUSTOMERS")
+ .username(DB2_CONTAINER.getUsername())
+ .password(DB2_CONTAINER.getPassword())
+ .deserializer(new JsonDebeziumDeserializationSchema())
+ .startupOptions(StartupOptions.initial())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ // enable checkpoint
+ env.enableCheckpointing(3000);
+ // set the source parallelism to 2
+ env.fromSource(sqlServerSource, WatermarkStrategy.noWatermarks(), "Db2IncrementalSource")
+ .setParallelism(2)
+ .print()
+ .setParallelism(1);
+
+ env.execute("Print DB2 Snapshot + Change Stream");
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2SourceTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2SourceTest.java
index a6c4041fe..509d837a6 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2SourceTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2SourceTest.java
@@ -70,7 +70,8 @@ public class Db2SourceTest extends Db2TestBase {
@Test
public void testConsumingAllEvents() throws Exception {
- DebeziumSourceFunction<SourceRecord> source = createDb2Source("DB2INST1.PRODUCTS1");
+ initializeDb2Table("inventory", "PRODUCTS");
+ DebeziumSourceFunction<SourceRecord> source = createDb2Source("DB2INST1.PRODUCTS");
TestSourceContext<SourceRecord> sourceContext = new TestSourceContext<>();
setupSource(source);
@@ -94,12 +95,12 @@ public class Db2SourceTest extends Db2TestBase {
}
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS1 VALUES (default,'robot','Toy robot',1.304)"); // 110
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (default,'robot','Toy robot',1.304)"); // 110
records = drain(sourceContext, 1);
assertInsert(records.get(0), "ID", 110);
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS1 VALUES (1001,'roy','old robot',1234.56)"); // 1001
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (1001,'roy','old robot',1234.56)"); // 1001
records = drain(sourceContext, 1);
assertInsert(records.get(0), "ID", 1001);
@@ -108,7 +109,7 @@ public class Db2SourceTest extends Db2TestBase {
// (TOMBSTONE is dropped)
// ---------------------------------------------------------------------------------------------------------------
statement.execute(
- "UPDATE DB2INST1.PRODUCTS1 SET ID=2001, DESCRIPTION='really old robot' WHERE ID=1001");
+ "UPDATE DB2INST1.PRODUCTS SET ID=2001, DESCRIPTION='really old robot' WHERE ID=1001");
records = drain(sourceContext, 2);
assertDelete(records.get(0), "ID", 1001);
assertInsert(records.get(1), "ID", 2001);
@@ -116,7 +117,7 @@ public class Db2SourceTest extends Db2TestBase {
// ---------------------------------------------------------------------------------------------------------------
// Simple UPDATE (with no schema changes)
// ---------------------------------------------------------------------------------------------------------------
- statement.execute("UPDATE DB2INST1.PRODUCTS1 SET WEIGHT=1345.67 WHERE ID=2001");
+ statement.execute("UPDATE DB2INST1.PRODUCTS SET WEIGHT=1345.67 WHERE ID=2001");
records = drain(sourceContext, 1);
assertUpdate(records.get(0), "ID", 2001);
@@ -126,8 +127,8 @@ public class Db2SourceTest extends Db2TestBase {
// Add a column with default to the 'products' table and explicitly update one record
// ...
statement.execute(
- "ALTER TABLE DB2INST1.PRODUCTS1 ADD COLUMN VOLUME FLOAT ADD COLUMN ALIAS VARCHAR(30) NULL");
- statement.execute("UPDATE DB2INST1.PRODUCTS1 SET VOLUME=13.5 WHERE ID=2001");
+ "ALTER TABLE DB2INST1.PRODUCTS ADD COLUMN VOLUME FLOAT ADD COLUMN ALIAS VARCHAR(30) NULL");
+ statement.execute("UPDATE DB2INST1.PRODUCTS SET VOLUME=13.5 WHERE ID=2001");
records = drain(sourceContext, 1);
assertUpdate(records.get(0), "ID", 2001);
@@ -139,6 +140,7 @@ public class Db2SourceTest extends Db2TestBase {
@Test
public void testCheckpointAndRestore() throws Exception {
+ initializeDb2Table("inventory", "PRODUCTS");
final TestingListState<byte[]> offsetState = new TestingListState<>();
final TestingListState<String> historyState = new TestingListState<>();
String prevLsn = "";
@@ -147,7 +149,7 @@ public class Db2SourceTest extends Db2TestBase {
// Step-1: start the source from empty state
// ---------------------------------------------------------------------------
final DebeziumSourceFunction<SourceRecord> source =
- createDb2Source("DB2INST1.PRODUCTS2");
+ createDb2Source("DB2INST1.PRODUCTS");
// we use blocking context to block the source to emit before last snapshot record
final BlockingSourceContext<SourceRecord> sourceContext =
new BlockingSourceContext<>(8);
@@ -207,7 +209,7 @@ public class Db2SourceTest extends Db2TestBase {
// Step-3: restore the source from state
// ---------------------------------------------------------------------------
final DebeziumSourceFunction<SourceRecord> source2 =
- createDb2Source("DB2INST1.PRODUCTS2");
+ createDb2Source("DB2INST1.PRODUCTS");
final TestSourceContext<SourceRecord> sourceContext2 = new TestSourceContext<>();
setupSource(source2, 1L, offsetState, historyState, true, 0, 1);
final CheckedThread runThread2 =
@@ -226,7 +228,7 @@ public class Db2SourceTest extends Db2TestBase {
Statement statement = connection.createStatement()) {
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS2 VALUES (default,'robot','Toy robot',1.304)"); // 110
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (default,'robot','Toy robot',1.304)"); // 110
List<SourceRecord> records = drain(sourceContext2, 1);
assertEquals(1, records.size());
assertInsert(records.get(0), "ID", 110);
@@ -245,10 +247,10 @@ public class Db2SourceTest extends Db2TestBase {
String lsn = JsonPath.read(state, "$.sourceOffset.commit_lsn");
assertTrue(lsn.compareTo(prevLsn) > 0);
- // execute 2 more DMLs to have more binlog
+ // execute 2 more DMLs to have more redo logs
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS2 VALUES (1001,'roy','old robot',1234.56)"); // 1001
- statement.execute("UPDATE DB2INST1.PRODUCTS2 SET WEIGHT=1345.67 WHERE ID=1001");
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (1001,'roy','old robot',1234.56)"); // 1001
+ statement.execute("UPDATE DB2INST1.PRODUCTS SET WEIGHT=1345.67 WHERE ID=1001");
}
// cancel the source
@@ -261,7 +263,7 @@ public class Db2SourceTest extends Db2TestBase {
// Step-5: restore the source from checkpoint-2
// ---------------------------------------------------------------------------
final DebeziumSourceFunction<SourceRecord> source3 =
- createDb2Source("DB2INST1.PRODUCTS2");
+ createDb2Source("DB2INST1.PRODUCTS");
final TestSourceContext<SourceRecord> sourceContext3 = new TestSourceContext<>();
setupSource(source3, 2L, offsetState, historyState, true, 0, 1);
@@ -275,7 +277,7 @@ public class Db2SourceTest extends Db2TestBase {
};
runThread3.start();
- // consume the unconsumed binlog
+ // consume the unconsumed redo logs
List<SourceRecord> records = drain(sourceContext3, 2);
assertInsert(records.get(0), "ID", 1001);
assertUpdate(records.get(1), "ID", 1001);
@@ -286,7 +288,7 @@ public class Db2SourceTest extends Db2TestBase {
// can continue to receive new events
try (Connection connection = getJdbcConnection();
Statement statement = connection.createStatement()) {
- statement.execute("DELETE FROM DB2INST1.PRODUCTS2 WHERE ID=1001");
+ statement.execute("DELETE FROM DB2INST1.PRODUCTS WHERE ID=1001");
}
records = drain(sourceContext3, 1);
assertDelete(records.get(0), "ID", 1001);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2TestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2TestBase.java
index 4a4c57b36..76b389c04 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2TestBase.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/Db2TestBase.java
@@ -17,6 +17,11 @@
package org.apache.flink.cdc.connectors.db2;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.apache.commons.lang3.StringUtils;
+import org.awaitility.Awaitility;
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,16 +33,32 @@ import org.testcontainers.utility.DockerImageName;
import java.net.URISyntaxException;
import java.net.URL;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static java.lang.String.format;
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
-/** Basic class for testing DB2 source, this contains a DB2 container which enables binlog. */
+/** Basic class for testing DB2 source, this contains a DB2 container which enables redo logs. */
public class Db2TestBase {
private static final Logger LOG = LoggerFactory.getLogger(Db2TestBase.class);
@@ -49,6 +70,7 @@ public class Db2TestBase {
.get())
.asCompatibleSubstituteFor("ibmcom/db2");
private static boolean db2AsnAgentRunning = false;
+ private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
protected static final Db2Container DB2_CONTAINER =
new Db2Container(DEBEZIUM_DOCKER_IMAGE_NAME)
@@ -85,6 +107,15 @@ public class Db2TestBase {
LOG.info("Db2 asn agent are started.");
}
+ @AfterClass
+ public static void stopContainers() {
+ LOG.info("Stopping containers...");
+ if (DB2_CONTAINER != null) {
+ DB2_CONTAINER.stop();
+ }
+ LOG.info("Containers are stopped.");
+ }
+
protected Connection getJdbcConnection() throws SQLException {
return DriverManager.getConnection(
DB2_CONTAINER.getJdbcUrl(),
@@ -103,4 +134,167 @@ public class Db2TestBase {
}
return path;
}
+
+ private static void dropTestTable(Connection connection, String tableName) {
+
+ try {
+ Awaitility.await(String.format("cdc remove table %s", tableName))
+ .atMost(30, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ String removeSql =
+ String.format(
+ "CALL ASNCDC.REMOVETABLE('DB2INST1', '%s')",
+ tableName);
+ connection.createStatement().execute(removeSql);
+ String reinitSql =
+ String.format(
+ "VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');",
+ tableName);
+ connection.createStatement().execute(reinitSql);
+ return true;
+ } catch (SQLException e) {
+ LOG.warn(
+ String.format(
+ "cdc remove TABLE %s failed (will be retried): {}",
+ tableName),
+ e.getMessage());
+ return false;
+ }
+ });
+ } catch (Exception e) {
+ throw new FlinkRuntimeException("Failed to remove cdc table " + tableName, e);
+ }
+
+ try {
+ Awaitility.await(String.format("Dropping table %s", tableName))
+ .atMost(30, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ String sql = String.format("DROP TABLE DB2INST1.%s", tableName);
+ connection.createStatement().execute(sql);
+ connection.commit();
+ return true;
+ } catch (SQLException e) {
+ LOG.warn(
+ String.format(
+ "DROP TABLE %s failed (will be retried): {}",
+ tableName),
+ e.getMessage());
+ return false;
+ }
+ });
+ } catch (Exception e) {
+ throw new FlinkRuntimeException("Failed to drop table", e);
+ }
+ }
+
+ private static boolean checkTableExists(Connection connection, String tableName) {
+ AtomicBoolean tableExists = new AtomicBoolean(false);
+ try {
+ Awaitility.await(String.format("check table %s exists or not", tableName))
+ .atMost(30, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ String tableExistSql =
+ String.format(
+ "SELECT COUNT(*) FROM SYSCAT.TABLES WHERE TABNAME = '%s' AND "
+ + "TABSCHEMA = 'DB2INST1';",
+ tableName);
+ ResultSet resultSet =
+ connection
+ .createStatement()
+ .executeQuery(tableExistSql);
+ if (resultSet.next()) {
+ if (resultSet.getInt(1) == 1) {
+ tableExists.set(true);
+ }
+ }
+ return true;
+ } catch (SQLException e) {
+ LOG.warn(
+ String.format(
+ "check table %s exists failed", tableName),
+ e.getMessage());
+ return false;
+ }
+ });
+ } catch (Exception e) {
+ throw new FlinkRuntimeException("Failed to check table " + tableName + " exists", e);
+ }
+ return tableExists.get();
+ }
+
+ /**
+ * Executes a JDBC statement using the default jdbc config without autocommitting the
+ * connection.
+ */
+ protected void initializeDb2Table(String sqlFile, String tableName) {
+ final String ddlFile = String.format("db2_server/%s.sql", sqlFile);
+ final URL ddlTestFile = Db2TestBase.class.getClassLoader().getResource(ddlFile);
+ assertNotNull("Cannot locate " + ddlFile, ddlTestFile);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ if (checkTableExists(connection, tableName)) {
+ LOG.info("{} table exist", tableName);
+ dropTestTable(connection, tableName.toUpperCase(Locale.ROOT));
+ // sleep 10 seconds to make sure ASN replication agent has been notified
+ Thread.sleep(10_000);
+ }
+ final List<String> statements =
+ Arrays.stream(
+ Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
+ .map(String::trim)
+ .filter(x -> !x.startsWith("--") && !x.isEmpty())
+ .map(
+ x -> {
+ final Matcher m =
+ COMMENT_PATTERN.matcher(x);
+ return m.matches() ? m.group(1) : x;
+ })
+ .collect(Collectors.joining("\n"))
+ .split(";"))
+ .collect(Collectors.toList());
+ for (String stmt : statements) {
+ statement.execute(stmt);
+ Thread.sleep(500);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static void assertEqualsInAnyOrder(List<String> expected, List<String> actual) {
+ assertTrue(expected != null && actual != null);
+ assertEqualsInOrder(
+ expected.stream().sorted().collect(Collectors.toList()),
+ actual.stream().sorted().collect(Collectors.toList()));
+ }
+
+ public static void assertEqualsInOrder(List<String> expected, List<String> actual) {
+ assertTrue(expected != null && actual != null);
+ assertEquals(expected.size(), actual.size());
+ assertArrayEquals(expected.toArray(new String[0]), actual.toArray(new String[0]));
+ }
+
+ public void executeSql(String sql) {
+ try (Connection connection = getJdbcConnection()) {
+ connection.createStatement().execute(sql);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public String getTableNameRegex(String[] captureCustomerTables) {
+ checkState(captureCustomerTables.length > 0);
+ if (captureCustomerTables.length == 1) {
+ return captureCustomerTables[0];
+ } else {
+ // pattern that matches multiple tables
+ return format("(%s)", StringUtils.join(captureCustomerTables, ","));
+ }
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceITCase.java
new file mode 100644
index 000000000..c3221aed8
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/Db2SourceITCase.java
@@ -0,0 +1,333 @@
+/*
+ * 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.connectors.db2.source;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.cdc.connectors.db2.Db2TestBase;
+import org.apache.flink.cdc.connectors.db2.source.Db2SourceBuilder.Db2IncrementalSource;
+import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.RpcServiceSharing;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+import static java.lang.String.format;
+import static org.testcontainers.containers.Db2Container.DB2_PORT;
+
+/** IT tests for {@link Db2IncrementalSource}. */
+public class Db2SourceITCase extends Db2TestBase {
+
+ @Rule public final Timeout timeoutPerTest = Timeout.seconds(300);
+
+ @Rule
+ public final MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+ .setRpcServiceSharing(RpcServiceSharing.DEDICATED)
+ .withHaLeadershipControl()
+ .build());
+
+ protected static final int DEFAULT_PARALLELISM = 4;
+
+ @Test
+ public void testReadSingleTableWithSingleParallelism() throws Exception {
+ testDb2ParallelSource(
+ 1, FailoverType.NONE, FailoverPhase.NEVER, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ @Test
+ public void testReadSingleTableWithMultipleParallelism() throws Exception {
+ testDb2ParallelSource(
+ 4, FailoverType.NONE, FailoverPhase.NEVER, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ // Failover tests
+ @Test
+ public void testTaskManagerFailoverInSnapshotPhase() throws Exception {
+ testDb2ParallelSource(
+ FailoverType.TM, FailoverPhase.SNAPSHOT, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ @Test
+ public void testTaskManagerFailoverInRedoLogsPhase() throws Exception {
+ testDb2ParallelSource(
+ FailoverType.TM, FailoverPhase.STREAM, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ @Test
+ public void testJobManagerFailoverInSnapshotPhase() throws Exception {
+ testDb2ParallelSource(
+ FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ @Test
+ public void testJobManagerFailoverInRedoLogsPhase() throws Exception {
+ testDb2ParallelSource(
+ FailoverType.JM, FailoverPhase.STREAM, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ @Test
+ public void testJobManagerFailoverSingleParallelism() throws Exception {
+ testDb2ParallelSource(
+ 1, FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {"DB2INST1.CUSTOMERS"});
+ }
+
+ @Test
+ public void testReadSingleTableWithSingleParallelismAndSkipBackfill() throws Exception {
+ testDb2ParallelSource(
+ DEFAULT_PARALLELISM,
+ FailoverType.TM,
+ FailoverPhase.SNAPSHOT,
+ new String[] {"DB2INST1.CUSTOMERS"},
+ true);
+ }
+
+ private void testDb2ParallelSource(
+ FailoverType failoverType, FailoverPhase failoverPhase, String[] captureCustomerTables)
+ throws Exception {
+ testDb2ParallelSource(
+ DEFAULT_PARALLELISM, failoverType, failoverPhase, captureCustomerTables);
+ }
+
+ private void testDb2ParallelSource(
+ int parallelism,
+ FailoverType failoverType,
+ FailoverPhase failoverPhase,
+ String[] captureCustomerTables)
+ throws Exception {
+ testDb2ParallelSource(
+ parallelism, failoverType, failoverPhase, captureCustomerTables, false);
+ }
+
+ private void testDb2ParallelSource(
+ int parallelism,
+ FailoverType failoverType,
+ FailoverPhase failoverPhase,
+ String[] captureCustomerTables,
+ boolean skipSnapshotBackfill)
+ throws Exception {
+
+ initializeDb2Table("customers", "CUSTOMERS");
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ env.setParallelism(parallelism);
+ env.enableCheckpointing(1000L);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 0));
+
+ String sourceDDL =
+ format(
+ "CREATE TABLE CUSTOMERS ("
+ + " ID INT NOT NULL,"
+ + " NAME STRING,"
+ + " ADDRESS STRING,"
+ + " PHONE_NUMBER STRING,"
+ + " primary key (ID) not enforced"
+ + ") WITH ("
+ + " 'connector' = 'db2-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = 'true',"
+ + " 'scan.incremental.snapshot.chunk.size' = '4',"
+ + " 'scan.incremental.snapshot.backfill.skip' = '%s'"
+ + ")",
+ DB2_CONTAINER.getHost(),
+ DB2_CONTAINER.getMappedPort(DB2_PORT),
+ DB2_CONTAINER.getUsername(),
+ DB2_CONTAINER.getPassword(),
+ DB2_CONTAINER.getDatabaseName(),
+ getTableNameRegex(captureCustomerTables),
+ skipSnapshotBackfill);
+
+ // first step: check the snapshot data
+ String[] snapshotForSingleTable =
+ new String[] {
+ "+I[101, user_1, Shanghai, 123567891234]",
+ "+I[102, user_2, Shanghai, 123567891234]",
+ "+I[103, user_3, Shanghai, 123567891234]",
+ "+I[109, user_4, Shanghai, 123567891234]",
+ "+I[110, user_5, Shanghai, 123567891234]",
+ "+I[111, user_6, Shanghai, 123567891234]",
+ "+I[118, user_7, Shanghai, 123567891234]",
+ "+I[121, user_8, Shanghai, 123567891234]",
+ "+I[123, user_9, Shanghai, 123567891234]",
+ "+I[1009, user_10, Shanghai, 123567891234]",
+ "+I[1010, user_11, Shanghai, 123567891234]",
+ "+I[1011, user_12, Shanghai, 123567891234]",
+ "+I[1012, user_13, Shanghai, 123567891234]",
+ "+I[1013, user_14, Shanghai, 123567891234]",
+ "+I[1014, user_15, Shanghai, 123567891234]",
+ "+I[1015, user_16, Shanghai, 123567891234]",
+ "+I[1016, user_17, Shanghai, 123567891234]",
+ "+I[1017, user_18, Shanghai, 123567891234]",
+ "+I[1018, user_19, Shanghai, 123567891234]",
+ "+I[1019, user_20, Shanghai, 123567891234]",
+ "+I[2000, user_21, Shanghai, 123567891234]"
+ };
+ tEnv.executeSql(sourceDDL);
+ TableResult tableResult = tEnv.executeSql("select * from CUSTOMERS");
+ CloseableIterator<Row> iterator = tableResult.collect();
+ JobID jobId = tableResult.getJobClient().get().getJobID();
+ List<String> expectedSnapshotData = new ArrayList<>();
+ for (int i = 0; i < captureCustomerTables.length; i++) {
+ expectedSnapshotData.addAll(Arrays.asList(snapshotForSingleTable));
+ }
+
+ // trigger failover after some snapshot splits read finished
+ if (failoverPhase == FailoverPhase.SNAPSHOT && iterator.hasNext()) {
+ triggerFailover(
+ failoverType, jobId, miniClusterResource.getMiniCluster(), () -> sleepMs(100));
+ }
+
+ assertEqualsInAnyOrder(
+ expectedSnapshotData, fetchRows(iterator, expectedSnapshotData.size()));
+
+ // second step: check the change stream data
+ for (String tableId : captureCustomerTables) {
+ makeFirstPartChangeStreamEvents(tableId);
+ }
+ if (failoverPhase == FailoverPhase.STREAM) {
+ triggerFailover(
+ failoverType, jobId, miniClusterResource.getMiniCluster(), () -> sleepMs(200));
+ }
+ for (String tableId : captureCustomerTables) {
+ makeSecondPartRedoLogsEvents(tableId);
+ }
+
+ String[] redoLogsForSingleTable =
+ new String[] {
+ "-U[103, user_3, Shanghai, 123567891234]",
+ "+U[103, user_3, Hangzhou, 123567891234]",
+ "-D[102, user_2, Shanghai, 123567891234]",
+ "+I[102, user_2, Shanghai, 123567891234]",
+ "-U[103, user_3, Hangzhou, 123567891234]",
+ "+U[103, user_3, Shanghai, 123567891234]",
+ "-U[1010, user_11, Shanghai, 123567891234]",
+ "+U[1010, user_11, Hangzhou, 123567891234]",
+ "+I[2001, user_22, Shanghai, 123567891234]",
+ "+I[2002, user_23, Shanghai, 123567891234]",
+ "+I[2003, user_24, Shanghai, 123567891234]"
+ };
+ List<String> expectedRedoLogsData = new ArrayList<>();
+ for (int i = 0; i < captureCustomerTables.length; i++) {
+ expectedRedoLogsData.addAll(Arrays.asList(redoLogsForSingleTable));
+ }
+ assertEqualsInAnyOrder(
+ expectedRedoLogsData, fetchRows(iterator, expectedRedoLogsData.size()));
+ tableResult.getJobClient().get().cancel().get();
+ }
+
+ private void makeFirstPartChangeStreamEvents(String tableId) {
+ executeSql("UPDATE " + tableId + " SET ADDRESS = 'Hangzhou' where ID = 103");
+ executeSql("DELETE FROM " + tableId + " where ID = 102");
+ executeSql("INSERT INTO " + tableId + " VALUES(102, 'user_2','Shanghai','123567891234')");
+ executeSql("UPDATE " + tableId + " SET ADDRESS = 'Shanghai' where ID = 103");
+ }
+
+ private void makeSecondPartRedoLogsEvents(String tableId) {
+ executeSql("UPDATE " + tableId + " SET ADDRESS = 'Hangzhou' where ID = 1010");
+ executeSql("INSERT INTO " + tableId + " VALUES(2001, 'user_22','Shanghai','123567891234')");
+ executeSql("INSERT INTO " + tableId + " VALUES(2002, 'user_23','Shanghai','123567891234')");
+ executeSql("INSERT INTO " + tableId + " VALUES(2003, 'user_24','Shanghai','123567891234')");
+ }
+
+ private void sleepMs(long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException ignored) {
+ }
+ }
+
+ private static List<String> fetchRows(Iterator<Row> iter, int size) {
+ List<String> rows = new ArrayList<>(size);
+ while (size > 0 && iter.hasNext()) {
+ Row row = iter.next();
+ rows.add(row.toString());
+ size--;
+ }
+ return rows;
+ }
+
+ /** The type of failover. */
+ protected enum FailoverType {
+ TM,
+ JM,
+ NONE
+ }
+
+ /** The phase of failover. */
+ protected enum FailoverPhase {
+ SNAPSHOT,
+ STREAM,
+ NEVER
+ }
+
+ protected static void triggerFailover(
+ FailoverType type, JobID jobId, MiniCluster miniCluster, Runnable afterFailAction)
+ throws Exception {
+ switch (type) {
+ case TM:
+ restartTaskManager(miniCluster, afterFailAction);
+ break;
+ case JM:
+ triggerJobManagerFailover(jobId, miniCluster, afterFailAction);
+ break;
+ case NONE:
+ break;
+ default:
+ throw new IllegalStateException("Unexpected value: " + type);
+ }
+ }
+
+ protected static void triggerJobManagerFailover(
+ JobID jobId, MiniCluster miniCluster, Runnable afterFailAction) throws Exception {
+ final HaLeadershipControl haLeadershipControl = miniCluster.getHaLeadershipControl().get();
+ haLeadershipControl.revokeJobMasterLeadership(jobId).get();
+ afterFailAction.run();
+ haLeadershipControl.grantJobMasterLeadership(jobId).get();
+ }
+
+ protected static void restartTaskManager(MiniCluster miniCluster, Runnable afterFailAction)
+ throws Exception {
+ miniCluster.terminateTaskManager(0).get();
+ afterFailAction.run();
+ miniCluster.startTaskManager();
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2ConnectorITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2ConnectorITCase.java
index 40e855931..7bf504c1d 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2ConnectorITCase.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2ConnectorITCase.java
@@ -17,17 +17,24 @@
package org.apache.flink.cdc.connectors.db2.table;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.cdc.connectors.db2.Db2TestBase;
+import org.apache.flink.runtime.minicluster.RpcServiceSharing;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.planner.factories.TestValuesTableFactory;
import org.apache.flink.table.utils.LegacyRowResource;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.junit.Before;
import org.junit.ClassRule;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,9 +53,23 @@ import static org.junit.Assert.assertThat;
import static org.testcontainers.containers.Db2Container.DB2_PORT;
/** Integration tests for DB2 CDC source. */
+@RunWith(Parameterized.class)
public class Db2ConnectorITCase extends Db2TestBase {
+
private static final Logger LOG = LoggerFactory.getLogger(Db2ConnectorITCase.class);
+ protected static final int DEFAULT_PARALLELISM = 2;
+
+ @Rule
+ public final MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+ .setRpcServiceSharing(RpcServiceSharing.DEDICATED)
+ .withHaLeadershipControl()
+ .build());
+
private final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
private final StreamTableEnvironment tEnv =
@@ -57,10 +78,27 @@ public class Db2ConnectorITCase extends Db2TestBase {
@ClassRule public static LegacyRowResource usesLegacyRows = LegacyRowResource.INSTANCE;
+ private final boolean incrementalSnapshot;
+
+ public Db2ConnectorITCase(boolean incrementalSnapshot) {
+ this.incrementalSnapshot = incrementalSnapshot;
+ }
+
+ @Parameterized.Parameters(name = "incrementalSnapshot: {0}")
+ public static Object[] parameters() {
+ return new Object[][] {new Object[] {true}, new Object[] {false}};
+ }
+
@Before
public void before() {
TestValuesTableFactory.clearAllData();
- env.setParallelism(1);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ if (incrementalSnapshot) {
+ env.setParallelism(DEFAULT_PARALLELISM);
+ env.enableCheckpointing(1000);
+ } else {
+ env.setParallelism(1);
+ }
}
private void cancelJobIfRunning(TableResult result)
@@ -75,6 +113,7 @@ public class Db2ConnectorITCase extends Db2TestBase {
@Test
public void testConsumingAllEvents()
throws SQLException, InterruptedException, ExecutionException {
+ initializeDb2Table("inventory", "PRODUCTS");
String sourceDDL =
String.format(
"CREATE TABLE debezium_source ("
@@ -89,16 +128,16 @@ public class Db2ConnectorITCase extends Db2TestBase {
+ " 'username' = '%s',"
+ " 'password' = '%s',"
+ " 'database-name' = '%s',"
- + " 'schema-name' = '%s',"
- + " 'table-name' = '%s'"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = '%s'"
+ ")",
DB2_CONTAINER.getHost(),
DB2_CONTAINER.getMappedPort(DB2_PORT),
DB2_CONTAINER.getUsername(),
DB2_CONTAINER.getPassword(),
DB2_CONTAINER.getDatabaseName(),
- "DB2INST1",
- "PRODUCTS");
+ "DB2INST1.PRODUCTS",
+ incrementalSnapshot);
String sinkDDL =
"CREATE TABLE sink ("
+ " name STRING,"
@@ -178,6 +217,7 @@ public class Db2ConnectorITCase extends Db2TestBase {
@Test
public void testAllTypes() throws Exception {
+ initializeDb2Table("column_type_test", "FULL_TYPES");
// NOTE: db2 is not case sensitive by default, the schema returned by debezium
// is uppercase, thus we need use uppercase when defines a db2 table.
String sourceDDL =
@@ -201,7 +241,8 @@ public class Db2ConnectorITCase extends Db2TestBase {
+ " DATE_C DATE,\n"
+ " TIME_C TIME(0),\n"
+ " DEFAULT_NUMERIC_C DECIMAL,\n"
- + " TIMESTAMP_PRECISION_C TIMESTAMP(9)\n"
+ + " TIMESTAMP_PRECISION_C TIMESTAMP(9),\n"
+ + " PRIMARY KEY (ID) NOT ENFORCED"
+ ") WITH ("
+ " 'connector' = 'db2-cdc',"
+ " 'hostname' = '%s',"
@@ -209,16 +250,16 @@ public class Db2ConnectorITCase extends Db2TestBase {
+ " 'username' = '%s',"
+ " 'password' = '%s',"
+ " 'database-name' = '%s',"
- + " 'schema-name' = '%s',"
- + " 'table-name' = '%s'"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = '%s'"
+ ")",
DB2_CONTAINER.getHost(),
DB2_CONTAINER.getMappedPort(DB2_PORT),
DB2_CONTAINER.getUsername(),
DB2_CONTAINER.getPassword(),
DB2_CONTAINER.getDatabaseName(),
- "DB2INST1",
- "FULL_TYPES");
+ "DB2INST1.FULL_TYPES",
+ incrementalSnapshot);
String sinkDDL =
"CREATE TABLE sink (\n"
+ " id INTEGER NOT NULL,\n"
@@ -236,7 +277,8 @@ public class Db2ConnectorITCase extends Db2TestBase {
+ " date_c DATE,\n"
+ " time_c TIME(0),\n"
+ " default_numeric_c DECIMAL,\n"
- + " timestamp_precision_c TIMESTAMP(9)\n"
+ + " timestamp_precision_c TIMESTAMP(9),\n"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ ") WITH ("
+ " 'connector' = 'values',"
+ " 'sink-insert-only' = 'false'"
@@ -254,28 +296,30 @@ public class Db2ConnectorITCase extends Db2TestBase {
statement.execute("UPDATE DB2INST1.FULL_TYPES SET SMALL_C=0 WHERE ID=1;");
}
- waitForSinkSize("sink", 3);
+ waitForSinkSize("sink", 2);
List<String> expected =
Arrays.asList(
"+I(1,32767,65535,2147483647,5.5,6.6,123.12345,404.4,Hello World,a,abc,2020-07-17T18:00:22.123,2020-07-17,18:00:22,500,2020-07-17T18:00:22.123456789)",
- "-U(1,32767,65535,2147483647,5.5,6.6,123.12345,404.4,Hello World,a,abc,2020-07-17T18:00:22.123,2020-07-17,18:00:22,500,2020-07-17T18:00:22.123456789)",
"+U(1,0,65535,2147483647,5.5,6.6,123.12345,404.4,Hello World,a,abc,2020-07-17T18:00:22.123,2020-07-17,18:00:22,500,2020-07-17T18:00:22.123456789)");
List<String> actual = TestValuesTableFactory.getRawResults("sink");
+ Collections.sort(expected);
+ Collections.sort(actual);
assertEquals(expected, actual);
-
cancelJobIfRunning(result);
}
@Test
public void testStartupFromLatestOffset() throws Exception {
+ initializeDb2Table("inventory", "PRODUCTS");
String sourceDDL =
String.format(
"CREATE TABLE debezium_source ("
+ " ID INT NOT NULL,"
+ " NAME STRING,"
+ " DESCRIPTION STRING,"
- + " WEIGHT DECIMAL(10,3)"
+ + " WEIGHT DECIMAL(10,3),"
+ + " PRIMARY KEY (ID) NOT ENFORCED"
+ ") WITH ("
+ " 'connector' = 'db2-cdc',"
+ " 'hostname' = '%s',"
@@ -283,17 +327,17 @@ public class Db2ConnectorITCase extends Db2TestBase {
+ " 'username' = '%s',"
+ " 'password' = '%s',"
+ " 'database-name' = '%s',"
- + " 'schema-name' = '%s',"
+ " 'table-name' = '%s' ,"
- + " 'scan.startup.mode' = 'latest-offset'"
+ + " 'scan.startup.mode' = 'latest-offset',"
+ + " 'scan.incremental.snapshot.enabled' = '%s'"
+ ")",
DB2_CONTAINER.getHost(),
DB2_CONTAINER.getMappedPort(DB2_PORT),
DB2_CONTAINER.getUsername(),
DB2_CONTAINER.getPassword(),
DB2_CONTAINER.getDatabaseName(),
- "DB2INST1",
- "PRODUCTS1");
+ "DB2INST1.PRODUCTS",
+ incrementalSnapshot);
String sinkDDL =
"CREATE TABLE sink "
+ " WITH ("
@@ -315,16 +359,16 @@ public class Db2ConnectorITCase extends Db2TestBase {
try (Connection connection = getJdbcConnection();
Statement statement = connection.createStatement()) {
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS1 VALUES (default,'jacket','water resistent white wind breaker',0.2)");
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (default,'jacket','water resistent white wind breaker',0.2)");
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS1 VALUES (default,'scooter','Big 2-wheel scooter ',5.18)");
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (default,'scooter','Big 2-wheel scooter ',5.18)");
statement.execute(
- "UPDATE DB2INST1.PRODUCTS1 SET DESCRIPTION='new water resistent white wind breaker', WEIGHT='0.5' WHERE ID=110");
- statement.execute("UPDATE DB2INST1.PRODUCTS1 SET WEIGHT='5.17' WHERE ID=111");
- statement.execute("DELETE FROM DB2INST1.PRODUCTS1 WHERE ID=111");
+ "UPDATE DB2INST1.PRODUCTS SET DESCRIPTION='new water resistent white wind breaker', WEIGHT='0.5' WHERE ID=110");
+ statement.execute("UPDATE DB2INST1.PRODUCTS SET WEIGHT='5.17' WHERE ID=111");
+ statement.execute("DELETE FROM DB2INST1.PRODUCTS WHERE ID=111");
}
- waitForSinkSize("sink", 7);
+ waitForSinkSize("sink", 5);
String[] expected =
new String[] {"110,jacket,new water resistent white wind breaker,0.500"};
@@ -337,6 +381,7 @@ public class Db2ConnectorITCase extends Db2TestBase {
@Test
public void testMetadataColumns() throws Throwable {
+ initializeDb2Table("inventory", "PRODUCTS");
String sourceDDL =
String.format(
"CREATE TABLE debezium_source ("
@@ -355,16 +400,16 @@ public class Db2ConnectorITCase extends Db2TestBase {
+ " 'username' = '%s',"
+ " 'password' = '%s',"
+ " 'database-name' = '%s',"
- + " 'schema-name' = '%s',"
- + " 'table-name' = '%s'"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = '%s'"
+ ")",
DB2_CONTAINER.getHost(),
DB2_CONTAINER.getMappedPort(DB2_PORT),
DB2_CONTAINER.getUsername(),
DB2_CONTAINER.getPassword(),
DB2_CONTAINER.getDatabaseName(),
- "DB2INST1",
- "PRODUCTS2");
+ "DB2INST1.PRODUCTS",
+ incrementalSnapshot);
String sinkDDL =
"CREATE TABLE sink ("
+ " database_name STRING,"
@@ -392,38 +437,38 @@ public class Db2ConnectorITCase extends Db2TestBase {
Statement statement = connection.createStatement()) {
statement.execute(
- "UPDATE DB2INST1.PRODUCTS2 SET DESCRIPTION='18oz carpenter hammer' WHERE ID=106;");
- statement.execute("UPDATE DB2INST1.PRODUCTS2 SET WEIGHT='5.1' WHERE ID=107;");
+ "UPDATE DB2INST1.PRODUCTS SET DESCRIPTION='18oz carpenter hammer' WHERE ID=106;");
+ statement.execute("UPDATE DB2INST1.PRODUCTS SET WEIGHT='5.1' WHERE ID=107;");
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS2 VALUES (110,'jacket','water resistent white wind breaker',0.2);");
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (110,'jacket','water resistent white wind breaker',0.2);");
statement.execute(
- "INSERT INTO DB2INST1.PRODUCTS2 VALUES (111,'scooter','Big 2-wheel scooter ',5.18);");
+ "INSERT INTO DB2INST1.PRODUCTS VALUES (111,'scooter','Big 2-wheel scooter ',5.18);");
statement.execute(
- "UPDATE DB2INST1.PRODUCTS2 SET DESCRIPTION='new water resistent white wind breaker', WEIGHT='0.5' WHERE ID=110;");
- statement.execute("UPDATE DB2INST1.PRODUCTS2 SET WEIGHT='5.17' WHERE ID=111;");
- statement.execute("DELETE FROM DB2INST1.PRODUCTS2 WHERE ID=111;");
+ "UPDATE DB2INST1.PRODUCTS SET DESCRIPTION='new water resistent white wind breaker', WEIGHT='0.5' WHERE ID=110;");
+ statement.execute("UPDATE DB2INST1.PRODUCTS SET WEIGHT='5.17' WHERE ID=111;");
+ statement.execute("DELETE FROM DB2INST1.PRODUCTS WHERE ID=111;");
}
waitForSinkSize("sink", 16);
List<String> expected =
Arrays.asList(
- "+I(testdb,DB2INST1,PRODUCTS2,101,scooter,Small 2-wheel scooter,3.140)",
- "+I(testdb,DB2INST1,PRODUCTS2,102,car battery,12V car battery,8.100)",
- "+I(testdb,DB2INST1,PRODUCTS2,103,12-pack drill bits,12-pack of drill bits with sizes ranging from #40 to #3,0.800)",
- "+I(testdb,DB2INST1,PRODUCTS2,104,hammer,12oz carpenter's hammer,0.750)",
- "+I(testdb,DB2INST1,PRODUCTS2,105,hammer,14oz carpenter's hammer,0.875)",
- "+I(testdb,DB2INST1,PRODUCTS2,106,hammer,16oz carpenter's hammer,1.000)",
- "+I(testdb,DB2INST1,PRODUCTS2,107,rocks,box of assorted rocks,5.300)",
- "+I(testdb,DB2INST1,PRODUCTS2,108,jacket,water resistent black wind breaker,0.100)",
- "+I(testdb,DB2INST1,PRODUCTS2,109,spare tire,24 inch spare tire,22.200)",
- "+U(testdb,DB2INST1,PRODUCTS2,106,hammer,18oz carpenter hammer,1.000)",
- "+U(testdb,DB2INST1,PRODUCTS2,107,rocks,box of assorted rocks,5.100)",
- "+I(testdb,DB2INST1,PRODUCTS2,110,jacket,water resistent white wind breaker,0.200)",
- "+I(testdb,DB2INST1,PRODUCTS2,111,scooter,Big 2-wheel scooter ,5.180)",
- "+U(testdb,DB2INST1,PRODUCTS2,110,jacket,new water resistent white wind breaker,0.500)",
- "+U(testdb,DB2INST1,PRODUCTS2,111,scooter,Big 2-wheel scooter ,5.170)",
- "-D(testdb,DB2INST1,PRODUCTS2,111,scooter,Big 2-wheel scooter ,5.170)");
+ "+I(testdb,DB2INST1,PRODUCTS,101,scooter,Small 2-wheel scooter,3.140)",
+ "+I(testdb,DB2INST1,PRODUCTS,102,car battery,12V car battery,8.100)",
+ "+I(testdb,DB2INST1,PRODUCTS,103,12-pack drill bits,12-pack of drill bits with sizes ranging from #40 to #3,0.800)",
+ "+I(testdb,DB2INST1,PRODUCTS,104,hammer,12oz carpenter's hammer,0.750)",
+ "+I(testdb,DB2INST1,PRODUCTS,105,hammer,14oz carpenter's hammer,0.875)",
+ "+I(testdb,DB2INST1,PRODUCTS,106,hammer,16oz carpenter's hammer,1.000)",
+ "+I(testdb,DB2INST1,PRODUCTS,107,rocks,box of assorted rocks,5.300)",
+ "+I(testdb,DB2INST1,PRODUCTS,108,jacket,water resistent black wind breaker,0.100)",
+ "+I(testdb,DB2INST1,PRODUCTS,109,spare tire,24 inch spare tire,22.200)",
+ "+U(testdb,DB2INST1,PRODUCTS,106,hammer,18oz carpenter hammer,1.000)",
+ "+U(testdb,DB2INST1,PRODUCTS,107,rocks,box of assorted rocks,5.100)",
+ "+I(testdb,DB2INST1,PRODUCTS,110,jacket,water resistent white wind breaker,0.200)",
+ "+I(testdb,DB2INST1,PRODUCTS,111,scooter,Big 2-wheel scooter ,5.180)",
+ "+U(testdb,DB2INST1,PRODUCTS,110,jacket,new water resistent white wind breaker,0.500)",
+ "+U(testdb,DB2INST1,PRODUCTS,111,scooter,Big 2-wheel scooter ,5.170)",
+ "-D(testdb,DB2INST1,PRODUCTS,111,scooter,Big 2-wheel scooter ,5.170)");
List<String> actual = TestValuesTableFactory.getRawResults("sink");
Collections.sort(expected);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java
index 7d4852a37..ff1e2eada 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/table/Db2TableSourceFactoryTest.java
@@ -17,7 +17,9 @@
package org.apache.flink.cdc.connectors.db2.table;
-import org.apache.flink.cdc.debezium.DebeziumSourceFunction;
+import org.apache.flink.cdc.connectors.base.options.JdbcSourceOptions;
+import org.apache.flink.cdc.connectors.base.options.SourceOptions;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.DataTypes;
@@ -28,12 +30,8 @@ import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.connector.source.DynamicTableSource;
-import org.apache.flink.table.connector.source.ScanTableSource;
-import org.apache.flink.table.connector.source.SourceFunctionProvider;
-import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.Factory;
import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
import org.apache.flink.util.ExceptionUtils;
import org.junit.Test;
@@ -46,7 +44,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
-import static org.apache.flink.cdc.connectors.utils.AssertUtils.assertProducedTypeOfSourceFunction;
import static org.apache.flink.cdc.debezium.utils.ResolvedSchemaUtils.getPhysicalSchema;
import static org.apache.flink.table.api.TableSchema.fromResolvedSchema;
import static org.junit.Assert.assertEquals;
@@ -87,8 +84,7 @@ public class Db2TableSourceFactoryTest {
private static final String MY_USERNAME = "flinkuser";
private static final String MY_PASSWORD = "flinkpw";
private static final String MY_DATABASE = "myDB";
- private static final String MY_SCHEMA = "flinkuser";
- private static final String MY_TABLE = "myTable";
+ private static final String MY_TABLE = "flinkuser.myTable";
private static final Properties PROPERTIES = new Properties();
@Test
@@ -103,13 +99,26 @@ public class Db2TableSourceFactoryTest {
50000,
MY_LOCALHOST,
MY_DATABASE,
- MY_SCHEMA,
MY_TABLE,
MY_USERNAME,
MY_PASSWORD,
ZoneId.of("UTC"),
PROPERTIES,
- StartupOptions.initial());
+ StartupOptions.initial(),
+ SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED.defaultValue(),
+ SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE.defaultValue(),
+ SourceOptions.CHUNK_META_GROUP_SIZE.defaultValue(),
+ SourceOptions.SCAN_SNAPSHOT_FETCH_SIZE.defaultValue(),
+ JdbcSourceOptions.CONNECT_TIMEOUT.defaultValue(),
+ JdbcSourceOptions.CONNECT_MAX_RETRIES.defaultValue(),
+ JdbcSourceOptions.CONNECTION_POOL_SIZE.defaultValue(),
+ JdbcSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND
+ .defaultValue(),
+ JdbcSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND
+ .defaultValue(),
+ null,
+ false,
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -129,13 +138,26 @@ public class Db2TableSourceFactoryTest {
50000,
MY_LOCALHOST,
MY_DATABASE,
- MY_SCHEMA,
MY_TABLE,
MY_USERNAME,
MY_PASSWORD,
ZoneId.of("Asia/Shanghai"),
dbzProperties,
- StartupOptions.latest());
+ StartupOptions.latest(),
+ SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED.defaultValue(),
+ SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE.defaultValue(),
+ SourceOptions.CHUNK_META_GROUP_SIZE.defaultValue(),
+ SourceOptions.SCAN_SNAPSHOT_FETCH_SIZE.defaultValue(),
+ JdbcSourceOptions.CONNECT_TIMEOUT.defaultValue(),
+ JdbcSourceOptions.CONNECT_MAX_RETRIES.defaultValue(),
+ JdbcSourceOptions.CONNECTION_POOL_SIZE.defaultValue(),
+ JdbcSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND
+ .defaultValue(),
+ JdbcSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND
+ .defaultValue(),
+ null,
+ false,
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
assertEquals(expectedSource, actualSource);
}
@@ -204,25 +226,31 @@ public class Db2TableSourceFactoryTest {
50000,
MY_LOCALHOST,
MY_DATABASE,
- MY_SCHEMA,
MY_TABLE,
MY_USERNAME,
MY_PASSWORD,
ZoneId.of("UTC"),
new Properties(),
- StartupOptions.initial());
+ StartupOptions.initial(),
+ SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED.defaultValue(),
+ SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE.defaultValue(),
+ SourceOptions.CHUNK_META_GROUP_SIZE.defaultValue(),
+ SourceOptions.SCAN_SNAPSHOT_FETCH_SIZE.defaultValue(),
+ JdbcSourceOptions.CONNECT_TIMEOUT.defaultValue(),
+ JdbcSourceOptions.CONNECT_MAX_RETRIES.defaultValue(),
+ JdbcSourceOptions.CONNECTION_POOL_SIZE.defaultValue(),
+ JdbcSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND
+ .defaultValue(),
+ JdbcSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND
+ .defaultValue(),
+ null,
+ false,
+ JdbcSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue());
expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType();
expectedSource.metadataKeys =
Arrays.asList("op_ts", "database_name", "table_name", "schema_name");
assertEquals(expectedSource, actualSource);
-
- ScanTableSource.ScanRuntimeProvider provider =
- db2TableSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE);
- DebeziumSourceFunction<RowData> debeziumSourceFunction =
- (DebeziumSourceFunction<RowData>)
- ((SourceFunctionProvider) provider).createSourceFunction();
- assertProducedTypeOfSourceFunction(debeziumSourceFunction, expectedSource.producedDataType);
}
private Map<String, String> getAllOptions() {
@@ -230,7 +258,6 @@ public class Db2TableSourceFactoryTest {
options.put("connector", "db2-cdc");
options.put("hostname", MY_LOCALHOST);
options.put("database-name", MY_DATABASE);
- options.put("schema-name", MY_SCHEMA);
options.put("table-name", MY_TABLE);
options.put("username", MY_USERNAME);
options.put("password", MY_PASSWORD);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/testutils/RecordsFormatter.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/testutils/RecordsFormatter.java
new file mode 100644
index 000000000..73b73bd0a
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/testutils/RecordsFormatter.java
@@ -0,0 +1,97 @@
+/*
+ * 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.connectors.db2.testutils;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils;
+import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
+import org.apache.flink.cdc.debezium.table.RowDataDebeziumDeserializeSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.conversion.RowRowConverter;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+import org.apache.kafka.connect.source.SourceRecord;
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Formatter that formats the {@link SourceRecord} to String. */
+public class RecordsFormatter {
+
+ private final TypeInformation<RowData> typeInfo;
+ private final DebeziumDeserializationSchema<RowData> deserializationSchema;
+ private final SimpleCollector collector;
+ private final RowRowConverter rowRowConverter;
+
+ public RecordsFormatter(DataType dataType) {
+ this(dataType, ZoneId.of("UTC"));
+ }
+
+ public RecordsFormatter(DataType dataType, ZoneId zoneId) {
+ this.typeInfo =
+ (TypeInformation<RowData>) TypeConversions.fromDataTypeToLegacyInfo(dataType);
+ this.deserializationSchema =
+ RowDataDebeziumDeserializeSchema.newBuilder()
+ .setPhysicalRowType((RowType) dataType.getLogicalType())
+ .setResultTypeInfo(typeInfo)
+ .setServerTimeZone(zoneId)
+ .build();
+ this.collector = new SimpleCollector();
+ this.rowRowConverter = RowRowConverter.create(dataType);
+ rowRowConverter.open(Thread.currentThread().getContextClassLoader());
+ }
+
+ public List<String> format(List<SourceRecord> records) {
+ records.stream()
+ // Keep DataChangeEvent only
+ .filter(SourceRecordUtils::isDataChangeRecord)
+ .forEach(
+ r -> {
+ try {
+ deserializationSchema.deserialize(r, collector);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ return collector.list.stream()
+ .map(rowRowConverter::toExternal)
+ .map(Row::toString)
+ .collect(Collectors.toList());
+ }
+
+ private static class SimpleCollector implements Collector<RowData> {
+
+ private List<RowData> list = new ArrayList<>();
+
+ @Override
+ public void collect(RowData record) {
+ list.add(record);
+ }
+
+ @Override
+ public void close() {
+ // do nothing
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/testutils/TestTable.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/testutils/TestTable.java
new file mode 100644
index 000000000..2ed36a17c
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/testutils/TestTable.java
@@ -0,0 +1,96 @@
+/*
+ * 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.connectors.db2.testutils;
+
+import org.apache.flink.cdc.debezium.table.RowDataDebeziumDeserializeSchema;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.conversion.RowRowConverter;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.kafka.connect.source.SourceRecord;
+
+import java.util.List;
+
+/**
+ * Test utility for creating converter, formatter and deserializer of a table in the test database.
+ */
+public class TestTable {
+
+ private final String databaseName;
+ private final String tableName;
+
+ private final String schemaName;
+
+ private final ResolvedSchema schema;
+
+ // Lazily initialized components
+ private RowRowConverter rowRowConverter;
+ private RowDataDebeziumDeserializeSchema deserializer;
+ private RecordsFormatter recordsFormatter;
+
+ public TestTable(
+ String databaseName, String schemaName, String tableName, ResolvedSchema schema) {
+ this.databaseName = databaseName;
+ this.schemaName = schemaName;
+ this.tableName = tableName;
+ this.schema = schema;
+ }
+
+ public String getTableId() {
+ return String.format("%s.%s.%s", databaseName, schemaName, tableName);
+ }
+
+ public RowType getRowType() {
+ return (RowType) schema.toPhysicalRowDataType().getLogicalType();
+ }
+
+ public RowDataDebeziumDeserializeSchema getDeserializer() {
+ if (deserializer == null) {
+ deserializer =
+ RowDataDebeziumDeserializeSchema.newBuilder()
+ .setPhysicalRowType(getRowType())
+ .setResultTypeInfo(InternalTypeInfo.of(getRowType()))
+ .build();
+ }
+ return deserializer;
+ }
+
+ public RowRowConverter getRowRowConverter() {
+ if (rowRowConverter == null) {
+ rowRowConverter = RowRowConverter.create(schema.toPhysicalRowDataType());
+ }
+ return rowRowConverter;
+ }
+
+ public RecordsFormatter getRecordsFormatter() {
+ if (recordsFormatter == null) {
+ recordsFormatter = new RecordsFormatter(schema.toPhysicalRowDataType());
+ }
+ return recordsFormatter;
+ }
+
+ public String stringify(RowData rowData) {
+ return getRowRowConverter().toExternal(rowData).toString();
+ }
+
+ public List<String> stringify(List<SourceRecord> sourceRecord) {
+ return getRecordsFormatter().format(sourceRecord);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/Dockerfile b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/Dockerfile
index d3f34edb9..fd8db80d3 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/Dockerfile
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/Dockerfile
@@ -25,7 +25,6 @@ ADD asncdcaddremove.sql /asncdctools/src
ADD asncdctables.sql /asncdctools/src
ADD dbsetup.sh /asncdctools/src
ADD startup-agent.sql /asncdctools/src
-ADD startup-cdc-demo.sql /asncdctools/src
ADD inventory.sql /asncdctools/src
ADD column_type_test.sql /asncdctools/src
ADD asncdc.c /asncdctools/src
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/asncdctables.sql b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/asncdctables.sql
index ba18877b4..45960636b 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/asncdctables.sql
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/asncdctables.sql
@@ -158,7 +158,7 @@ COMPATIBILITY)
VALUES (
10080,
10080,
-30,
+5,
300,
10080,
10080,
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/column_type_test.sql b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/column_type_test.sql
index b1412a108..b21e33ed5 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/column_type_test.sql
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/column_type_test.sql
@@ -37,8 +37,13 @@ CREATE TABLE DB2INST1.FULL_TYPES (
PRIMARY KEY (ID)
);
-
INSERT INTO DB2INST1.FULL_TYPES VALUES (
1, 32767, 65535, 2147483647, 5.5, 6.6, 123.12345, 404.4443,
'Hello World', 'a', 'abc', '2020-07-17 18:00:22.123', '2020-07-17', '18:00:22', 500,
'2020-07-17 18:00:22.123456789');
+
+VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
+
+CALL ASNCDC.ADDTABLE('DB2INST1', 'FULL_TYPES');
+
+VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/customers.sql b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/customers.sql
new file mode 100644
index 000000000..e40d5acdc
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/customers.sql
@@ -0,0 +1,48 @@
+-- Copyright 2023 Ververica Inc.
+--
+-- Licensed 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.
+
+CREATE TABLE DB2INST1.CUSTOMERS (
+ ID INTEGER NOT NULL PRIMARY KEY,
+ NAME VARCHAR(255) NOT NULL DEFAULT 'flink',
+ ADDRESS VARCHAR(1024),
+ PHONE_NUMBER VARCHAR(512)
+);
+
+INSERT INTO DB2INST1.CUSTOMERS
+VALUES (101,'user_1','Shanghai','123567891234'),
+ (102,'user_2','Shanghai','123567891234'),
+ (103,'user_3','Shanghai','123567891234'),
+ (109,'user_4','Shanghai','123567891234'),
+ (110,'user_5','Shanghai','123567891234'),
+ (111,'user_6','Shanghai','123567891234'),
+ (118,'user_7','Shanghai','123567891234'),
+ (121,'user_8','Shanghai','123567891234'),
+ (123,'user_9','Shanghai','123567891234'),
+ (1009,'user_10','Shanghai','123567891234'),
+ (1010,'user_11','Shanghai','123567891234'),
+ (1011,'user_12','Shanghai','123567891234'),
+ (1012,'user_13','Shanghai','123567891234'),
+ (1013,'user_14','Shanghai','123567891234'),
+ (1014,'user_15','Shanghai','123567891234'),
+ (1015,'user_16','Shanghai','123567891234'),
+ (1016,'user_17','Shanghai','123567891234'),
+ (1017,'user_18','Shanghai','123567891234'),
+ (1018,'user_19','Shanghai','123567891234'),
+ (1019,'user_20','Shanghai','123567891234'),
+ (2000,'user_21','Shanghai','123567891234');
+
+VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
+
+CALL ASNCDC.ADDTABLE('DB2INST1', 'CUSTOMERS');
+
+VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
\ No newline at end of file
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/dbsetup.sh b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/dbsetup.sh
index bad0f9950..44ad8e5d9 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/dbsetup.sh
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/dbsetup.sh
@@ -58,12 +58,8 @@ db2 -tvmf /asncdctools/src/asncdcaddremove.sql
-# create sample table and data
-db2 -tvmf /asncdctools/src/inventory.sql
-db2 -tvmf /asncdctools/src/column_type_test.sql
+# startup-agent
db2 -tvmf /asncdctools/src/startup-agent.sql
-sleep 10
-db2 -tvmf /asncdctools/src/startup-cdc-demo.sql
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/inventory.sql b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/inventory.sql
index 08c0c1115..a665d96fc 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/inventory.sql
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/inventory.sql
@@ -22,22 +22,6 @@ CREATE TABLE DB2INST1.PRODUCTS (
WEIGHT FLOAT
);
-CREATE TABLE DB2INST1.PRODUCTS1 (
- ID INTEGER NOT NULL GENERATED BY DEFAULT AS IDENTITY
- (START WITH 101, INCREMENT BY 1) PRIMARY KEY,
- NAME VARCHAR(255) NOT NULL,
- DESCRIPTION VARCHAR(512),
- WEIGHT FLOAT
-);
-
-CREATE TABLE DB2INST1.PRODUCTS2 (
- ID INTEGER NOT NULL GENERATED BY DEFAULT AS IDENTITY
- (START WITH 101, INCREMENT BY 1) PRIMARY KEY,
- NAME VARCHAR(255) NOT NULL,
- DESCRIPTION VARCHAR(512),
- WEIGHT FLOAT
-);
-
INSERT INTO DB2INST1.PRODUCTS(NAME,DESCRIPTION,WEIGHT)
VALUES ('scooter','Small 2-wheel scooter',3.14),
('car battery','12V car battery',8.1),
@@ -49,24 +33,8 @@ VALUES ('scooter','Small 2-wheel scooter',3.14),
('jacket','water resistent black wind breaker',0.1),
('spare tire','24 inch spare tire',22.2);
-INSERT INTO DB2INST1.PRODUCTS1(NAME,DESCRIPTION,WEIGHT)
-VALUES ('scooter','Small 2-wheel scooter',3.14),
- ('car battery','12V car battery',8.1),
- ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8),
- ('hammer','12oz carpenter''s hammer',0.75),
- ('hammer','14oz carpenter''s hammer',0.875),
- ('hammer','16oz carpenter''s hammer',1.0),
- ('rocks','box of assorted rocks',5.3),
- ('jacket','water resistent black wind breaker',0.1),
- ('spare tire','24 inch spare tire',22.2);
+VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
-INSERT INTO DB2INST1.PRODUCTS2(NAME,DESCRIPTION,WEIGHT)
-VALUES ('scooter','Small 2-wheel scooter',3.14),
- ('car battery','12V car battery',8.1),
- ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8),
- ('hammer','12oz carpenter''s hammer',0.75),
- ('hammer','14oz carpenter''s hammer',0.875),
- ('hammer','16oz carpenter''s hammer',1.0),
- ('rocks','box of assorted rocks',5.3),
- ('jacket','water resistent black wind breaker',0.1),
- ('spare tire','24 inch spare tire',22.2);
+CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS');
+
+VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
\ No newline at end of file
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/startup-cdc-demo.sql b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/startup-cdc-demo.sql
deleted file mode 100644
index 0c9a60820..000000000
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/db2_server/startup-cdc-demo.sql
+++ /dev/null
@@ -1,23 +0,0 @@
--- 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.
-
-VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
-
-CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS');
-CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS1');
-CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS2');
-CALL ASNCDC.ADDTABLE('DB2INST1', 'FULL_TYPES');
-
-VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/log4j2-test.properties b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/log4j2-test.properties
index 11c65fe08..f0d32fb59 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/log4j2-test.properties
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/resources/log4j2-test.properties
@@ -15,11 +15,11 @@
# Set root logger level to OFF to not flood build logs
# set manually to INFO for debugging purposes
-rootLogger.level=OFF
+rootLogger.level=INFO
rootLogger.appenderRef.test.ref = TestLogger
appender.testlogger.name = TestLogger
appender.testlogger.type = CONSOLE
-appender.testlogger.target = SYSTEM_OUT
+appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
appender.testlogger.layout.pattern = %-4r [%t] %-5p %c - %m%n
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceTestBase.java
index 8682e871a..0ca776beb 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceTestBase.java
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceTestBase.java
@@ -73,7 +73,9 @@ public abstract class MySqlSourceTestBase extends TestLogger {
@AfterClass
public static void stopContainers() {
LOG.info("Stopping containers...");
- MYSQL_CONTAINER.stop();
+ if (MYSQL_CONTAINER != null) {
+ MYSQL_CONTAINER.stop();
+ }
LOG.info("Containers are stopped.");
}
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-sql-connector-db2-cdc/pom.xml b/flink-cdc-connect/flink-cdc-source-connectors/flink-sql-connector-db2-cdc/pom.xml
index 1dd4f8bbc..34b5ada07 100644
--- a/flink-cdc-connect/flink-cdc-source-connectors/flink-sql-connector-db2-cdc/pom.xml
+++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-sql-connector-db2-cdc/pom.xml
@@ -46,7 +46,7 @@ limitations under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>3.1.1</version>
+ <version>3.2.4</version>
<executions>
<execution>
<id>shade-flink</id>
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/org/apache/flink/cdc/connectors/tests/Db2E2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/org/apache/flink/cdc/connectors/tests/Db2E2eITCase.java
index 9094da69e..29ecd1e59 100644
--- a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/org/apache/flink/cdc/connectors/tests/Db2E2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/java/org/apache/flink/cdc/connectors/tests/Db2E2eITCase.java
@@ -19,30 +19,43 @@ package org.apache.flink.cdc.connectors.tests;
import org.apache.flink.cdc.common.test.utils.JdbcProxy;
import org.apache.flink.cdc.common.test.utils.TestUtils;
+import org.apache.flink.cdc.connectors.db2.Db2TestBase;
import org.apache.flink.cdc.connectors.tests.utils.FlinkContainerTestEnvironment;
+import org.awaitility.Awaitility;
import org.junit.After;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
+import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Db2Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.images.builder.ImageFromDockerfile;
import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.shaded.org.awaitility.core.ConditionTimeoutException;
import org.testcontainers.utility.DockerImageName;
import java.net.URISyntaxException;
import java.net.URL;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.junit.Assert.assertNotNull;
@@ -60,22 +73,41 @@ public class Db2E2eITCase extends FlinkContainerTestEnvironment {
public static final String DB2_IMAGE = "ibmcom/db2";
public static final String DB2_CUSTOM_IMAGE = "custom/db2-cdc:1.4";
- private static final DockerImageName DEBEZIUM_DOCKER_IMAGE_NAME =
- DockerImageName.parse(
- new ImageFromDockerfile(DB2_CUSTOM_IMAGE)
- .withDockerfile(getFilePath("docker/db2/Dockerfile"))
- .get())
- .asCompatibleSubstituteFor(DB2_IMAGE);
+ private static DockerImageName debeziumDockerImageName;
private static boolean db2AsnAgentRunning = false;
-
private static Db2Container db2Container;
+ private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");
+
+ @Parameterized.Parameter(1)
+ public boolean parallelismSnapshot;
+
+ @Parameterized.Parameters(name = "flinkVersion: {0}, parallelismSnapshot: {1}")
+ public static List<Object[]> parameters() {
+ final List<String> flinkVersions = getFlinkVersion();
+ List<Object[]> params = new ArrayList<>();
+ for (String flinkVersion : flinkVersions) {
+ params.add(new Object[] {flinkVersion, true});
+ params.add(new Object[] {flinkVersion, false});
+ }
+ return params;
+ }
+
+ @BeforeClass
+ public static void beforeClass() {
+ debeziumDockerImageName =
+ DockerImageName.parse(
+ new ImageFromDockerfile(DB2_CUSTOM_IMAGE)
+ .withDockerfile(getFilePath("docker/db2/Dockerfile"))
+ .get())
+ .asCompatibleSubstituteFor(DB2_IMAGE);
+ }
@Before
public void before() {
super.before();
LOG.info("Starting db2 containers...");
db2Container =
- new Db2Container(DEBEZIUM_DOCKER_IMAGE_NAME)
+ new Db2Container(debeziumDockerImageName)
.withDatabaseName("testdb")
.withUsername("db2inst1")
.withPassword("flinkpw")
@@ -118,6 +150,7 @@ public class Db2E2eITCase extends FlinkContainerTestEnvironment {
@Test
public void testDb2CDC() throws Exception {
+ initializeDb2Table("inventory", "PRODUCTS");
List<String> sqlLines =
Arrays.asList(
String.format(
@@ -125,7 +158,8 @@ public class Db2E2eITCase extends FlinkContainerTestEnvironment {
+ " ID INT NOT NULL,"
+ " NAME STRING,"
+ " DESCRIPTION STRING,"
- + " WEIGHT DECIMAL(10,3)"
+ + " WEIGHT DECIMAL(10,3),"
+ + " primary key (`ID`) not enforced"
+ ") WITH ("
+ " 'connector' = 'db2-cdc',"
+ " 'hostname' = '%s',"
@@ -133,16 +167,18 @@ public class Db2E2eITCase extends FlinkContainerTestEnvironment {
+ " 'username' = '%s',"
+ " 'password' = '%s',"
+ " 'database-name' = '%s',"
- + " 'schema-name' = '%s',"
- + " 'table-name' = '%s'"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' = '"
+ + parallelismSnapshot
+ + "',"
+ + " 'scan.incremental.snapshot.chunk.size' = '4'"
+ ");",
INTER_CONTAINER_DB2_ALIAS,
DB2_PORT,
db2Container.getUsername(),
db2Container.getPassword(),
db2Container.getDatabaseName(),
- "DB2INST1",
- "PRODUCTS"),
+ "DB2INST1.PRODUCTS"),
"CREATE TABLE products_sink (",
" `id` INT NOT NULL,",
" name STRING,",
@@ -227,4 +263,102 @@ public class Db2E2eITCase extends FlinkContainerTestEnvironment {
}
return path;
}
+
+ private static void dropTestTable(Connection connection, String tableName) {
+
+ try {
+ Awaitility.await(String.format("cdc remove table %s", tableName))
+ .atMost(60, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ String sql =
+ String.format(
+ "CALL ASNCDC.REMOVETABLE('DB2INST1', '%s')",
+ tableName);
+ connection.createStatement().execute(sql);
+ return true;
+ } catch (SQLException e) {
+ LOG.warn(
+ String.format(
+ "cdc remove TABLE %s failed (will be retried): {}",
+ tableName),
+ e.getMessage());
+ return false;
+ }
+ });
+ } catch (ConditionTimeoutException e) {
+ throw new IllegalStateException("Failed to cdc remove test table", e);
+ }
+
+ try {
+ Awaitility.await(String.format("Dropping table %s", tableName))
+ .atMost(60, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ String sql = String.format("DROP TABLE DB2INST1.%s", tableName);
+ connection.createStatement().execute(sql);
+ connection.commit();
+ return true;
+ } catch (SQLException e) {
+ LOG.warn(
+ String.format(
+ "DROP TABLE %s failed (will be retried): {}",
+ tableName),
+ e.getMessage());
+ return false;
+ }
+ });
+ } catch (ConditionTimeoutException e) {
+ throw new IllegalStateException("Failed to drop test database", e);
+ }
+ }
+
+ /**
+ * Executes a JDBC statement using the default jdbc config without autocommitting the
+ * connection.
+ */
+ protected void initializeDb2Table(String sqlFile, String tableName) {
+ final String ddlFile = String.format("docker/db2/%s.sql", sqlFile);
+ final URL ddlTestFile = Db2TestBase.class.getClassLoader().getResource(ddlFile);
+ assertNotNull("Cannot locate " + ddlFile, ddlTestFile);
+ try (Connection connection = getDb2Connection();
+ Statement statement = connection.createStatement()) {
+ String tableExistSql =
+ String.format(
+ "SELECT COUNT(*) FROM SYSCAT.TABLES WHERE TABNAME = '%s' AND "
+ + "TABSCHEMA = 'DB2INST1';",
+ tableName);
+ ResultSet resultSet = statement.executeQuery(tableExistSql);
+ int count = 0;
+ if (resultSet.next()) {
+ count = resultSet.getInt(1);
+ }
+ if (count == 1) {
+ LOG.info("{} table exist", tableName);
+ dropTestTable(connection, tableName.toUpperCase(Locale.ROOT));
+ }
+ final List<String> statements =
+ Arrays.stream(
+ Files.readAllLines(Paths.get(ddlTestFile.toURI())).stream()
+ .map(String::trim)
+ .filter(x -> !x.startsWith("--") && !x.isEmpty())
+ .map(
+ x -> {
+ final Matcher m =
+ COMMENT_PATTERN.matcher(x);
+ return m.matches() ? m.group(1) : x;
+ })
+ .collect(Collectors.joining("\n"))
+ .split(";"))
+ .collect(Collectors.toList());
+ for (String stmt : statements) {
+ statement.execute(stmt);
+ Thread.sleep(500);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/Dockerfile b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/Dockerfile
index d3f34edb9..fd8db80d3 100644
--- a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/Dockerfile
+++ b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/Dockerfile
@@ -25,7 +25,6 @@ ADD asncdcaddremove.sql /asncdctools/src
ADD asncdctables.sql /asncdctools/src
ADD dbsetup.sh /asncdctools/src
ADD startup-agent.sql /asncdctools/src
-ADD startup-cdc-demo.sql /asncdctools/src
ADD inventory.sql /asncdctools/src
ADD column_type_test.sql /asncdctools/src
ADD asncdc.c /asncdctools/src
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/column_type_test.sql b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/column_type_test.sql
index b5d2b6273..b21e33ed5 100644
--- a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/column_type_test.sql
+++ b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/column_type_test.sql
@@ -33,10 +33,17 @@ CREATE TABLE DB2INST1.FULL_TYPES (
DATE_C DATE,
TIME_C TIME,
DEFAULT_NUMERIC_C NUMERIC,
+ TIMESTAMP_PRECISION_C TIMESTAMP(9),
PRIMARY KEY (ID)
);
-
INSERT INTO DB2INST1.FULL_TYPES VALUES (
1, 32767, 65535, 2147483647, 5.5, 6.6, 123.12345, 404.4443,
- 'Hello World', 'a', 'abc', '2020-07-17 18:00:22.123', '2020-07-17', '18:00:22', 500);
+ 'Hello World', 'a', 'abc', '2020-07-17 18:00:22.123', '2020-07-17', '18:00:22', 500,
+ '2020-07-17 18:00:22.123456789');
+
+VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
+
+CALL ASNCDC.ADDTABLE('DB2INST1', 'FULL_TYPES');
+
+VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/customers.sql b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/customers.sql
new file mode 100644
index 000000000..e40d5acdc
--- /dev/null
+++ b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/customers.sql
@@ -0,0 +1,48 @@
+-- Copyright 2023 Ververica Inc.
+--
+-- Licensed 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.
+
+CREATE TABLE DB2INST1.CUSTOMERS (
+ ID INTEGER NOT NULL PRIMARY KEY,
+ NAME VARCHAR(255) NOT NULL DEFAULT 'flink',
+ ADDRESS VARCHAR(1024),
+ PHONE_NUMBER VARCHAR(512)
+);
+
+INSERT INTO DB2INST1.CUSTOMERS
+VALUES (101,'user_1','Shanghai','123567891234'),
+ (102,'user_2','Shanghai','123567891234'),
+ (103,'user_3','Shanghai','123567891234'),
+ (109,'user_4','Shanghai','123567891234'),
+ (110,'user_5','Shanghai','123567891234'),
+ (111,'user_6','Shanghai','123567891234'),
+ (118,'user_7','Shanghai','123567891234'),
+ (121,'user_8','Shanghai','123567891234'),
+ (123,'user_9','Shanghai','123567891234'),
+ (1009,'user_10','Shanghai','123567891234'),
+ (1010,'user_11','Shanghai','123567891234'),
+ (1011,'user_12','Shanghai','123567891234'),
+ (1012,'user_13','Shanghai','123567891234'),
+ (1013,'user_14','Shanghai','123567891234'),
+ (1014,'user_15','Shanghai','123567891234'),
+ (1015,'user_16','Shanghai','123567891234'),
+ (1016,'user_17','Shanghai','123567891234'),
+ (1017,'user_18','Shanghai','123567891234'),
+ (1018,'user_19','Shanghai','123567891234'),
+ (1019,'user_20','Shanghai','123567891234'),
+ (2000,'user_21','Shanghai','123567891234');
+
+VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
+
+CALL ASNCDC.ADDTABLE('DB2INST1', 'CUSTOMERS');
+
+VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
\ No newline at end of file
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/dbsetup.sh b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/dbsetup.sh
index ebda64467..44ad8e5d9 100644
--- a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/dbsetup.sh
+++ b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/dbsetup.sh
@@ -56,11 +56,12 @@ db2 -tvmf /asncdctools/src/asncdctables.sql
db2 -tvmf /asncdctools/src/asncdcaddremove.sql
-# create sample table and data
-db2 -tvmf /asncdctools/src/inventory.sql
-db2 -tvmf /asncdctools/src/column_type_test.sql
+
+
+# startup-agent
db2 -tvmf /asncdctools/src/startup-agent.sql
-sleep 10
-db2 -tvmf /asncdctools/src/startup-cdc-demo.sql
+
+
+
echo "db2 setup done"
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/inventory.sql b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/inventory.sql
index 08c0c1115..a665d96fc 100644
--- a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/inventory.sql
+++ b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/inventory.sql
@@ -22,22 +22,6 @@ CREATE TABLE DB2INST1.PRODUCTS (
WEIGHT FLOAT
);
-CREATE TABLE DB2INST1.PRODUCTS1 (
- ID INTEGER NOT NULL GENERATED BY DEFAULT AS IDENTITY
- (START WITH 101, INCREMENT BY 1) PRIMARY KEY,
- NAME VARCHAR(255) NOT NULL,
- DESCRIPTION VARCHAR(512),
- WEIGHT FLOAT
-);
-
-CREATE TABLE DB2INST1.PRODUCTS2 (
- ID INTEGER NOT NULL GENERATED BY DEFAULT AS IDENTITY
- (START WITH 101, INCREMENT BY 1) PRIMARY KEY,
- NAME VARCHAR(255) NOT NULL,
- DESCRIPTION VARCHAR(512),
- WEIGHT FLOAT
-);
-
INSERT INTO DB2INST1.PRODUCTS(NAME,DESCRIPTION,WEIGHT)
VALUES ('scooter','Small 2-wheel scooter',3.14),
('car battery','12V car battery',8.1),
@@ -49,24 +33,8 @@ VALUES ('scooter','Small 2-wheel scooter',3.14),
('jacket','water resistent black wind breaker',0.1),
('spare tire','24 inch spare tire',22.2);
-INSERT INTO DB2INST1.PRODUCTS1(NAME,DESCRIPTION,WEIGHT)
-VALUES ('scooter','Small 2-wheel scooter',3.14),
- ('car battery','12V car battery',8.1),
- ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8),
- ('hammer','12oz carpenter''s hammer',0.75),
- ('hammer','14oz carpenter''s hammer',0.875),
- ('hammer','16oz carpenter''s hammer',1.0),
- ('rocks','box of assorted rocks',5.3),
- ('jacket','water resistent black wind breaker',0.1),
- ('spare tire','24 inch spare tire',22.2);
+VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
-INSERT INTO DB2INST1.PRODUCTS2(NAME,DESCRIPTION,WEIGHT)
-VALUES ('scooter','Small 2-wheel scooter',3.14),
- ('car battery','12V car battery',8.1),
- ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8),
- ('hammer','12oz carpenter''s hammer',0.75),
- ('hammer','14oz carpenter''s hammer',0.875),
- ('hammer','16oz carpenter''s hammer',1.0),
- ('rocks','box of assorted rocks',5.3),
- ('jacket','water resistent black wind breaker',0.1),
- ('spare tire','24 inch spare tire',22.2);
+CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS');
+
+VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');
\ No newline at end of file
diff --git a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/startup-cdc-demo.sql b/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/startup-cdc-demo.sql
deleted file mode 100644
index 0c9a60820..000000000
--- a/flink-cdc-e2e-tests/flink-cdc-source-e2e-tests/src/test/resources/docker/db2/startup-cdc-demo.sql
+++ /dev/null
@@ -1,23 +0,0 @@
--- 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.
-
-VALUES ASNCDC.ASNCDCSERVICES('status','asncdc');
-
-CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS');
-CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS1');
-CALL ASNCDC.ADDTABLE('DB2INST1', 'PRODUCTS2');
-CALL ASNCDC.ADDTABLE('DB2INST1', 'FULL_TYPES');
-
-VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');