eskabetxe opened a new pull request, #4516: URL: https://github.com/apache/flink-cdc/pull/4516
## What is the purpose of this pull request? Bumps the Debezium dependency from **1.9.8.Final to 2.7.4.Final** — [FLINK-36605](https://issues.apache.org/jira/browse/FLINK-36605). The upgrade **staged one minor at a time**, as eight commits. Each commit sets a single `<debezium.version>` and is independently compilable and testable, so the series can be bisected and reviewed hop by hop rather than as one 128-file diff. **2.7.4 is deliberately the end point.** Debezium 3.x requires Java 17/21, which can force the minimal Java version and to evaluate the removal of Flink 1.x. This PR stays on **Java 11 / Flink 1.x** throughout (verified from the bytecode: `debezium-core` and `debezium-connector-binlog` 2.7.4 class files are major version 55, and `debezium-build-parent` declares `jdk.min.version` = 11). ## Brief change log `128 files changed, 6470 insertions(+), 6382 deletions(-)` across 8 commits: | Commit | Version | Size | Headline change and how it was solved | |---|---|---|---| | `d631aafb` | **2.0.1** | 77 files, +852/−1879 | `history` → `SchemaHistory` rename (`database.history.*` → `schema.history.internal.*`); `database.server.name` → `topic.prefix`; `TopicSelector` → `TopicNamingStrategy`; `DataCollectionId` moved package; `*Partition` ctors gained `databaseName`. **Solved by** tree-wide config-key/import migration + re-basing the forks onto 2.0.1; several forks **deleted** because the bug they patched was fixed upstream. `database.server.id` became `.required()` → the MySQL config factory now injects a default. | | `25c95451` | **2.1.4** | 12 files, +183/−84 | Postgres `WalPositionLocator` +`Operation` arg and the dead `doSnapshot` param dropped; SqlServer ctor arg → `SqlServerJdbcConfiguration`; ANTLR `ColumnDeclarationContext.uid()` → `.fullColumnName()`. **Solved by** re-basing the two postgres forks and one-line owned-code fixes. | | `90d6cc1e` | **2.2.1** | 30 files, +94/−55 | `SchemaNameAdjuster` moved `io.debezium.util` → `io.debezium.schema` (18 files); `SchemaHistory` interface shrank; emitter ctors gained args; `EmbeddedEngineChangeEvent` gained a 4-arg ctor and a third type param. **Solved by** import migration + fork re-bases. | | `cb20bba2` | **2.3.7** | 28 files, +125/−581 | `OffsetContext` threaded through the whole schema-change dispatch path; **Kafka Connect 3.5** bump (`OffsetBackingStore.connectorPartitions`); `ErrorHandler` +`replacedErrorHandler`. **Solved by** updating both owned dispatchers and every fork call site; `MySqlDefaultValueConverter` fork **deleted** (DBZ-6824 fixed upstream). | | `4b120931` | **2.4.2** | 17 files, +512/−114 | `ChangeEventSourceContext` gained **5 abstract methods** (blocking snapshot, DBZ-6685); `SnapshottingTask` moved to top level with a 5-arg ctor; snapshot ctors gained `NotificationService`; `TableChanges.drop(Table)` → `drop(TableId)`, with a DROP change now carrying a null `Table`. **Solved by** no-op context methods in 8 implementations, threading the snapshotting task, an **empty** `NotificationService` (not `null` — `PostgresSnapshotSplitReadTask` inherits `execute` and dereferences it), and a null-guard when serializing a DROP. | | `56585ffd` | **2.5.4** | 48 files, +2312/−2269 | MySQL **"strategy" refactor**: `MySqlConnection` → `…mysql.strategy.mysql`, `GtidSet` became an interface, the GTID merge and binlog-position read moved out of the forks. Also: Debezium demoted **jsr305** from `provided` to `runtime`, taking `javax.annotation` off the compile classpath in 4 modules (including `mongodb-cdc`, which has no Debezium fork). **Solved by** replacing the old `MySqlConnection` fork with three new ones in `strategy.mysql`, and declaring jsr305 explicitly in the root pom. | | `3cf89aa8` | **2.6.2** | 26 files, +705/−384 | `SnapshotterService` threaded through every snapshot and streaming source; the Postgres `Snapshotter` SPI **deleted**; `getBlockingSnapshottingTask` went `default` → `abstract`. **Solved by** a new `SnapshotterServiceFactory` helper (Debezium builds this in `BaseSourceTask`, which Flink CDC never runs), plus `validateLogPosition` added to the Postgres/Db2 connection forks. | | `d676bbf8` | **2.7.4** | 40 files, +3657/−2986 | The **`debezium-connector-binlog` split** — the shared binlog engine moves to a new artifact and the MySQL connector shrinks to thin subclasses. **Solved by** re-targeting the forks into `io.debezium.connector.binlog`; net effect is a **smaller** fork surface: three forks deleted, two added. | ### Cross-cutting notes - **Forked Debezium classes: 30 → 29.** Deleted this series: bothtSource` copies, `MySqlConnectorAdapter`,`MySqlDefaultValueConverter`, and several 1.9-era forks whose fixes landed upstream. - **Re-fork method:** copy the new stock file, re-apply the documented Flink patches, run `spotless:apply`. Every fork header records the version it was copied from and the patches it carries. - **Binary-compatibility audit.** At each hop, every forked class was compared against the Debezium jar it shadows with `javap`. This catches the failure mode the compiler cannot: a fork missing a signature that Debezium's own precompiled code calls. It found a real one at 2.5.4 — `SqlServerStreamingChangeEventSource`'s constructor gained a `NotificationService` argument, and the un-updated fork would have compiled cleanly and thrown `NoSuchMethodError` on the legacy embedded-engine path. - **New file:** `SnapshotterServiceFactory` (`flink-connector-debium 2.6+ snapshotter service outside the Kafka Connect tasklifecycle. --- ## Verifying this change This change updates existing tests and is verified by the project's CI suite: - **Updated unit tests** - `BinlogOffsetTest`, `GtidUtilsTest`, `FilterGtidSetTest` — placeholder GTID source ids (`aaa-111`, `A`) replaced with real server UUIDs. From Debezium 2.5 the GTID set delegates parsing to the binlog client, which validates ids with `UUID.fromString`. Test-only change: production GTIDs are always UUIDs. - `FilterGtidSetTest` moved and rewritten to drive `MySqlConnec 2.5 moved the GTID merge onto the connection. - `SqlServerEventDeserializerTest` — `drop(table)` → `drop(table.id())`. - **Existing integration tests** cover the change unmodified; the are the relevant ones, along with the legacy DataStream ITs(`LegacyMySqlSourceITCase`, `PostgreSQLSourceTest`), which are the only tests that drive the forks through Debezium's own task and coordinator lifecycle — the path where fork signature mismatches surface. ### ⚠️ Reviewer checklist — decisions and behaviours to confirm 1. **MySQL 8.4 support relies on a probe.** `MySqlConnection` probes `SHOW BINARY LOG STATUS` once on construction and falls back to `SHOW MASTER STATUS`. The probe site moved three times across the series as Debezium ren read (snapshot source → connector adapter → back to snapshotsource). Please confirm it is reached on both MySQL 8.0 and 8.4. 2. **The two binlog forks shadow classes the MariaDB connector al`debezium-connector-mariadb`, so this is contained — but it is a wider blast radius than the previous mysql-only forks and deserve 3. **`PostgresReplicationConnection.validateSlotIsInExpectedState` is overridden to a no-op** (since 2.1.4). Debezium calls `pg_replication_slot_advance(slot, storedLsn)` from `startStreaming`; Flink CDC drives the start position from its own incremental-snapshot offsets, so the stored LSN is not a valid advance target and the seek throws `invrestart loop. Please confirm this is the right call rather thanchanging the offset we store. 4. **Empty `NotificationService` and `SnapshotterService` instancun Debezium's Kafka Connect task lifecycle, so these areconstructed directly, with no notification channels registered. Confirm no notification-dependent behaviour is expected. 5. **Two subtleties that are invisible to the compiler**, both fixed in the commit that introduced them — worth understanding, since they explain code that would otherwise look redundant: - `SnapshotterServiceFactory` builds its **own** service registry with `connector.class` set, rather than reusing `connectorConfig.getServiceRegistry()`. Debezium resolves connector-specific snapshot implementations via `Class.forName(config.getString("connector.class"))`, and Flink CDC builds its own Debezium config without that key. - `BinlogStreamingChangeEventSource.execute` initialises the effective offset context when it is null. Debezium's coordinator calls `init(offsetContext)` before `execute(...)`; Flink CDC drives the source directly and never calls `init`. ### ⚠️ Needs a maintainer decision: MySQL GTID new-channel-position Debezium 2.0 removed the `MySqlConnectorConfig.GTID_NEW_CHANNEL_POSITION` enum that selected between the **EARLIEST** reconciliation (`GtidUtils.fixOldChannelsGtidSet`) and the **LATEST** path added This PR drives the choice from a **pass-through property** `gtid.new.channel.position`, defaulting to `earliest` — which matches upstream Debezium 2.0 and **preserves the exact pre-bump runtime behaviour**, since product enum. **The question:** is it acceptable for the `LATEST` path (FLINK-39149) to be reachable only via the raw Debezium property, or should Flink CDC expose a first-class connector option for it? This is an API-surface question, not a correctness bug. *Suggested reviewers: the FLINK-39149 author(s) and MySQL connector maintainers.* ## Documentation - Does this pull request introduce a new feature? **no** - If yes, how is the feature documented? **not applicable** Dependency upgrade only. No public connector option is added or removed. One behavioural surface changes shape — the MySQL GTID new-channel-position selector, see the decision item above — and `docs/` may need a note if maintainers choose to expose it as a first-class option. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
