[Feature][Connector-V2] Refactoring to have per-connector version management of Debezium

[vsantonastaso]

Purpose of this pull request

This PR introduces an abstraction layer for Debezium version management
across all CDC connectors, enabling independent Debezium version upgrades
per connector without forcing a monolithic upgrade across the entire CDC
module.
This PR implements an adapter pattern with Service Provider Interface (SPI)
that:

• Decouples SeaTunnel CDC framework from specific Debezium versions
• Allows each connector to declare its own debezium.version property
• Provides version-agnostic interfaces (DebeziumAdapter,
  DebeziumEventDispatcher, DebeziumSchemaHistory)
• Enables future independent connector upgrades (e.g., PostgreSQL → 2.7.4
  while MySQL stays on 1.9.8)

Implementation Details:

1. New Abstraction Interfaces (in connector-cdc-base):

   • DebeziumAdapter - SPI for version-specific implementations
   • DebeziumEventDispatcher<P> - Version-agnostic event dispatcher
   • DebeziumSchemaHistory - Version-agnostic schema history
   • DebeziumTopicNaming<T> - Version-agnostic topic naming
   • TableChangeInfo - Version-independent table schema representation
   • DebeziumEventDispatcherConfig - Builder for dispatcher configuration

2. Per-Connector Adapters (created for each database):

   • PostgreSQL: PostgresDebeziumAdapter,
     PostgresEventDispatcherAdapter, PostgresSchemaHistoryAdapter
   • MySQL: MySqlDebeziumAdapter, MySqlEventDispatcherAdapter,
     MySqlSchemaHistoryAdapter
   • Oracle: OracleDebeziumAdapter, OracleEventDispatcherAdapter,
     OracleSchemaHistoryAdapter
   • SQL Server: SqlServerDebeziumAdapter,
     SqlServerEventDispatcherAdapter, SqlServerSchemaHistoryAdapter
   • MongoDB: Version property added (uses different architecture, no
     adapter needed)

3. ServiceLoader SPI Registration:

   • Each connector registers its adapter via META-INF/services/org.apache .seatunnel.connectors.cdc.base.debezium.DebeziumAdapter
   • Runtime discovery using
     DebeziumAdapterFactory.getAdapter("postgres")

4. Refactored Components:

   • JdbcSourceEventDispatcher - Now uses generic partition types
   • All *SourceFetchTaskContext classes - Load adapters and create
     components via SPI
   • Removed direct EmbeddedDatabaseHistory calls - Now use
     DebeziumSchemaHistory abstraction

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[DanielLeens]

Hi @vsantonastaso, I rechecked the current PR head locally as seatunnel-review-10799 at 417b1b84181c. I reviewed the full diff against upstream/dev and did not run local Maven/tests in this batch; this is a source-level review.

This is a broad CDC infrastructure refactor, and the normal paths do hit the new adapter layer:

CDC source reader configure(split)
  -> DebeziumAdapterFactory.getAdapter(connectorType)
  -> connector-specific adapter creates EventDispatcher / SchemaHistory / TopicNaming
  -> JdbcSourceEventDispatcher emits DataChangeEvent into SeaTunnel queue
  -> registerDatabaseHistory(...) seeds EmbeddedDatabaseHistory for snapshot/incremental splits
  -> Debezium streaming/snapshot task continues with connector-specific Debezium classes

The direction is reasonable for per-connector Debezium version management: the PR keeps connector-specific Debezium construction in connector modules and registers adapters through META-INF/services. I did not find a concrete runtime blocker in the current source-level pass.

There are still two review notes I would like to keep visible:

1. Please wait for the currently running Build check to finish green, because this PR changes compilation-sensitive Debezium API boundaries across MySQL/Postgres/Oracle/SQLServer.
2. The new DebeziumAdapterFactoryTest only exercises the no-provider path; adding at least one ServiceLoader success-path test with a test service file would make the factory contract much safer.

