Skip to content

Conversation

@CloverDew
Copy link
Contributor

Purpose of this pull request: Fixes #9980

Problem

After implementing CDC schema evolution support in Flink engine, several issues were identified:

  1. The coordinator's state resides in the TaskManager's JVM; if a job fails and restarts, this state is easily lost.
  2. Flink's proprietary implementation logic should not be placed in the public API SupportSchemaEvolutionSinkWriter. Furthermore, flush requires manual API implementation.
  3. Multi-parallelism configuration is not supported.

Solution

This PR made some minor adjustments to the architecture.:

  • BroadcastSchemaSinkOperator operator is introduced to solidify the response state.
  • Coordinator is only responsible for communication: pure messenger, stateless.
  • Flushing is guaranteed by checkpoints, not manual implementation

Key Changes:

1. Enhanced LocalSchemaCoordinator

  • JobId isolation: Prevents multi-job interference using Map<String, WeakReference<LocalSchemaCoordinator>>
  • Pure messenger: No persistent state, only temporary communication futures

3. Streamlined BroadcastSchemaSinkOperator

  • Idempotency: tracks lastProcessedEpoch using Flink state

4. API Compliance (Addresses #9980)

  • API separation: SupportSchemaEvolutionSinkWriter contains only generic applySchemaChange() method
  • Flink-specific logic moved: All coordination logic moved from API to Flink translation layer

Testing

  • Verified MySQL CDC → MySQL sink schema evolution scenarios
  • Confirmed no data loss during schema changes
  • Tested multi-job concurrent execution

Breaking Changes

None. This is a refactoring that maintains API compatibility while improving internal implementation.

Related Issues

Check list

@CloverDew CloverDew force-pushed the feature/supports-multiple-parallelisms-on-flink branch 3 times, most recently from 1e42e4d to 460e584 Compare November 24, 2025 14:00
@Carl-Zhou-CN Carl-Zhou-CN changed the title [Fix][Flink]supports multiple parallelisms and remove flink-specific logic from API [Impove][Flink]supports multiple parallelisms and remove flink-specific logic from API Nov 25, 2025
@CloverDew CloverDew changed the title [Impove][Flink]supports multiple parallelisms and remove flink-specific logic from API [Improve][Flink]supports multiple parallelisms and remove flink-specific logic from API Nov 25, 2025
@CloverDew CloverDew force-pushed the feature/supports-multiple-parallelisms-on-flink branch from 460e584 to ae2c6b3 Compare November 27, 2025 09:38

boolean isStreaming =
envConfig.hasPath("job.mode")
&& "STREAMING".equalsIgnoreCase(envConfig.getString("job.mode"));
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thx, please use org.apache.seatunnel.common.constants.JobMode#STREAMING instead of STREAMING

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Bug][Flink] Move Flink-specific flush coordination from API to translation layer

2 participants