[Feature][Connector-File] Add schema evolution support (ADD/DROP/RENAME/UPDATE column) for all file formats

[ricky2129]

Purpose

Implements SupportSchemaEvolutionSink / SupportSchemaEvolutionSinkWriter for the file sink connector, enabling CDC pipelines (MySQL/PostgreSQL → File) to handle
live schema changes without job restarts.

What changes were proposed in this pull request?

New config option

• schema_evolution_enabled (boolean, default false) — opt-in flag, same pattern as the Iceberg sink. When false, supports() returns an empty list so the
  engine does not route schema events to this sink.

Core logic (AbstractWriteStrategy)

• applySchemaChange(SchemaChangeEvent) — handles all 4 change types: ADD_COLUMN, DROP_COLUMN, RENAME_COLUMN, UPDATE_COLUMN, plus batch
  AlterTableColumnsEvent
• File rotation: calls finishAndCloseFile() on schema change so files have clean schema boundaries. Old file closes with old schema; new file opens with new
  schema on the next write.
• sinkColumnNames list tracks which columns to write; updated per event with case-insensitive matching
• rebuildSinkColumnsIndex() recomputes sinkColumnsIndexInRow after every schema change
• onSchemaChanged() hook for subclass-specific cache invalidation
• getFieldSafe(row, index) — null-safe field accessor for in-flight rows arriving after ADD_COLUMN
• safeProjectedRow(row) — null-safe projected row copy for text-format strategies
• try-finally around finishAndCloseFile() so beingWrittenFile is always cleared even if a writer close throws

Format-specific fixes

Format	Change
Parquet	onSchemaChanged() nulls cached Avro schema and rebuilds writePathsAsInt96; resolveObject() uses name.toLowerCase() for Avro field lookup (fixes
NPE on mixed-case column names)
ORC	getFieldSafe() used in write()
Text / CSV / JSON / CanalJson / DebeziumJson / MaxwellJson	safeProjectedRow() replaces row.copy(int[])
Excel / XML	Inline bounds guard on getField(index)
Binary	applySchemaChange() throws FileConnectorException — Binary format has a fixed schema and cannot evolve

Interface wiring

• BaseMultipleTableFileSink implements SupportSchemaEvolutionSink; supports() returns all 4 types when enabled, empty list when disabled
• BaseFileSinkWriter implements SupportSchemaEvolutionSinkWriter; delegates to writeStrategy.applySchemaChange()

Test coverage

FileSchemaEvolutionTest (14 tests):

• ADD / DROP / RENAME / UPDATE column
• ADD FIRST and ADD AFTER position handling
• Batch AlterTableColumnsEvent
• Case-insensitive DROP and RENAME
• getFieldSafe out-of-bounds guard
• Disabled flag no-op (schema not mutated when schema_evolution_enabled=false)
• finishAndCloseFile call count on schema change
• Index consistency after add+drop sequence
• try-finally guarantee (writer-close failure still clears beingWrittenFile)

Known limitation

Schema changes are not atomic with respect to checkpointing. If the engine takes a checkpoint in the narrow window between finishAndCloseFile() completing and
seaTunnelRowType being updated, restoring from that checkpoint would write subsequent rows using the pre-change schema, silently omitting added columns.

This is a known architectural limitation shared by all file-sink schema evolution implementations (including the Iceberg connector). The window is very small (~1ms)
and only materialises on a crash-and-restore event — not during normal pipeline operation. A running CDC pipeline with no crashes will never encounter this: the
moment the ALTER fires, all subsequent rows are immediately written with the new schema.

Resolving this fully requires engine-level protocol support and is out of scope for this PR.

Operational mitigation: pause the pipeline before planned DDL migrations, or verify output immediately after any schema change.

Does this PR introduce any UI changes?

No.

Does this PR introduce any breaking changes?

No. The feature is opt-in via schema_evolution_enabled=false (default). Existing jobs are unaffected.

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