Conclusion: can merge after fixes

Blocking item:

1. Merge only after Build is green.

Suggested non-blocking improvement:

• Add a positive ServiceLoader adapter lookup test to cover the success path, not only exception/cache behavior.

[davidzollo]

A great work!
Thanks for your effort.

[DanielLeens]

Hi @vsantonastaso, thanks for pushing this Debezium-version isolation work. The direction is valuable: moving Debezium API differences out of cdc-base and into connector-specific adapters is exactly the kind of separation that should make future upgrades less painful. I reviewed the latest head locally on seatunnel-review-10799 (0fc3c8558, base 60f1007ab) and focused on the real CDC startup path.

What This PR Fixes

• User pain point: cdc-base currently has to know too much about Debezium version-specific APIs used by MySQL/Postgres/Oracle/SQLServer connectors.
• Fix approach: introduce a DebeziumAdapter SPI, load connector-specific adapters through ServiceLoader, and let each connector own its dispatcher/schema-history/topic-naming adaptation.
• One-sentence value: this PR aims to make Debezium version management per-connector instead of globally coupled in the base module.

Core Flow Reviewed

CDC source startup
  -> MySQL/Postgres/Oracle/SQLServer FetchTaskContext.configure()
  -> DebeziumAdapterFactory.getAdapter(connectorType) [DebeziumAdapterFactory.java:34-57]
  -> FetchTaskContext builds queue / heartbeat / topicSelector
  -> DebeziumEventDispatcherConfig.builder()
       .topicNaming((DebeziumTopicNaming<?>) (Object) topicSelector)
       -- topicSelector is Debezium TopicSelector<TableId>
       -- it does not implement SeaTunnel DebeziumTopicNaming
       -- this cast can fail during normal dispatcher initialization
  -> adapter.createEventDispatcher(dispatcherConfig)
  -> EventDispatcherAdapter casts config.getTopicNaming() back to TopicSelector [MySqlEventDispatcherAdapter.java:50-52]

Findings

Issue 1: TopicSelector is cast to DebeziumTopicNaming, so CDC dispatcher initialization can fail at runtime

• Location: seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java:181
• Severity: High

The same pattern appears in MySQL, Postgres, Oracle, and SQLServer:

.topicNaming((DebeziumTopicNaming<?>) (Object) topicSelector)

topicSelector is Debezium's TopicSelector<TableId>, not a SeaTunnel DebeziumTopicNaming. The intermediate (Object) only bypasses compile-time generic checks; the runtime cast still requires the object to implement the target interface. Normal CDC startup can therefore fail before the adapter dispatcher is even created.

The mismatch is also visible on the consumer side: MySqlEventDispatcherAdapter immediately casts config.getTopicNaming() back to TopicSelector<TableId> (MySqlEventDispatcherAdapter.java:50-52). That means the config field currently has two incompatible meanings.

Suggested fix: make this field type consistent. Either pass the native Debezium TopicSelector as an Object/connector-specific object into the adapter, or construct a real DebeziumTopicNaming wrapper first and have the adapter consume that wrapper. Avoid converting TopicSelector -> DebeziumTopicNaming -> TopicSelector by casts.

Issue 2: Postgres still constructs both the adapter dispatcher and a native PostgresEventDispatcher

• Location: seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java:281
• Severity: Medium

After creating PostgresEventDispatcherAdapter and assigning this.dispatcher, the code still directly creates a native PostgresEventDispatcher for this.pgEventDispatcher (PostgresSourceFetchTaskContext.java:286-301). If the goal is per-connector Debezium version isolation, this keeps part of the Postgres path tied to the current Debezium API directly in the fetch context.

Suggested fix: either create/wrap this through the Postgres adapter too, or document why it intentionally stays native and ensure tests cover the version-specific path.

Compatibility

• User config/defaults: unchanged.
• Internal CDC startup behavior: high impact; if the cast fails, existing CDC jobs cannot start.
• External protocol/data format: unchanged.

