[FLINK-36605] Bump debezium to last 2.x version

[eskabetxe]

What is the purpose of this pull request?

Bumps the Debezium dependency from 1.9.8.Final to 2.7.4.Final — FLINK-36605.

The upgrade is 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 160-file diff.

2.7.4 is deliberately the end point. Debezium 3.x requires Java 17/21, which would force raising the minimum Java version and evaluating 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

160 files changed, 8580 insertions(+), 6961 deletions(-) across 8 commits. Referenced by version rather than SHA, since the branch is rebased as review progresses.

Hop	Size	Headline change and how it was solved
2.0.1	94 files, +1529/−2016	history → SchemaHistory rename (database.history.* → schema.history.internal.*); database.server.name → topic.prefix; TopicSelector → TopicNamingStrategy; DataCollectionId moved package; *Partition ctors gained databaseName. Solved by a tree-wide config-key/import migration plus re-basing the forks onto 2.0.1; several forks deleted because the bug they patched was fixed upstream. database.server.id became .required(), so the MySQL config factory now injects a default.
2.1.4	15 files, +319/−90	Postgres WalPositionLocator gained an Operation arg and dropped the dead doSnapshot param; SqlServer ctor arg → SqlServerJdbcConfiguration; ANTLR ColumnDeclarationContext.uid() → .fullColumnName().uid(). Solved by re-basing the two postgres forks plus one-line owned-code fixes.
2.2.1	31 files, +423/−55	SchemaNameAdjuster moved io.debezium.util → io.debezium.schema (18 files); the SchemaHistory interface shrank; emitter ctors gained args; EmbeddedEngineChangeEvent gained a 4-arg ctor and a third type param. Solved by import migration and fork re-bases.
2.3.7	30 files, +133/−581	OffsetContext threaded through the whole schema-change dispatch path; Kafka Connect 3.5 bump (OffsetBackingStore.connectorPartitions); ErrorHandler gained replacedErrorHandler. Solved by updating both owned dispatchers and every fork call site; the MySqlDefaultValueConverter fork was deleted (DBZ-6824 fixed upstream).
2.4.2	20 files, +536/−115	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.
2.5.4	49 files, +2363/−2275	MySQL "strategy" refactor: MySqlConnection → …mysql.strategy.mysql, GtidSet became an interface, and the GTID merge and binlog-position read moved out of the forks. Separately, 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 at all). Solved by replacing the old MySqlConnection fork with three new ones under strategy.mysql, and declaring jsr305 explicitly in the root pom.
2.6.2	28 files, +749/−388	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 inside BaseSourceTask, which Flink CDC never runs), plus validateLogPosition added to the Postgres and Db2 connection forks.
2.7.4	65 files, +4665/−3578	The debezium-connector-binlog split — the shared binlog engine moves to a new artifact and the MySQL connector becomes thin subclasses. Solved by re-targeting the forks into io.debezium.connector.binlog. Net effect is a smaller fork surface.

Cross-cutting notes

• Forked Debezium classes: 30 → 29. Deleted across the series: MySqlConnection (old location), MySqlDefaultValueConverter, MySqlValueConverters, DefaultValueParserListener, HistorizedRelationalDatabaseConnectorConfig. Added: BinlogSnapshotChangeEventSource, mysql/jdbc/MySqlConnection, mysql/jdbc/MySqlConnectionConfiguration, and oracle/antlr/listener/ColumnDefinitionParserListener.
• 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 using 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-debezium), which builds Debezium 2.6+'s snapshotter service outside the Kafka Connect task lifecycle.

---

Findings that only integration testing caught

Every item below compiled cleanly, passed spotless, passed the unit tests, and passed the javap binary-compatibility audit. They were found only by running the full suite, and each is fixed in the hop that introduced it, so every commit remains independently correct.

This section is the most useful part of the diff to review, because these are the changes that look arbitrary without the explanation.