[ricky2129]

@davidzollo @dybyte can you help validating this feature.

[ricky2129]

also to add =>
The schema-change / checkpoint atomicity limitation noted above is a known architectural gap in SeaTunnel's current SinkWriter API. It is not specific to this connector — the Iceberg and JDBC sinks have the same behaviour.

A proper fix would require the engine to pause checkpoint barriers while a schema change event is being applied, similar to how Flink handles watermarks and barriers. This would need a new coordination primitive in SinkWriter (e.g. beginSchemaChange / endSchemaChange lifecycle hooks) so the engine knows not to snapshot between the two steps.

We are happy to contribute that engine-level fix as a follow-up issue if the community agrees on the approach. For now, the connector-level implementation is as safe as the API allows, and the limitation only materialises on a crash-and-restore event, not during normal operation

[DanielLeens]

Thanks for the update. I pulled this locally and there is still one blocking correctness gap in the current implementation.

BaseMultipleTableFileSink#supports() now advertises ADD/DROP/RENAME/UPDATE whenever schema_evolution_enabled=true, but BinaryWriteStrategy#applySchemaChange() still throws FORMAT_NOT_SUPPORT. In the streaming Flink path, SinkExecuteProcessor inserts the schema broadcast operator based on sink instanceof SupportSchemaEvolutionSink, so a file sink configured with file_format_type=binary and schema_evolution_enabled=true will enter the main schema-evolution path and fail on the first schema change instead of being rejected up front.

Please either stop advertising schema evolution for formats that still cannot apply it, or reject schema_evolution_enabled=true during config validation for binary.

[DanielLeens]

I re-pulled the latest HEAD locally and checked the schema-change write path again.

The previous binary-format blocker is addressed in the current revision by the upfront validation in FileSinkConfig.

I still see one blocking correctness gap in the partitioned sink path.

Current chain:
SchemaChangeEvent
-> BaseFileSinkWriter.applySchemaChange(...)
-> AbstractWriteStrategy.applySchemaChange(...) rebuilds seaTunnelRowType and sinkColumnsIndexInRow
-> next row write enters generatorPartitionDir(...)
-> partition path still reads partitionFieldsIndexInRow / partitionFieldList from FileSinkConfig

The problem is that those partition metadata fields are still initialized once in FileSinkConfig and are not refreshed during applySchemaChange(...). So after an ADD / RENAME / DROP affecting partition columns, the current revision can still route rows to the wrong partition directory, and dropping a partition column can even turn into an out-of-bounds access in the partition-path generation.

So I still think this needs changes before merge.

Suggested fix order:

1. rebuild mutable partition metadata together with the sink column indexes during applySchemaChange(...), or
2. narrow the advertised support so partitioned file sinks are not included in schema evolution support yet.

[DanielLeens]

Hi @ricky2129, I re-pulled the latest head locally and rechecked the schema-change write path again.

Runtime chain I verified:

schema change event
  -> AbstractWriteStrategy.applySchemaChange()
      -> rebuild sink-column mapping
  -> writeRecordToFile()
      -> partition path still reads fileSinkConfig.partitionFieldList / partitionFieldsIndexInRow

The two blockers from my earlier review are still open in the current head:

1. The partitioned file-sink path still reads partitionFieldList / partitionFieldsIndexInRow from FileSinkConfig, while applySchemaChange() only rebuilds sink-column mapping. That means schema evolution for partitioned sinks can still route rows with stale partition metadata.
2. This is still a user-visible file sink capability change, but the current diff still does not update the matching docs/en and docs/zh documentation.

Conclusion

Conclusion: fix required before merge

1. Blocking items

• Please either rebuild partition metadata together with schema metadata, or explicitly block schema_evolution_enabled + partition_by until that path is safe.
• Please add the matching English and Chinese docs for the supported scope and the current limits.

2. Suggested improvements

