Repository navigation
Conversation
|
Hi @suntectec, thanks for working on the DB2 pipeline connector! I’m the author of #4454, which explored the same feature. After comparing the implementations and tracing the shared DB2 source code, I noticed a few issues that may be worth addressing before merging:
Some of these are inherited limitations of the shared DB2 source rather than regressions introduced by this PR. However, they affect the capabilities exposed by the new pipeline connector. Tests covering real DECFLOAT values, ALTER/recapture followed by DML, and checkpoint/restart across schema changes would be particularly valuable. Thanks again for the contribution—I hope these observations help strengthen the implementation! |
Add a DB2 source connector for the YAML pipeline API, reusing the existing DB2 CDC source and IBM SQL Replication. - Forward jdbc.properties.* to Debezium as database.* options, letting explicit debezium.* settings win - Fail fast on unsupported column types (DECFLOAT, XML, CLOB, DBCLOB, BLOB) during schema inference to avoid schema/record mismatch - Exclude the IBM JDBC driver from the shaded artifact per its license - Document options, driver installation, schema-change boundaries and the UPDATE requirement on SQL Replication capture tables - Register the module in the build, CI labeler, module detection and the docs sidebar
fb8f369 to
7c54546
Compare
|
Re: the review above (@Ho1ySh1tC0d1ng) Thanks for the detailed review. All five points are addressed in the reworked branch (rebased on master, single commit
Test status: 26 cases per generation (8 unit + 13 factory IT + 5 pipeline IT) pass on Java 11 / Flink 1.20.3 and Java 17 / Flink 2.2.0, plus Spotless and Checkstyle; the documentation builds for both locales. |
|
@yuxiqian sorry for the direct ping. The three workflow runs ( One more thing: the labeler rule added in this PR expects a |
What is the purpose of this pull request?
Add a DB2 pipeline connector so the YAML Pipeline API can read from IBM DB2, alongside the existing MySQL / Postgres / Oracle / SQL Server pipeline connectors. Before this change DB2 was only reachable from the DataStream / SQL APIs through
flink-connector-db2-cdc.JIRA: FLINK-40508
Brief change log
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-db2, registered in the pipeline-connectors parentpom.xml.Db2DataSourceFactory(identifierdb2) plusDb2DataSource,Db2PipelineSource,Db2EventDeserializer,Db2MetadataAccessor,Db2PipelineRecordEmitter,Db2SchemaDataTypeInference,Db2SchemaUtilsandDb2TypeUtils.scan.startup.modeacceptsinitial(default) andlatest-offset, captured tables are declared asschema.table.jdbc.properties.*are forwarded to Debezium asdatabase.*connector properties, and an explicitly configureddebezium.*property wins. A test readsVALUES CURRENT SCHEMAback from a real connection to prove the properties reach JDBC.DECFLOAT,XML,CLOB,DBCLOB,BLOB) fail fast during schema inference instead of being silently advertised, so schema and record fields cannot drift apart.FLINK_HOME/libof the submitting client and every JM/TM is documented.schema-change.enabledonly forwards schema change events reported by the underlying source and does not enable runtime DDL discovery, checkpoint/savepoint restore does not refresh an outdated schema, and UPDATE events requireCHG_UPD_TO_DEL_INS='Y'on the capture table because directUrecords are not supported..github/workflows/modules.pyadds the module toMODULES_DB2, so CI builds and tests it for Flink 1.20 and Flink 2.x;.github/labeler.ymladds thedb2-pipeline-connectorrule.docs/contentanddocs/i18n/zh, plus the pipeline-connectors overview pages and the Docusaurus sidebar.flink-connector-db2-cdcis not modified.src/test/resources/db2_server/is copied from the source connector becauseDb2TestBaseresolves the Docker build context throughPaths.get(url.toURI()), which cannot read resources out of a test-jar.Verifying this change
This change added tests and can be verified as follows:
Db2TypeUtilsTestcovers the supported mappings, length and precision handling, and the rejection path for unsupported types.Db2DataSourceFactoryITCase(13 cases) runs against a real DB2 instance through Testcontainers, includingjdbc.properties.*/debezium.*precedence.Db2PipelineITCase(5 cases) coversinitialwith and without schema changes, an 18-column full-type table for snapshot plus streaming INSERT/UPDATE/DELETE, and fail-fast on theUNSUPPORTED_TYPESfixture.mvn verifypasses on Java 11 / Flink 1.20.3 and Java 17 / Flink 2.2.0 (-Pflink2): 26 tests per generation, plus Spotless and Checkstyle.npm testandnpm run buildfor both locales.Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: Qoder