Performance And Side Effects

• ServiceLoader + cache is fine from a performance perspective.
• The main risk is not performance; it is startup correctness and adapter type safety.
• The current failure would be deterministic for affected CDC connectors, so retries would not help.

Tests And Docs

The SPI/unit tests are a good start, but they do not cover real FetchTaskContext dispatcher initialization. Please add tests that instantiate the MySQL/Postgres/Oracle/SQLServer dispatcher path far enough to catch the TopicSelector/DebeziumTopicNaming mismatch.

Merge Conclusion

Conclusion: merge after fixes

1. Blocking items:

• Issue 1: fix the TopicSelector/DebeziumTopicNaming type model so CDC dispatcher initialization cannot fail on the normal path.
• The PR is currently draft and the Build job failed (72356979292).

2. Non-blocking suggestions:

• Issue 2: clean up or clearly justify the Postgres dual dispatcher path so the adapter boundary remains maintainable.

Overall, the architectural goal is good and worth continuing. I would not merge the current version yet; the next revision should make the adapter config type-safe and add startup-path tests for the connector adapters.

[DanielLeens]

What This PR Fixes

• User pain point: the CDC base layer wants to reduce direct coupling to connector-specific Debezium implementations so each connector can manage its own Debezium dependency more cleanly.
• Fix approach: this PR adds a DebeziumAdapter SPI plus a ServiceLoader-based factory and connector-specific adapters for dispatcher, schema history, and related Debezium pieces.
• One-line summary: the direction makes sense, but the current head still has a high-risk type-boundary bug on the real CDC configure path.

Local Review Basis

I reviewed the full diff locally against upstream/dev using the latest fetched PR head 818569e7d22d and the existing local review branch context for PR #10799. I did not run local Maven in this batch.

Current Runtime Chain

CDC fetch task configure(split)
  -> DebeziumAdapterFactory.getAdapter("mysql" / "oracle" / "postgres" / "sqlserver")
  -> build native Debezium TopicSelector
  -> DebeziumEventDispatcherConfig.builder()
       .topicNaming((DebeziumTopicNaming<?>) (Object) topicSelector)
  -> adapter.createEventDispatcher(config)
  -> adapter ctor casts config.getTopicNaming() back to TopicSelector

Findings

1. The normal configure path for MySQL, Oracle, Postgres, and SQLServer all now do the same unsafe cast from a native Debezium TopicSelector into the custom DebeziumTopicNaming interface:
   • .../mysql/.../MySqlSourceFetchTaskContext.java:181-192
   • .../oracle/.../OracleSourceFetchTaskContext.java:155-166
   • .../postgres/.../PostgresSourceFetchTaskContext.java:268-279
   • .../sqlserver/.../SqlServerSourceFetchTaskContext.java:158-169
2. Those call sites do not create a wrapper that actually implements DebeziumTopicNaming; they still pass the native Debezium selector object.
3. MySqlDebeziumAdapter.createTopicNaming() even throws UnsupportedOperationException, which is another signal that topic naming has not really crossed the abstraction boundary yet.
4. The current Build check is red (72734124876), with failures across multiple updated-modules / connector integration suites.

Conclusion: can merge after fixes

Blocking items:

1. Please make the topic-naming abstraction real. Either pass an actual DebeziumTopicNaming adapter/wrapper through the config, or remove topic naming from this SPI for now and keep the native TopicSelector on the connector side.
2. Please rerun and recover the latest Build result after the runtime type-boundary issue is fixed.

Suggested non-blocking improvement:

• The new DebeziumAdapterFactoryTest is better than before, but I would still add at least one minimal configure()-level regression test so the real fetch-task path catches this kind of boundary mistake earlier.

Thank you for pushing this refactor forward. The broad direction looks right; the current blocker is that one piece of the new abstraction is not actually safe on the real CDC path yet.

[DanielLeens]