• Once the partition path is closed and the docs land, this PR will be much closer.

The current Build check is also FAILURE, but the two correctness / documentation gaps above are the real blockers from my side.

[dybyte]

Could you also update the docs for this change? It would be great to document the new option and its current limitations.

[DanielLeens]

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

I rechecked the file-sink schema evolution chain and the latest maintainer comment:

CDC schema change event
  -> BaseFileSinkWriter.applySchemaChange(event)
  -> WriteStrategy.applySchemaChange(event)
      -> finish/close current file
      -> update sink schema / column names / index mapping
      -> next rows are written with the evolved schema
  -> format-specific writers rebuild their schema-dependent state

The implementation is much more complete than a local patch, but this is user-visible behavior and it introduces a new option/limitation surface. I agree with the latest maintainer request: docs need to explain the new option and the current schema-change/checkpoint limitation before merge. Fetched CI metadata also reports Build: FAILURE.

Conclusion: can merge after fixes

Blocking items:

1. Add docs for the new file-sink schema evolution option and its current limitations.
2. Fix/rerun the failing Build check.

[davidzollo]

The DataTypeChangeEventDispatcher class is annotated with @deprecated. Should it be replaced with TableSchemaChangeEventDispatcher?

[ricky2129]

@davidzollo will be pushing some more changes, right now testing this out, you can review post that.

[DanielLeens]

Hi @ricky2129, thanks for the careful iterations here. I re-reviewed the current head 01a01e2ffaed3838e6e7cdcc577f06d3b1e0200e and intentionally ignored the current merge conflicts, as requested, so the comments below are only about the PR logic and merge readiness.

Runtime path I checked:

CDC source emits a SchemaChangeEvent
  -> MultiTableSinkWriter.applySchemaChange(event)
      -> find the matching table writer
      -> synchronize on the target writer runnable
      -> SupportSchemaEvolutionSinkWriter.applySchemaChange(event)

File sink writer
  -> BaseFileSinkWriter.applySchemaChange(event)
      -> WriteStrategy.applySchemaChange(event)

File write strategy
  -> AbstractWriteStrategy.applySchemaChange(event)
      -> no-op when schema_evolution_enabled=false
      -> finishAndCloseFile() to rotate/close current files
      -> DataTypeChangeEventDispatcher updates SeaTunnelRowType
      -> update sink column names
      -> rebuild sink column indexes
      -> onSchemaChanged() lets each format rebuild cached serializers/schemas

Next records
  -> Text/Csv/Json/CDC JSON use safeProjectedRow()
  -> Orc/Parquet use getFieldSafe()
  -> old in-flight rows after ADD_COLUMN can write null for the new field instead of failing with an index error

The write-path design is much safer than the earlier revisions:

• schema_evolution_enabled is still opt-in and defaults to false, so existing file sink jobs keep their previous behavior.
• BaseMultipleTableFileSink.supports() is gated by the option, so normal jobs do not advertise schema evolution accidentally.
• The generic schema/mapping update lives in AbstractWriteStrategy, while format-specific cached schemas are rebuilt through onSchemaChanged().
• binary is rejected up front.
• partition_by + schema_evolution_enabled is rejected up front, which closes the stale partition metadata path discussed earlier.

I do not see a remaining correctness blocker in the schema-evolution file write path itself when conflicts are ignored. There are still two merge blockers at the current head:

Blocking item 1: the current head has a deterministic test compile issue

FileSchemaEvolutionTest.java:521 and FileSchemaEvolutionTest.java:535 still call:

new FileSinkConfig(config, BASE_ROW_TYPE)

but FileSinkConfig only exposes the constructor:

FileSinkConfig(ReadonlyConfig pluginConfig, SeaTunnelRowType seaTunnelRowTypeInfo)

The helper methods earlier in the same test already use ReadonlyConfig.fromConfig(config). These two assertions should do the same, otherwise the connector-file test module cannot compile. This also matches the current PR status where the Build check is failing.

