davidzollo opened a new issue, #11381:
URL: https://github.com/apache/seatunnel/issues/11381
### Motivation
PR #11156 fixes an unsafe ordering in the Doris sink by flushing the
in-flight Stream Load before applying a schema change in non-2PC mode, and by
rejecting unsafe `sink.enable-2pc=true` with positional formats such as CSV.
That connector-level fix is still useful, but SeaTunnel should also provide
a stronger engine-level guarantee: before a `SchemaChangeEvent` reaches sink
writers and downstream DDL is applied, all records written with the previous
schema should have crossed a checkpoint / flush boundary.
Without this guarantee, a source or transform path that emits a
`SchemaChangeEvent` without first triggering the existing schema-change
checkpoint protocol can still let buffered old-schema rows be committed after
the target table has already been altered.
### Current behavior
SeaTunnel already has part of the required mechanism:
- CDC deserialization can mark schema-change checkpoint boundaries through
`Collector.markSchemaChangeBeforeCheckpoint()` and
`Collector.markSchemaChangeAfterCheckpoint()`.
- `SeaTunnelSourceCollector` captures these signals.
- `SourceFlowLifeCycle` triggers `SCHEMA_CHANGE_BEFORE_POINT_TYPE` /
`SCHEMA_CHANGE_AFTER_POINT_TYPE` checkpoints and pauses source collection while
schema change is in progress.
- Sink checkpoint handling calls `SinkWriter.prepareCommit(checkpointId)`
and `SinkWriter.snapshotState(checkpointId)`, which is the right place for
connectors to flush / pre-commit buffered data.
However, the sink-side schema event path still applies DDL directly when the
event arrives:
- `SinkFlowLifeCycle.received(...)` handles `SchemaChangeEvent` by directly
calling `SupportSchemaEvolutionSinkWriter.applySchemaChange(event)` or the
deprecated `SinkWriter.applySchemaChange(event)`.
This means the global protection depends on every source producing the
correct schema-change-before checkpoint signal before the schema event. It is
not currently enforced by the engine for all `SchemaChangeEvent` paths.
### Proposal
Add an engine-level schema-change drain protocol so that a schema change
cannot be applied to sinks until previous-schema records have been drained
through a checkpoint / flush boundary.
A possible implementation direction:
1. Treat every non-create `SchemaChangeEvent` as requiring a
schema-change-before drain unless the event is already known to be inside a
completed schema-change checkpoint phase.
2. Reuse the existing schema-change checkpoint types instead of calling
`SinkWriter.prepareCommit()` directly outside checkpoint context, because
`prepareCommit()` may create 2PC committables that must be completed or aborted
by the checkpoint lifecycle.
3. Gate forwarding of `SchemaChangeEvent` to
`SinkFlowLifeCycle.applySchemaChange(...)` until all relevant sink subtasks
have acknowledged the schema-change-before checkpoint.
4. Keep connector-specific compatibility checks in connectors. For example,
Doris still needs to reject unsafe `sink.enable-2pc=true` with positional
formats such as CSV because only the connector knows whether its load format is
name-based or position-based.
5. Add unit / integration coverage for the ordering:
```text
old-schema records
-> schema-change-before checkpoint
-> sink prepareCommit / snapshotState
-> checkpoint complete
-> SchemaChangeEvent reaches sink writer
-> sink applies DDL
-> new-schema records
```
### Expected benefits
- Prevents downstream DDL from racing ahead of old-schema buffered rows.
- Gives schema evolution a single engine-level ordering contract instead of
relying on each connector to rediscover the same drain requirement.
- Keeps connector-specific transaction and format validation local to each
connector.
- Reduces the chance of data corruption for positional formats and schema
changes such as drop / rename / reorder-sensitive changes.
### Related context
- PR #11156: connector-level Doris fix for flushing before schema change and
rejecting unsafe 2PC schema evolution formats.
- Existing schema-change checkpoint flow: `SeaTunnelSourceCollector`,
`SourceFlowLifeCycle`, `CheckpointType.SCHEMA_CHANGE_BEFORE_POINT_TYPE`, and
`SinkFlowLifeCycle`.
--
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]