Hi @vsantonastaso, I rechecked the latest head locally after the new commit.

What changed after Daniel's previous review

• The earlier blocking bug was the fake DebeziumTopicNaming abstraction: the configure path still passed native Debezium TopicSelector objects through an unsafe cast.
• The latest head fixes that boundary:
  • DebeziumAdapterFactory.java:34-57 now loads adapters with the caller classloader.
  • MySqlDebeziumAdapter.java:45-54, PostgresDebeziumAdapter.java:45-54, and the other connector adapters now create real *TopicNamingAdapter wrappers.
  • The fetch-task configure path builds DebeziumEventDispatcherConfig with those wrappers instead of relying on the old unsafe cast.

Runtime chain I rechecked

SourceFetchTaskContext.configure(split)
  -> DebeziumAdapterFactory.getAdapter(type, classLoader)
  -> adapter.createTopicNaming(logicalName, heartbeatPrefix)
      -> real *TopicNamingAdapter wrapper
  -> DebeziumEventDispatcherConfig.builder().topicNaming(wrapper)
  -> adapter.createEventDispatcher(config)

Findings

1. The main runtime blocker from my previous review is fixed on the current head. Topic naming is now a real adapter boundary instead of an unsafe cast.
2. I do not see a reopened code blocker in the latest source diff.
3. The remaining gate is the current GitHub Build, which still is not green.

Merge conclusion

Conclusion: merge after fixes

Blocking item:

• Please rerun and get the latest Build green.

Non-blocking note:

• A small configure-level regression test would still be a nice follow-up, but I would not block the PR on that once CI is green.

Thanks for pushing this refactor through. The current head is in much better shape than the version Daniel previously blocked.

[vsantonastaso]

@davidzollo the build is green, all action jobs passed on my forked repo. Addressed also the feedback mentioned in the previous comments

[DanielLeens]

Hi @vsantonastaso, thanks for pushing this refactor forward. I agree with the problem statement: SeaTunnel CDC needs a path toward per-connector Debezium version management. However, after rechecking the current head locally, I do not think the current implementation is safe to merge yet.

The primary blocking concern is architectural, not stylistic.

1. The PR introduces adapter interfaces and ServiceLoader, but connector-cdc-base is still materially coupled to Debezium types and behavior. The base module still owns core runtime classes such as JdbcSourceEventDispatcher, and those classes still depend directly on Debezium APIs. That means the PR has not yet demonstrated the stronger property it is aiming for: that different connectors can evolve against different Debezium lines without the base layer becoming the compatibility boundary again.
2. In other words, the current code moves some construction logic into connector modules, but it does not fully remove Debezium-version-sensitive responsibilities from the shared base module. From a review perspective, this leaves the main architectural risk unresolved.
3. I would therefore treat the multi-version Debezium isolation is not yet actually proven problem as the first blocking item. Before merge, I would expect either a stronger design boundary that keeps connector-cdc-base version-neutral in the critical paths, or concrete proof that the remaining base-level Debezium dependencies do not break the intended upgrade model.

There are also two code-level issues on top of that architectural blocker:

1. The new DebeziumSchemaHistory path is not wired into the real runtime flow yet. In all four JDBC CDC fetch contexts, configure() still calls super.registerDatabaseHistory(...), so the normal path still goes through JdbcSourceFetchTaskContext.registerDatabaseHistory(...) and directly into EmbeddedDatabaseHistory.registerHistory(...). The new connector-specific registerDatabaseHistory(...) overrides are therefore dead code on the current path.
2. Even if that path is later wired in, the new schema-history bridge currently reserializes every TableChange as create(...), which would lose ALTER and DROP semantics during recovery.

Because of the above, my recommendation is: request changes.

What I would like to see before merge:

• A clearer and enforceable architectural boundary proving that connector-cdc-base is no longer the Debezium-version lock point for the targeted runtime paths.
• The schema-history path actually switched to the new abstraction on the real configure() -> registerDatabaseHistory() -> recover() flow.
• A fix for preserving CREATE / ALTER / DROP semantics in the schema-history bridge.
• At least one configure-level regression test that exercises the real fetch-context startup path, not only the factory/config helper classes.

I like the direction, but in its current form this still looks like an intermediate refactor rather than a completed isolation boundary.

[davidzollo]

Hi @vsantonastaso, thanks for pushing this refactor forward. I agree with the problem statement: SeaTunnel CDC needs a path toward per-connector Debezium version management. However, after rechecking the current head locally, I do not think the current implementation is safe to merge yet.

The primary blocking concern is architectural, not stylistic.

1. The PR introduces adapter interfaces and ServiceLoader, but connector-cdc-base is still materially coupled to Debezium types and behavior. The base module still owns core runtime classes such as JdbcSourceEventDispatcher, and those classes still depend directly on Debezium APIs. That means the PR has not yet demonstrated the stronger property it is aiming for: that different connectors can evolve against different Debezium lines without the base layer becoming the compatibility boundary again.
2. In other words, the current code moves some construction logic into connector modules, but it does not fully remove Debezium-version-sensitive responsibilities from the shared base module. From a review perspective, this leaves the main architectural risk unresolved.
3. I would therefore treat the multi-version Debezium isolation is not yet actually proven problem as the first blocking item. Before merge, I would expect either a stronger design boundary that keeps connector-cdc-base version-neutral in the critical paths, or concrete proof that the remaining base-level Debezium dependencies do not break the intended upgrade model.

There are also two code-level issues on top of that architectural blocker:

1. The new DebeziumSchemaHistory path is not wired into the real runtime flow yet. In all four JDBC CDC fetch contexts, configure() still calls super.registerDatabaseHistory(...), so the normal path still goes through JdbcSourceFetchTaskContext.registerDatabaseHistory(...) and directly into EmbeddedDatabaseHistory.registerHistory(...). The new connector-specific registerDatabaseHistory(...) overrides are therefore dead code on the current path.
2. Even if that path is later wired in, the new schema-history bridge currently reserializes every TableChange as create(...), which would lose ALTER and DROP semantics during recovery.

Because of the above, my recommendation is: request changes.

What I would like to see before merge:

• A clearer and enforceable architectural boundary proving that connector-cdc-base is no longer the Debezium-version lock point for the targeted runtime paths.
• The schema-history path actually switched to the new abstraction on the real configure() -> registerDatabaseHistory() -> recover() flow.
• A fix for preserving CREATE / ALTER / DROP semantics in the schema-history bridge.
• At least one configure-level regression test that exercises the real fetch-context startup path, not only the factory/config helper classes.

I like the direction, but in its current form this still looks like an intermediate refactor rather than a completed isolation boundary.

@vsantonastaso Thank you very much for making such an important and meaningful major optimization. Could you take a look at this comment from @DanielLeens? I think this comment makes a lot of sense. What do you think?

[DanielLeens]

Hi @vsantonastaso, thanks for the continued work here. I pulled the latest head locally again and rechecked the real CDC runtime path, especially the schema-history boundary.

What this PR is trying to solve

• User pain: Debezium-version differences should ideally be isolated per connector instead of staying coupled inside one shared runtime layer.
• Fix approach: this PR introduces connector-specific adapters / wrappers for parts of the Debezium integration.
• One-line summary: the direction is valuable, but the current implementation still does not own the real schema-history runtime boundary, so I still cannot recommend merge.

Runtime path I checked

MySqlSourceFetchTaskContext.configure(...)
  -> super.registerDatabaseHistory(sourceSplitBase, connection)
      -> JdbcSourceFetchTaskContext.registerDatabaseHistory(...)
          -> EmbeddedDatabaseHistory.registerHistory(...)
  -> validateAndLoadDatabaseHistory(...)
  -> adapter.createEventDispatcher(...)