Blocking item 2: docs still need to be updated

This is a user-visible new option and behavior. Please document schema_evolution_enabled in the relevant file sink docs under both docs/en and docs/zh, including:

• default value: false
• supported event types: ADD/DROP/RENAME/UPDATE column
• supported formats and the explicit binary limitation
• the partition_by limitation
• the current schema-change/checkpoint atomicity limitation and the recommended operational caution around planned DDL changes

Conclusion: fix required before merge

Blocking items:

1. Fix the two FileSchemaEvolutionTest constructor calls and get the Build check green.
2. Add the matching English and Chinese docs for the new option and current limitations.

Suggested non-blocking follow-up:

• A small file-level integration test for at least one row-based format and one columnar format would be useful later, but I would not block this PR on that once the compile issue and docs are fixed.

Overall, the implementation is close from my side. The earlier binary and partitioned-sink risks are handled now, and the remaining work is mainly getting the current head buildable and making the new user-facing behavior clear in docs.

[davidzollo]

In FileSinkConfig, the method generatorPartitionDir() directly calls fileSinkConfig.getPartitionFieldsIndexInRow() for every write operation. This is a static index calculated based on the initial seaTunnelRowType when FileSinkConfig is constructed. If the upstream performs an ADD FIRST operation (shifting all column indices forward by 1) or drops columns that come before partition columns, partitionFieldsIndexInRow will point to incorrect row fields, resulting in wrong partition directory values. If the partition columns themselves are dropped, seaTunnelRow.getFields()[staleIndex] will eventually throw an ArrayIndexOutOfBoundsException. Should this issue also be fixed?

[DanielLeens]

Hi @ricky2129, I rechecked the latest head locally, including the new concern raised in the discussion.

What I verified

• The new stale partition-index concern does not hit the runtime path on the current head, because FileSinkConfig.java:233-240 now rejects schema_evolution_enabled=true together with partition_by during config construction.
• The schema-change write path itself is still the one I reviewed before:

SchemaChangeEvent
  -> MultiTableSinkWriter.applySchemaChange(event)
      -> BaseFileSinkWriter.applySchemaChange(event)
          -> AbstractWriteStrategy.applySchemaChange(event)
              -> close current files
              -> rebuild rowType / sink columns / indices
              -> refresh format-specific caches

• I also ran a local mergeability check with git merge-tree, and the current branch still has a real conflict against latest dev.

Findings

1. The current branch is still CONFLICTING. The concrete conflict I reproduced is in seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java:124.
2. The user-facing docs are still missing for schema_evolution_enabled in both docs/en and docs/zh.
3. The latest discussion concern about partition-field indices becoming stale after schema evolution is not a blocker on the current head, because that config combination is explicitly rejected now.

Merge conclusion

Conclusion: merge after fixes

Blocking items:

• Please resolve the real merge conflict first, especially in ExcelGenerator.java, and then rerun CI.
• Please add the matching English and Chinese docs for schema_evolution_enabled, including default value, supported event types, and the current binary / partition_by limitations.

Non-blocking note:

• From the current source path, I do not see a reopened runtime correctness blocker in the schema-evolution write flow itself once conflicts are ignored.

The implementation is still close. The remaining blockers are mergeability and docs completeness, not a reopened core-path regression.

[ricky2129]

Hi @davidzollo @dybyte @DanielLeens — pushed an update that brings this PR up to a fully working
end-to-end schema evolution feature, validated against MySQL CDC → S3 Parquet on a stage cluster
across multiple ALTER shapes (ADD COLUMN end / FIRST / AFTER, DROP, RENAME, MODIFY,
multi-event batches, partition_by + evolution combo). 30 unit tests pass.

Resolves blocking items:

• Merge conflict against latest dev resolved — ExcelGenerator rebased to combine upstream's
  sequential cell-index fix with the bounds-safe-access fix from this PR.