Hop	Finding	Why it was invisible
2.0.1	SqlServer TLS. Debezium 2.0 pulls mssql-jdbc 10.2 (replacing 9.4), which flipped the encrypt default false → true. SqlServerValidator builds its JDBC URL by hand and could no longer connect.	A transitive driver bump, not an API change.
2.0.1	SqlServer streaming loop. SQL Server is Debezium's only multi-partition connector, so upstream's execute() is a stub that throws UnsupportedOperationException; the real loop lives in SqlServerChangeEventSourceCoordinator, which Flink CDC never runs. The re-fork inherited the throwing stub.	Signature-compatible; only the body changed.
2.0.1	gh-ost / pt-osc renames. Debezium 2.x emits one schema-change event per renamed table for a multi-table RENAME TABLE, so the old table-id matching never fired and the stored ALTER was never re-emitted.	A data-shape change, not an API change.
2.0.1	debezium-storage-kafka missing from the fat jars. Debezium 2.0 moved the Kafka schema history out of debezium-core. Building a historized connector's ConfigDef eagerly resolves the default value of schema.history.internal in ConfigDef$ConfigKey.<init>, so the class must be present even though Flink CDC always overrides it.	Only EmbeddedEngine builds that ConfigDef, and only the legacy DebeziumSourceFunction path uses EmbeddedEngine. Unit tests have the artifact transitively; only the shaded E2E jar did not.
2.1.4	Oracle TIMESTAMP WITH TIME ZONE. 2.1.4 made ZonedTimestamp scale-aware, so it keeps trailing zeros (.00789 → .007890). Test expectation updated; the stored value is unchanged.	Formatting-only upstream change.
2.2.1	Oracle SYS.XMLTYPE NPE. 2.2.1 made FieldNameCache unconditional; that path calls ColumnImpl.equals, which NPEs on the null typeName Oracle's DDL parser has always produced for schema-qualified types. Fixed with a new fork adding an explicit SYS.XMLTYPE → Types.SQLXML branch.	Latent for years; only exposed when the cache stopped being optional.
2.4.2	Vitess protobuf conflict. vitess-client 12 → 17 needs protobuf 3.19+ (GeneratedMessageV3#isStringEmpty), but the test-scoped mysql-connector-java drags in 3.11.4, which wins nearest-wins resolution.	The NoSuchMethodError is raised on gRPC's ThreadlessExecutor, where waitAndDrain swallows it — so the test hangs instead of failing.
2.5.4	effectiveOffsetContext NPE. 2.5 routes offset access through a field populated by init(offsetContext), which Debezium's coordinator calls but Flink CDC does not. The fork patch initialising it was lost in the re-fork.	Deleting a fork patch is invisible to every static check.
2.6.2	SnapshotterService NPE. It resolves the connector-specific snapshotter via Class.forName(config.getString("connector.class")); Flink CDC assembles its Debezium config by hand and never sets that key.	Config-driven reflection.
2.7.4	debezium-connector-binlog missing from the fat jars after the artifact split. Verified against the built jar, confirming both presence and that the fork wins the class overlap.	Shade <includes> are not type-checked.
2.7.4	SqlServer time(4) rounding. Bisected to 2.7.4. An earlier hypothesis blaming the mssql-jdbc version was disproved by a live JDBC probe: drivers 9.4.1 and 12.4.2 behave identically.	Debezium's own read path changed.
2.7.4	Db2 schema-change window. 2.7 widened it from currentMaxLsn to currentMaxLsn.increment(), making it inclusive. Flink CDC's long-standing Db2Connection.getChangeTables() fork reads startLsn from a column tracking capture progress, so startLsn == currentMaxLsn on nearly every poll — the same capture instance is queued repeatedly and migrateTable() re-reads the live table schema each time. After an ALTER TABLE ... ADD COLUMN that installs the new wider schema while the open result set still yields the old column count: Data row is smaller than a column index. The fork restores the exclusive bound, i.e. exactly the pre-2.7 behaviour.	One-token upstream change with a non-local effect.

Deliberately out of scope: the Db2 getChangeTables() column mapping is left unchanged from master. Whether startLsn should come from a different column is a pre-existing Flink CDC question, not something a version bump should decide.

---

Verifying this change

• 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: production GTIDs are always UUIDs.
  • FilterGtidSetTest moved and rewritten to drive MySqlConnection, since 2.5 moved the GTID merge onto the connection.
  • SqlServerEventDeserializerTest — drop(table) → drop(table.id()).
  • OraclePipelineITCase#restartFromSavePoint — see below.
  • SqlServerConnectorITCase#testAllTypes — time(4) expectation, see the findings table.
• Existing integration tests cover the change unmodified. The legacy DataStream ITs (LegacyMySqlSourceITCase, PostgreSQLSourceTest, Db2SourceTest, the [2] variants of the E2E suites) matter most here: they are the only tests that drive the forks through Debezium's own task and coordinator lifecycle, which is where fork mismatches surface. Several findings above were reachable only through that path.

One test change worth calling out

OraclePipelineITCase#testSnapshotModeRestartFromSavepoint slept a fixed 30 s and then called stopWithSavepoint. With StartupOptions.snapshot() the source is Boundedness.BOUNDED, so the job legitimately finishes — and stopWithSavepoint can never succeed on a terminated job. Under 2.x the snapshot completes inside that window.

The test now waits until the job is RUNNING in snapshot mode (unbounded modes keep the original settle time, so the passing testInitialModeRestartFromSavepoint is untouched), and the retry helper fails fast on a terminal state instead of burning 600 retries and reporting a misleading timeout. IncrementalSource.getBoundedness() is not modified by this PR — the behaviour it encodes is the same on master; only the test's assumption was wrong.

⚠️ 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 reorganised the read path (snapshot source → connector adapter → back to snapshot source). Please confirm it is reached on both MySQL 8.0 and 8.4.
2. The two binlog forks shadow classes the MariaDB connector also uses. Flink CDC does not ship debezium-connector-mariadb, so this is contained — but it is a wider blast radius than the previous mysql-only forks and deserves a look.
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, producing a restart loop. Please confirm this is the right call rather than changing the offset we store.
4. Empty NotificationService and SnapshotterService instances. Flink CDC does not run Debezium's Kafka Connect task lifecycle, so these are constructed directly with no notification channels registered. Confirm no notification-dependent behaviour is expected.
5. Two subtleties 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().
   • BinlogStreamingChangeEventSource.execute initialises the effective offset context when it is null, because Flink CDC 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 in FLINK-39149.

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 production Flink CDC code never set the old 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?

• Yes (please specify the tool below)

Generated-by: Claude Code

[jeff-jk]

Thanks for doing this — the one-minor-per-commit staging makes it reviewable in a way a single
160-file diff would not have been, and the 2.7.4 stopping point with the Java 11 rationale matches
what I measured independently.

I built the same migration separately before finding this PR (same 1.9.8 → 2.7.4 target, same Java 11
reasoning), so rather than open a competing PR I would rather hand over what I found. Four items, in
severity order. The first three I can reproduce; the fourth is a judgement call about scope.

1. The Java 11 guarantee does not survive a JDK 17 build machine

This one undercuts the PR's own headline claim, which is why it is first.

The root pom sets target.java.version to 11, but the java-17-target profile auto-activates on
<jdk>[17,)</jdk> and hard-codes the compiler plugin directly:

<profile>
  <id>java-17-target</id>
  <activation><jdk>[17,)</jdk></activation>
  <properties><java.version>17</java.version></properties>
  ... <source>17</source><target>17</target> ...
</profile>

So a release built on a JDK 17 toolchain emits class-file major 61 regardless of the declared
property. I verified this is true on master today and unchanged on this branch. It predates this
PR — but this PR is the one whose argument rests on "we stay on Java 11", so it seems worth pinning
down here or in a companion change.

Fix that worked for me: have both profiles derive <source>/<target> from
${source.java.version}/${target.java.version} and carry only the --add-exports/--add-opens
arguments their JDK actually needs. Then the declared property is authoritative, and
-Dtarget.java.version=17 still works for anyone who wants it.

I am happy to send that as a separate small PR so it does not enlarge this one.

2. specific-offset on the legacy MySQL source cannot start on a GTID-enabled server

Reproducible, and it is a startup hang rather than a failure.

Debezium 2.7 calls client.setGtidSet("") in the "No GTID stored in the offset" branch — on this
branch that is BinlogStreamingChangeEventSource.java:341 in the forked copy. Debezium 1.9.8 set
filename and position and left the client's GTID state alone.

In com.zendesk:mysql-binlog-connector-java:0.29.2, setGtidSet("") sets gtidEnabled = true and,
because the argument is the empty string specifically, skips assigning gtidSet — leaving it null.
gtidEnabled has exactly two putfield sites (the constructor and that setter), so the client
cannot leave GTID mode once latched.

Observed on an equivalent 2.7.4 branch, against gtid_mode=ON: a startup
NullPointerException from DumpBinaryLogGtidCommand#toByteArray, after which the engine logs
"Can't start the connector, will retry later..." every 10 seconds indefinitely — zero records, no
checkpoint progress, and no fast failure. Reproduced twice.

The incremental source is unaffected. earliest-offset and timestamp on the legacy source also
survive; it is specific-offset that breaks.

I fixed it by handing the legacy path a client that refuses an empty/blank GTID set, which is where
the state actually latches and needs no knowledge of which branch Debezium took. I have a test
against a real gtid_mode=ON server that fails without the fix — happy to contribute it here.

3. A fatal binlog-thread error hangs the reader instead of failing the job

MySqlErrorHandler now passes null for Debezium's new fourth ErrorHandler argument
(replacedErrorHandler). I did the same thing initially. The consequence is that a fatal error
raised on the binlog-client thread is recorded on the error handler but never reaches
BinlogSplitReader, which stays blocked in ChangeEventQueue.poll.

flink-cdc-base's IncrementalSourceScanFetcher has its own checkReadException, but I do not see
an equivalent on the MySQL binlog reader path on this branch (getProducerThrowable has no call
sites). In CI this shows up as a job that never finishes rather than a red build, which is
unpleasant to debug — it is how a MySQL test class in my own branch appeared "flaky" for a while.

4. Removed debezium.* passthrough keys are silently ignored (scope judgement)

Users can set arbitrary debezium.* properties, and Debezium's Configuration does not reject
unknown keys. After this bump, a job carrying debezium.database.server.name keeps starting, but the
key no longer does anything — which changes the offset partition key and triggers a full
re-snapshot with duplicate rows rather than an error.

I added a validator that fails fast at job submission for the keys removed between 1.9.8 and 2.7.4,
with the replacement named in the message. Two things I got wrong first time and would flag:

• never, schema_only and schema_only_recovery are all still valid at 2.7.4 (the enum carries
  both spellings), so a 3.x-era list of removed values is wrong here. Only Postgres's exported
  actually disappears between 1.9.8 and 2.7.4.
• database.history.instance.name never existed; the real 1.9 key is database.history.name.

Whether that belongs in this PR or a follow-up is your call — it is a behaviour change, not a bump.

Things you caught that I did not, or did better

• Your PostgresOffsetUtils fix also covers transaction_id and the incremental-snapshot prefix
  keys; mine only handled the transaction-order counters. Yours is the more complete allow-list.
• The jsr305 demotion at 2.5 taking javax.annotation off the compile classpath, including in
  mongodb-cdc which has no Debezium fork at all — I hit that as a mystery build break and did not
  diagnose it as cleanly.
• Staging by minor is simply better than my topical commits for bisecting an upgrade like this.

Happy to send the Java-target fix as its own PR, and to contribute the GTID reproduction and test
here if useful.