Blocking issues

1. The schema-history critical path is still controlled by the base layer.

• Location: .../MySqlSourceFetchTaskContext.java:128-147, .../JdbcSourceFetchTaskContext.java:167-208
• The new adapter layer exists, but the real runtime path still starts with super.registerDatabaseHistory(...), and the base implementation still writes directly into EmbeddedDatabaseHistory.

2. The incremental history bridge still collapses DDL semantics too aggressively.

• Location: .../JdbcSourceFetchTaskContext.java:177-199
• The current bridge expects a single TableChange per table, which is not enough to convincingly represent a full CREATE/ALTER/DROP history chain.

Conclusion: do not merge yet

1. Blocking items

• Please move the real schema-history runtime boundary under the adapter-owned path, or explicitly narrow the PR goal.
• Please add proof (code + regression coverage) that the history bridge preserves the required semantics.

2. Suggested follow-ups

• The current Build is green, but that does not close the architectural blockers above.

I appreciate the direction here, and I think the topic-naming / dispatcher cleanup is useful. The blocker is that the most version-sensitive runtime boundary is still not truly isolated yet.

[DanielLeens]

I re-reviewed the latest head after the new commit and traced the MySQL CDC schema-history path locally.

What this PR solves

• User pain: shared Debezium implementation becomes harder to maintain as connector-specific version gaps grow.
• Fix approach: introduce a DebeziumAdapter SPI and move MySQL-specific event-dispatcher, topic-naming, and schema-history initialization behind that adapter.
• One-line value: the refactoring direction is good, but the schema-history boundary is still not fully isolated per connector.

Runtime path

MySqlSourceFetchTaskContext.configure() [122-142]
  -> DebeziumAdapterFactory.getAdapter("mysql", classLoader) [126]
  -> registerDatabaseHistory(...) [129]
     -> super.buildTableSchemaHistory(...) [224]
     -> adapter.createSchemaHistory(instanceName, tableChangeInfos) [231-232]
     -> schemaHistory.start() [233]
  -> validateAndLoadDatabaseHistory(...) [142]

MySqlSchemaHistoryAdapter
  -> new ConnectTableChangeSerializer() [48]
  -> EmbeddedDatabaseHistory.registerHistory(...) [60]
  -> stop() -> EmbeddedDatabaseHistory.removeHistory(...) [84-85]

Close path
  -> MySqlSourceFetchTaskContext.close() only closes connection / binaryLogClient [214-217]
  -> no matching schemaHistory.stop()

Review findings

Issue 1: The schema-history path still depends on shared serializer and storage classes, so the version-isolation boundary is not fully moved into the connector layer

• Location: MySqlSchemaHistoryAdapter.java:20-24, :48-50, :60-66
• Why this matters: the adapter entry point is connector-specific now, but the most compatibility-sensitive path still uses shared ConnectTableChangeSerializer and EmbeddedDatabaseHistory.
• Risk: once connector versions diverge further, the history protocol can still remain coupled in the shared layer.
• Better fix: move schema-history serialization and storage fully behind the connector-specific adapter.

Issue 2: schemaHistory.start() is not paired with lifecycle cleanup in the fetch task

• Location: MySqlSourceFetchTaskContext.java:221-233 and :214-217
• Why this matters: the fetch task starts schema history but does not keep and stop it during close.
• Risk: stale history state across long-lived processes, retries, or reader recreation.
• Better fix: store the schema-history instance and call stop() from close().

Merge conclusion

Conclusion: Merge after fixes

Blocking items:

• Issue 1 should be fixed before merge.
• CI also needs attention: Notify test workflow is currently failing with There was a new unsynced commit pushed. Please retrigger the workflow.

Non-blocking suggestions:

• I strongly recommend fixing Issue 2 in the same round so the lifecycle stays symmetric.

CI status:

• The PR does not currently have a trustworthy green build state because the workflow notification step is already out of sync with the latest head.