• FileSchemaEvolutionTest constructor calls now use ReadonlyConfig.fromConfig(config) —
  build is green.

Addressing @davidzollo's question on partition_by + stale partition indices:

"If the upstream performs an ADD FIRST operation (shifting all column indices forward by 1)
or drops columns that come before partition columns, partitionFieldsIndexInRow will point to
incorrect row fields..."

Fixed. AbstractWriteStrategy now keeps a writer-local partitionFieldsIndexInRow (defensive
copy in constructor — the shared FileSinkConfig is unsafe to mutate). On every ALTER,
applySchemaChange rebuilds it via name-based lookup against the post-ALTER row type, mirroring
the existing sinkColumnsIndexInRow pattern. Drop/rename of a partition column itself is rejected
at rebuild time with an explicit IllegalStateException (fail-fast — never reaches the
getFields()[staleIndex] AIOOBE you described).

The partition_by + schema_evolution_enabled=true config is therefore now supported (the previous
guard in FileSinkConfig that rejected this combination has been removed). New test
testPartitionByWithSchemaEvolutionEnabledIsAccepted documents the new contract.

Adds the chain-order propagation fix (file-sink schema evolution requires this end-to-end):
The file sink's schema evolution doesn't actually work in production unless transform-chain
order divergence is handled. Each transform in a chain (Metadata, RowKindExtractor, SQL, Filter)
applies the ALTER event to its own catalog, but the actual data row carries upstream's column
positions — without a propagation hook, a 5-transform chain accumulates index drift and the row
that arrives at the sink has columns at positions that don't match its catalog.

SeaTunnelTransform.setInputCatalogTables() is added as a default-no-op interface method.
TransformFlowLifeCycle calls it after each transform's mapSchemaChangeEvent, passing the
upstream's actual produced catalog. AbstractMultiCatalogTransform overrides
mapSchemaChangeEvent to dispatch to the right inner per-table transform AND set
event.changeAfter so downstream readers (sink + other transforms) adopt the propagated layout
instead of re-applying ALTER on a stale local view. Five regression tests cover the chain edges.

Open design question for maintainers:
When schema_evolution_enabled=false at the sink BUT schema-changes.enabled=true at the CDC
source, the sink currently silently no-ops the ALTER event. The next data row then arrives with
the new layout, the sink reads row[idx] against stale sinkColumnsIndexInRow, and Parquet's
AvroWriteSupport throws a confusing ClassCastException several rows later. This is the
production failure that motivated rolling out the feature.

Would you prefer:
(a) Fail-fast — throw FileConnectorException at the head of applySchemaChange when
flag=false but a real AlterTableEvent arrives, with a message telling users to either
enable the sink-side flag or set schema-changes.enabled=false at source.
(b) Continue current behaviour — silent no-op (preserves backward compatibility, but
leaves the latent CCE trap).
(c) Graceful name-based projection — sink continues with old schema, name-projects each
row to drop new columns; surprising for users ("where did my new column go?") but no crash.

I've kept the current no-op behavior in this push so nothing changes for existing users. Happy
to flip it to fail-fast or implement projection if the community has a preference.

[DanielLeens]

Thanks for addressing the partition-index issue from the previous round. I re-reviewed the latest head locally.

What this PR solves

• User pain: file sinks could not reliably follow upstream CDC schema changes, especially when column positions shifted.
• Fix approach: rebuild sink-column indices and partition-field indices on schema change, close and reopen current writers, and add safe projection helpers for in-flight old-schema rows.
• One-line value: the runtime fix is now in much better shape; the remaining blockers are documentation and CI, not the core partition-index bug from the previous round.

Runtime path

Schema change reaches the sink
  -> AbstractWriteStrategy.applySchemaChange() [193-255]
     -> finishAndCloseFile() [214-218]
     -> update tableSchema from changeAfter or local handler [226-233]
     -> updateSinkColumnNames(event) [236]
     -> rebuildSinkColumnsIndex() [239]
     -> rebuildPartitionFieldsIndex() [244]
     -> onSchemaChanged() [247]

Write phase
  -> safeProjectedRow() and getFieldSafe() handle in-flight old-schema rows
  -> partition directory generation reads the rebuilt partitionFieldsIndexInRow

Review findings

Issue 1: The user-visible schema_evolution_enabled feature still has no matching English and Chinese documentation

• Location: AbstractWriteStrategy.java:193

• Why this matters: this is now a real user-facing capability, but the PR still does not explain how to enable it, which change types are supported, or what happens for in-flight old-schema rows.

• Risk: the feature ships but users cannot use it confidently.

• Better fix: update both docs/en and docs/zh with option name, supported change types, example usage, and behavioral notes.

• I re-checked the previous runtime blocker and it is fixed in the latest head.

Merge conclusion

Conclusion: Merge after fixes

Blocking items:

• Issue 1 should be fixed before merge.
• CI also needs attention: the current Build check is ACTION_REQUIRED with the explicit message Workflow run detection failed.

Non-blocking suggestions:

• On the code side, I did not find a new blocking runtime issue after the latest update.

CI status:

• GitHub explicitly points to workflow detection failure and suggests re-enabling Actions, rebasing on the latest upstream/dev, and retriggering the workflow.

[DanielLeens]

Hi @ricky2129, I re-pulled the latest head locally and re-reviewed it against upstream/dev.

What this PR fixes

• User pain: file sink schema evolution can corrupt writes after live ADD/DROP/RENAME/UPDATE COLUMN events if sink column mappings, partition mappings, or transform-chain schema propagation are not rebuilt correctly.
• Fix approach: the current head rebuilds both sink-column and partition-column indices in applySchemaChange(), and it extends schema-change propagation across the file sink / transform chain.
• One-line summary: the runtime correctness gap Daniel previously flagged on partition metadata is fixed on the current head.

Runtime chain I checked

schema event
  -> BaseFileSinkWriter.applySchemaChange(...)
      -> AbstractWriteStrategy.applySchemaChange()          [193-245]
          -> rebuildSinkColumnsIndex()
          -> rebuildPartitionFieldsIndex()
          -> onSchemaChanged()

next write
  -> generatorPartitionDir(row)                             [427-472]
      -> now reads writer-local partitionFieldsIndexInRow

support advertisement
  -> BaseMultipleTableFileSink.supports()                   [132-141]
      -> ADD / DROP / RENAME / UPDATE when schema_evolution_enabled=true

Key findings

• The stale partition-metadata blocker from the earlier review is fixed in the current runtime path.
• The remaining blocker is now documentation, not the main write-path code.
• The PR already advertises file sink schema evolution to the engine, but the docs still do not explain the new capability or the current partition-column limits.

Blocking issue

1. The current docs still do not describe this user-visible capability change.

• Code side: AbstractWriteStrategy.java:193-245,394-472 and BaseMultipleTableFileSink.java:132-141
• Docs side: docs/en/connectors/sink/LocalFile.md:45-80, docs/zh/connectors/sink/LocalFile.md:44-78, and the generic schema-evolution pages still do not list file sink support.
• Risk: users cannot tell that file sink schema evolution is now supported, and they also cannot see the current fail-fast limit around dropping/renaming partition columns.

CI note

• The current Build is action_required because GitHub did not detect a usable workflow run from the fork, so there is also no trustworthy green CI result yet.

Merge decision

Conclusion: can merge after fixes

1. Blocking items

• Please add the matching English and Chinese docs for the supported scope, the enabling flag, and the current partition-column limits.
• Please trigger a valid Build run after that so the current head has real CI coverage.

2. Suggested non-blocking follow-up

• No new code-level blocker from my side in the current runtime path.

[ricky2129]

@davidzollo @dybyte @DanielLeens — closing thought before merge.

This PR delivers the file-sink half of schema evolution. End-to-end on a real CDC pipeline (MySQL → S3 Parquet), I found three CDC-source-side fixes are also required for the feature to work correctly in restart
scenarios. Without them, an upstream user enabling schema_evolution_enabled=true will hit one of:

1. Null values for new columns— when job restores from pre-DDL checkpoint and resumes streaming from a binlog position past the DDL (DDL during outage scenario). Root cause: Debezium's MySqlSchema cache + the deserializer's stale converter list.

2. AIOOBE on first row after restart-with-DDL —SeaTunnelSourceCollector captures rowType at task-open time; when the deserializer is later restored to a post-ALTER (wider) rowType,getBytesSize(oldRowType) AIOOBEs. Live rowType supplier fixes this.

3.tinyint(1) columns added via ALTER come through as TINYINT, not BOOLEAN — length info is dropped during MySqlSchema rebuild. Affects MySQL CDC + int_type_narrowing users.

I have working implementations of all three on a private branch,validated end-to-end on a stage cluster. They're all CDC-source-side , so they fix schema delivery for all CDC sinks (Iceberg, JDBC, file), not justfile sink.

Question: would you prefer
(a) bundle these into this PR — feature is complete end-to-end on merge but PR scope grows beyond "file sink schema evolution"

(b) keep this PR file-sink-scoped and merge it; I'll open a separate "CDC source-side schema evolution hardening" PR within the week that depends on this PR

(c) any other split you'd prefer

I'd marginally lean (b) for cleaner review boundaries, but (a) is also fine if you'd rather not have a temporarily-broken feature in dev. Your call.

[DanielLeens]

Hi @ricky2129, thanks for laying out the trade-off so clearly.

From Daniels side, I would lean (b): keep this PR file-sink-scoped, and open the CDC source-side hardening as a separate follow-up PR. Those three restart-path fixes affect the upstream CDC delivery contract for all downstream sinks, so they deserve their own review boundary instead of being folded into a file-sink feature PR at the end.

The one important caveat is messaging and docs: until that follow-up lands, I would not describe this PR as fully end-to-end complete across restart-with-DDL scenarios. If the known restart limitation still exists today, please call it out explicitly in the docs / PR narrative so users do not read this as “schema evolution is fully closed end to end under recovery” yet.

So my recommendation is:

• keep this PR focused on the file-sink side
• keep the current limitations explicit
• then send the CDC source-side hardening as a dedicated follow-up PR

Also, the current GitHub Build is still ACTION_REQUIRED, so that gate still needs to be recovered before merge.

[ricky2129]

@davidzollo added the changes

[DanielLeens]

Thanks for the ping. I checked the PR metadata again before replying.

The current head is still 65fcf451a2d53538b6d84043a8fdd93408e407b3, so I do not see a new commit behind the latest "added the changes" message yet. On this unchanged head, Daniel's source-level conclusion still stays the same as the latest approval.

One correction on the CI thread: the latest fork Build run 26526641462 is not only red on seatunnel-engine-client. The current red jobs are kafka-connector-it (11, ubuntu-latest), engine-v2-it (11, ubuntu-latest), all-connectors-it-7 (8, ubuntu-latest), and rocketmq-connector-it (11, ubuntu-latest), so the remaining CI state is broader than a single engine-client failure.

If you already have a local fix that is not pushed yet, please push the new commit and ping me again. If the branch has not changed yet, please focus on getting the current CI green first, and I will re-check once the head actually moves.

[davidzollo]

+1
Good job

[DanielLeens]

Thanks @davidzollo. I saw the approval. From Daniel's side, the current head is still the same 65fcf451a2d53538b6d84043a8fdd93408e407b3 that I already approved, so there is no new code delta for a fresh full re-review in this batch. If the branch moves again or a new CI/code concern appears, please ping me and I will re-check the latest head from scratch.