[zhengxiang378928908-code]

Purpose of this pull request

Closes #10623. Takes over #10755 from @chl-wxp, who is kept as co-author. The takeover was proposed in #10755 on 2026-10-15, with no reply from the author since.

Root cause. With a primary key (upsert mode, CDC, or iceberg.table.primary-keys), the sink writes through Iceberg's BaseEqualityDeltaWriter. It keeps insertedRowMap, a copy of every key it has written, so that a later change of the same key can become a position delete, and it buffers those position deletes in SortingPositionOnlyDeleteWriter. Both are released only when the task writer completes, which today happens only at a checkpoint. A 100 million row upsert load with a long checkpoint interval, or a batch job that checkpoints only at the end, keeps every key in memory until it OOMs. Data files already roll by write.target-file-size-bytes, so they are not the problem.

Change.

1. flush_row_size (optional, > 0, unset by default = current behavior): IcebergRecordWriter completes the task writer after that many rows and starts a new one. Each completed task writer becomes its own WriteResult, in order.
2. Ordered commits. As @DanielLeens pointed out in [Bugfix][connectors-v2/connector-iceberg]Fix the OOM problem of iceberg when large batches of data are continuously written #10755, rotating alone is not correct in upsert mode: an equality delete written after the rotation lands in the same snapshot as the row it targets, and Iceberg applies equality deletes only to data files with a smaller sequence number, so the old row stays. IcebergAggregatedCommitter.planCommits groups the i-th results of all writers into round i; a round that carries equality deletes starts a new group, and the groups are committed as separate snapshots in order. Flink's IcebergFilesCommitter.commitDeltaTxn commits pending results one by one for the same reason.
   • Append-only writes, and checkpoints without a rotation, still produce exactly one snapshot, unchanged.
   • Snapshots of a split checkpoint carry seatunnel.checkpoint-id plus seatunnel.commit-group / seatunnel.commit-group-count. restoreCommit commits only the groups that are missing. A snapshot with the checkpoint id but without seatunnel.commit-group (single-snapshot commit, including ones written by older versions) counts as the whole checkpoint. With checkpoint id 0, each group is checked by its file paths as before.

On "rotate only where in-flight keys cannot cross": a sink cannot know when a key will not appear again, so the PR keeps rotation in all modes and makes the commit order carry the semantics, instead of restricting rotation to append-only, which would not help the upsert case reported in #10623.

Same bug on dev today, without this option. applySchemaUpdate already completes the task writer in the middle of a checkpoint. In upsert mode, an UPDATE of a row written before an ADD COLUMN in the same checkpoint leaves the old row (schemaChangeInsideCheckpointKeepsUpsertCorrect below fails on dev with [1=a, 1=a2]). The ordered commit fixes this too.

Addresses the open review items on #10755: > 0 validation, EN/ZH docs, focused tests including INSERT → rotate → DELETE/UPDATE.

Does this PR introduce any user-facing change?

Yes.

• New option flush_row_size for the Iceberg sink (docs en/zh, with when to use it and the trade-off: more and smaller files, and more snapshots for upsert/CDC).
• Behavior change, recorded in incompatible-changes.md (en/zh): an upsert or CDC checkpoint in which the task writer was completed mid-checkpoint (by flush_row_size or by a schema change) is committed as several ordered snapshots instead of one, and readers can see them one by one. Each is a prefix of the checkpoint's changes. Before, such a checkpoint was one snapshot but could keep deleted or replaced rows.

How was this patch tested?

Unit tests against real Iceberg format v2 tables (InMemoryCatalog), written through IcebergWriterFactory and committed through IcebergAggregatedCommitter, read back with IcebergGenerics (IcebergWriterRotationTest, 9 cases):

• append-only with rotation: 2 results, 1 snapshot, no group property
• upsert, unpartitioned and bucket(id, 4): INSERT 1, 2 → rotate → DELETE 1, UPDATE_AFTER 2 → INSERT 3 gives [2=b2, 3=c] in 3 snapshots
• CDC (primary key, upsert off): INSERT → rotate → UPDATE_BEFORE/UPDATE_AFTER gives only the new row
• two parallel writers with different rotation counts
• restore after only the first group was committed: commits the rest once; restoring again adds nothing
• restore of a checkpoint committed as a single snapshot (no group property): nothing re-committed
• schema change inside a checkpoint in upsert mode (the dev bug above)
• control case: the same rotated files committed as one RowDelta (the old committer) give [1=a, 2=b, 2=b2], which is the stale-row problem

Mutation check: with the round split disabled (all rounds in one group), 6 of the 9 cases fail with stale rows.

IcebergSinkConfigTest: unset by default; 0 and -1 rejected with OptionValidationException. Existing IcebergAggregatedCommitterTest / IcebergFilesCommitterTest pass unchanged. Whole connector-iceberg module: 144 tests pass; ConnectorOptionCheckTest passes.

Heap, local run (JDK 11, -Xmx512m, upsert, unpartitioned, 4,000,000 distinct keys written without a checkpoint; not part of the PR):

flush_row_size	used heap after GC	result
unset	222 MB at 0.5M rows, 403 MB at 1M rows	OutOfMemoryError before 1.5M rows
200000	37 to 79 MB through 4M rows	done, 20 results

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

[zhengxiang378928908-code]

CI status for head 43bbdeb41 (fork run 39561113519):

• iceberg-connector-it passed on JDK 11 and 17.
• The only failures are the four unit-test jobs, all on LLMBooleanOutputTest.readsLegacyFactoryWrapperWithValidationDisabled in seatunnel-transforms-v2 (InvalidClassException on AbstractMultiCatalogTransform). That is [Bug][Transform-V2] Legacy LLM wrapper fixture fails on current dev #12929: it fails the same way on current dev (bab2d089c) locally, and this PR does not touch seatunnel-transforms-v2. I'll sync dev and rerun once a fix lands.

[DanielLeens]

Thank you @zhengxiang378928908-code for taking this over from #10755 and for the very thorough PR description, the mutation check, and the real-table tests. The ordered-commit idea is exactly the right answer to the equality-delete sequence-number problem raised on the original PR. This is my first review on this PR; I traced the full writer-to-committer-to-restore path on head 43bbdeb41.

What Problem Does This PR Solve?

Pain point. With a primary key (upsert, CDC, or iceberg.table.primary-keys) the Iceberg delta task writer keeps every inserted key (insertedRowMap) plus buffered position deletes until the writer is completed. In SeaTunnel the writer is only completed in IcebergSinkWriter.prepareCommit, so a long checkpoint interval or a batch job that checkpoints once grows the heap with the row count and ends in OOM (#10623).

Fix approach. (1) New optional flush_row_size: IcebergRecordWriter.write completes the task writer after N rows and starts a new one; every completed writer becomes its own WriteResult. (2) Because a rotation puts an equality delete and the row it targets into the same snapshot (where Iceberg does not apply it, since equality deletes only apply to strictly smaller sequence numbers), IcebergAggregatedCommitter.planCommits groups the i-th results of all writers into rounds, opens a new group at each round that carries equality deletes, and IcebergFilesCommitter.doCommitInOrder commits the groups as separate, ordered snapshots tagged with seatunnel.commit-group / -count. Restore commits only missing groups.

One-sentence summary. Adds an opt-in row-count rotation of the Iceberg sink task writer to bound memory, and makes the committer commit rotated upsert/CDC checkpoints as ordered snapshots so deletes stay correct.

1. Code Change Review

1.1 Core Logic Analysis

Runtime path (normal path is untouched when the option is unset):

IcebergSinkWriter.write -> IcebergRecordWriter.write
   -> rowsInWriter++ ; if flushRowSize != null && rowsInWriter >= flushRowSize -> flush()
         flush(): writer.complete() -> writerResults.add(WriteResult) ; writer=null ; rowsInWriter=0
         next write(): writer==null -> resetWriter() (new task writer, empty insertedRowMap)
IcebergSinkWriter.prepareCommit(ckpt) -> IcebergRecordWriter.complete()
   -> flush() + return ALL accumulated WriteResults (ordered)
   -> IcebergCommitInfo(results, ckpt) -> Zeta aggregated committer
SinkAggregatedCommitterTask.notifyCheckpointComplete
   -> IcebergAggregatedCommitter.commit -> commitFiles(restoring=false)
        planCommits -> 1 group : doCommit (unchanged single snapshot)
                    -> N groups: doCommitInOrder -> one snapshot per group (RowDelta / AppendFiles)
restore: restoreCommit -> commitFiles(restoring=true) -> committedGroups(ckpt) scans snapshot chain
        -> skip groups whose seatunnel.commit-group is already present

Normal path hit: When flush_row_size is unset, rowsInWriter is incremented but flush() is never triggered mid-checkpoint, so complete() returns exactly one result per writer per checkpoint, planCommits returns a single group, and the commit goes through the old doCommit with group = -1 (no new snapshot properties). I verified that the restore branch keeps the old semantics for a single group (isAlreadyCommitted by checkpoint id, or by file paths when the id is 0). Note that the existing schema-change flush in applySchemaUpdate also produces a second result in the same checkpoint, so that path (previously buggy in upsert mode as the author demonstrates) now goes through the new ordered commit as well.

Correctness of the grouping (traced against the Iceberg semantics):

• In upsert mode BaseDeltaTaskWriter calls deleteKey before every write, so every non-empty round in upsert mode carries equality deletes and each rotation becomes its own snapshot. In CDC (non-upsert) mode only updates and deletes of keys not seen in the current writer produce equality deletes. Both are handled correctly by hasEqualityDeletes.
• Equality deletes of round i can only target rows of rounds less than i in the same checkpoint (or earlier checkpoints). Because round i starts a new group, its delete sequence number is greater than the data files of all earlier groups. Rows of round i itself are covered by in-writer position deletes. Good.
• Rounds without equality deletes are merged into the previous group. That is safe, because they contain no deletes that need ordering against earlier data. Good.
• Restore is deterministic: the plan is derived from the same persisted IcebergCommitInfo list, so the group indexes line up with the snapshot properties. A snapshot with checkpoint id but no group property is treated as the whole checkpoint, which keeps old-version snapshots safe.
• Failure in the middle of the group loop: notifyCheckpointComplete has no internal retry, the exception fails the task, and the job restarts into restoreCommit, which skips committed groups. I do not see a double-commit window in the engine path.

1.2 Compatibility Impact

Partially incompatible (documented, opt-in trigger).

• Config: new optional flush_row_size, no default, so existing job configs are unchanged. No option was renamed or removed.
• State/serialization: IcebergCommitInfo / WriteResult / IcebergSinkState are unchanged; old checkpoints restore as before.
• Table-visible behavior: a checkpoint that rotated (by flush_row_size or by the existing schema-change flush) in upsert/CDC mode is now committed as several snapshots instead of one, so readers can observe intermediate prefixes, and the snapshot count grows. This is recorded in incompatible-changes.md (en and zh), which is the right place and the right level of detail. Old-version snapshots (no commit-group) are handled on restore. Rolling back to an older engine version after a split commit is safe for data (every snapshot is valid) but an older version would not recognise the group properties; that only matters if the same checkpoint is restored by an older binary, which is an unusual downgrade.

1.3 Performance / Side-Effect Analysis

• Memory: insertedRowMap and position-delete buffers are released every N rows, so heap is bounded by N rather than the checkpoint size. The accumulated List<WriteResult> only holds file metadata, which is small. The reported heap table in the PR description is convincing.
• Files: each rotation closes all open files, so for a partitioned table a rotation can produce one small file per active partition. The docs mention "more and smaller files"; see the recommended doc note below.
• Snapshots: in upsert mode every rotation becomes a snapshot, so a 100M-row checkpoint with flush_row_size = 200000 produces about 500 snapshots from a single checkpoint, which stresses table metadata (manifest lists, snapshot history, expiry). The option is opt-in and documented, but this deserves a more explicit warning (Issue 2).
• Restore cost: committedGroups walks the whole snapshot chain (it does not stop at the first match, because it has to collect every group). This only runs on restore and is metadata-only, so it is acceptable, but see Issue 3.
• Thread safety: rowsInWriter is a plain long while writer is volatile; in Zeta the write and prepareCommit calls run on the task thread under the checkpoint lock, so this is fine and consistent with the existing field style.

1.4 Error Handling and Logging

The error handling is sound: validation throws OptionValidationException with the key and the value, IOException from complete() is propagated, and the committer logs when it splits a checkpoint and when it skips an already-committed group. I did not find a swallowed exception. I have no blocking findings in the code. These are the improvements I recommend:

Issue 1 (Medium, test coverage): no E2E coverage for the new option and for the multi-snapshot commit.

• Location: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iceberg-e2e/src/test/java/org/apache/seatunnel/e2e/connector/iceberg/IcebergSinkIT.java and IcebergSinkCDCIT.java (not changed in this PR; the new unit test is .../sink/commit/IcebergWriterRotationTest.java).
• Problem: The Iceberg connector already has sink and CDC ITs, and the repository guidelines ask for the existing IT to be extended when a connector area that has one is changed. The unit tests use InMemoryCatalog and call the writer/committer directly, so they do not exercise the real Zeta checkpoint, prepareCommit, notifyCheckpointComplete, and restore handshake with flush_row_size set.
• Risk: A regression in how the engine drives complete() and the aggregated committer (the real source of the OOM) would not be caught by the unit tests.
• Best improvement: Add one case to IcebergSinkCDCIT (or IcebergSinkIT) with a small flush_row_size (for example 2 or 3) on an upsert/CDC job with an UPDATE and a DELETE of an earlier row in the same checkpoint, and assert the final table content.

Issue 2 (Low, documentation): the snapshot-count trade-off for upsert mode is understated.

• Location: docs/en/connectors/sink/Iceberg.md (new flush_row_size section) and docs/zh/connectors/sink/Iceberg.md.
• Problem: The text says "more snapshots", but in upsert mode every rotation becomes its own snapshot (every row writes an equality delete via deleteKey), so snapshot count is roughly rows per checkpoint divided by flush_row_size. The example value 1000000 is reasonable, but users may pick a small value for memory reasons.
• Risk: Metadata bloat and slow planning on the target table if the value is too small; users cannot tell this from the current wording.
• Best improvement: Add one sentence stating that in upsert mode each rotation produces one snapshot (so keep the value large enough, and use snapshot expiry), and mention that for partitioned tables a rotation closes the files of all open partitions.

Issue 3 (Low, efficiency): committedGroups always walks the full snapshot chain.

• Location: seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergFilesCommitter.java:123-140
• Problem: The loop only stops at the end of the chain, even after all expected commit-group-count groups were found.
• Risk: Only on restore and metadata-only, so impact is small, but tables with very long unexpired history pay the cost for every restored checkpoint.
• Best improvement: Pass the expected group count (it is stored in seatunnel.commit-group-count) and break as soon as all groups are found, or break once the chain goes below the oldest relevant snapshot. This is optional.

2. Code Quality Assessment

2.1 Coding Standards

The code is clear and follows the module's style; new public/package-visible methods (planCommits, doCommitInOrder, the new constants, RecordWriter.complete contract) carry meaningful Javadoc that explains the ordering requirement, which is exactly the kind of comment this logic needs. OptionRule registration, option validation and ConnectorOptionCheckTest are covered. No unused imports or wildcard imports seen. The RecordWriter.complete() Javadoc change makes an implicit contract explicit, which is a nice touch.

2.2 Test Coverage and Test Stability

Rating: Stable (new tests, no existing tests modified). IcebergWriterRotationTest (9 cases, real format-v2 tables via InMemoryCatalog, read back through IcebergGenerics) covers append-only, upsert (unpartitioned and bucketed), CDC, parallel writers with different rotation counts, restore of partial groups, restore of an old single-snapshot checkpoint, the schema-change-in-checkpoint bug, and a control case that proves the stale-row problem. These tests are deterministic (no sleeps, no containers, no timing), and the control case plus the mutation check in the description show they fail when the grouping is removed. IcebergSinkConfigTest covers unset, 0, and -1. No existing assertions were weakened. The only gap is the E2E coverage in Issue 1.

2.3 Documentation Updates

English and Chinese Iceberg.md and incompatible-changes.md are all updated, with an option-table row, a detailed section, and an example. I checked the documented behavior against the code: default unset, > 0 validation, the group property names, and the restore rule all match IcebergSinkOptions, IcebergSinkConfig, and IcebergFilesCommitter. See Issue 2 for the one suggested addition.

3. Architectural Soundness

3.1 Fix Quality

Precise fix at the correct layer. The memory problem is in the writer lifecycle and the ordering problem is in the committer, and both are fixed in the Iceberg connector itself without touching engine/core/API classes. The approach mirrors Flink's IcebergFilesCommitter.commitDeltaTxn, which commits pending results one by one for the same sequence-number reason.

3.2 Alternatives

Restricting rotation to append-only would not address the upsert OOM in #10623, and rotating without ordered commits would silently keep stale rows, so I agree with the chosen design.

3.3 Maintainability

The grouping logic is isolated in a small static planCommits method that is directly unit-testable, and restore reuses the same plan. That is easy to maintain.

3.4 Historical-Version Compatibility

Old checkpoints, old snapshots without group properties, and jobs that do not set flush_row_size behave as before. No serialized state format changed. The only visible change is the documented multi-snapshot commit for rotated upsert/CDC checkpoints.

4. Issue Summary

Number	Issue	Location	Severity
1	No E2E case for flush_row_size with the real checkpoint/commit/restore path	connector-iceberg-e2e IcebergSinkCDCIT / IcebergSinkIT	Medium
2	Docs understate that each rotation is one snapshot in upsert mode (and closes all partition files)	docs/en/connectors/sink/Iceberg.md, docs/zh/connectors/sink/Iceberg.md	Low
3	committedGroups walks the whole snapshot chain without early exit	IcebergFilesCommitter.java:123-140	Low

5. Merge Recommendation

Conclusion: Ready to merge after fixes

Blockers:

• CI: the Build check on this head is red. The author's note in the PR (fork run) states the only failures are the four unit-test jobs on LLMBooleanOutputTest.readsLegacyFactoryWrapperWithValidationDisabled in seatunnel-transforms-v2, while iceberg-connector-it passed on JDK 11 and 17. This is tracked in #12929, which is still open with no fix PR yet. I checked that this PR does not touch seatunnel-transforms-v2, so it looks unrelated to the diff. Please re-run the failed jobs after #12929 is fixed and the branch is synced with dev.

Recommended fixes (non-blocking): Issue 1 (E2E case), Issue 2 (doc sentence), Issue 3 (optional early exit).

Overall assessment: The design is correct, the equality-delete ordering is carefully reasoned, the default behavior is preserved, restore is idempotent per group, docs (en/zh) and the incompatible-changes entry are in place, and the tests are deterministic and meaningful. I found no non-CI blocker in the code. Thank you again for the careful work and for crediting the original author. Once CI is green and Issue 1 is addressed (or explicitly deferred), this looks good to me from my side.

[zhengxiang378928908-code]

@DanielLeens Thanks for the review. Pushed e218a3a93 for the three items:

1. E2E: two new cases in IcebergSinkIT, both with flush_row_size = 2 and FakeSource changelog rows, so rows 1 and 3 are updated and row 2 is deleted by later task writers than the ones that wrote them, all in one checkpoint:

   • testFlushRowSizeUpsertE2e: upsert mode, unpartitioned.
   • testFlushRowSizeCdcPartitionedE2e: changelog (no upsert) into a table partitioned by region.

   Both assert the exact final content ({1: A2, 3: C2, 4: D}) and that the checkpoint was committed as ordered snapshot groups 0..n-1, each exactly once. They are disabled on Spark, which does not carry the row kind (same as IcebergSinkCDCIT).
   Fork run 39605402657: iceberg-connector-it passed on JDK 11 and 17; both cases ran on all 5 non-Spark containers on each, and IcebergSinkIT is 46/46.

2. Docs (EN/ZH): in upsert mode every rotation becomes its own snapshot, so a checkpoint is committed as about rows per writer / flush_row_size snapshots; a rotation closes the files of all open partitions; keep the value large and expire snapshots.

3. committedGroups: returns as soon as all groups of the checkpoint are found.

The four unit-test jobs still fail only on LLMBooleanOutputTest.readsLegacyFactoryWrapperWithValidationDisabled (#12929). I'll sync dev and re-run once that is fixed.

[DanielLeens]

Thank you @zhengxiang378928908-code for the fast and careful follow-up. All three items from my last review are addressed in e218a3a93, and I re-read the whole diff from scratch on this head rather than only the delta. Sorry for missing one point in the last review (Issue 1 below, a carryover): I called the downgrade case safe for data, and a closer look at the old restore path shows it is not in one narrow window.

Status of my previous findings

• Old Issue 1 (no E2E for the real checkpoint/commit path): resolved. testFlushRowSizeUpsertE2e and testFlushRowSizeCdcPartitionedE2e in IcebergSinkIT use flush_row_size = 2, assert the exact final rows and assert that split checkpoints carry groups 0..n-1 once and in order.
• Old Issue 2 (docs understate the snapshot trade-off): resolved in EN and ZH (one snapshot per rotation in upsert mode, partition files closed together, keep the value large, expire snapshots).
• Old Issue 3 (committedGroups full-chain walk): resolved, it now returns once all groups are found.

What Problem Does This PR Solve?

Pain point. With a primary key (upsert, CDC, or iceberg.table.primary-keys) the Iceberg delta task writer keeps every inserted key (insertedRowMap) and buffered position deletes until the task writer completes. In SeaTunnel that only happens in IcebergSinkWriter.prepareCommit, so a long checkpoint interval, or a batch job that checkpoints once, grows the heap with the row count and ends in OOM (#10623).

Fix approach. A new optional flush_row_size completes the task writer every N rows (each completed writer is its own WriteResult). Because an equality delete is applied by Iceberg only to data files with a smaller sequence number, IcebergAggregatedCommitter.planCommits splits a checkpoint into ordered groups and IcebergFilesCommitter.doCommitInOrder commits them as separate snapshots tagged seatunnel.commit-group / seatunnel.commit-group-count. Restore commits only the missing groups.

One-sentence summary. Opt-in row-count rotation of the Iceberg sink task writer to bound memory, plus ordered multi-snapshot commits so upsert/CDC deletes stay correct.

1. Code Change Review

1.1 Core Logic Analysis

Does the normal user path reach the changed logic? Only partially, by design. With flush_row_size unset nothing rotates mid-checkpoint, complete() returns one result per writer, planCommits returns a single group and the old doCommit runs with group = -1 (no new snapshot properties). The existing schema-change rotation (applySchemaUpdate) also yields two results in one checkpoint, so that path now takes the ordered commit; the author shows it was a real stale-row bug on dev. The reported OOM is real: insertedRowMap is released only on complete().

IcebergSinkWriter.write -> IcebergRecordWriter.write
   rowsInWriter++ ; >= flushRowSize -> flush(): writer.complete() -> writerResults.add ; writer=null ; rowsInWriter=0
   next write(): writer==null -> resetWriter()
IcebergSinkWriter.prepareCommit(ckpt) -> IcebergRecordWriter.complete() -> flush() + all ordered results
   -> IcebergCommitInfo -> IcebergAggregatedCommitter.commit -> commitFiles(restoring=false)
        planCommits: 1 group -> doCommit (unchanged) ; N groups -> doCommitInOrder (one snapshot per group)
restoreCommit -> commitFiles(restoring=true) -> committedGroups(ckpt) (snapshot-chain walk, early exit)
        -> skip groups already present; legacy snapshot with checkpoint id and no group property = whole checkpoint

Correctness (re-traced). Equality deletes of round i can only target rows of earlier rounds or earlier checkpoints, and a round with equality deletes opens a new group, so its delete sequence number exceeds all earlier data. Rounds without equality deletes merge into the previous group, which is safe. Rows of the same round are covered by in-writer position deletes. Restore is deterministic because the plan is derived from the persisted commit infos. The restore of a checkpoint that an older version committed as one snapshot is treated as fully committed, so there is no double commit on upgrade.

1.2 Compatibility Impact

Partially incompatible (documented in incompatible-changes.md EN/ZH, opt-in trigger).

• Config: new optional flush_row_size, no default, nothing renamed or removed.
• State: IcebergCommitInfo, WriteResult, IcebergSinkState are unchanged; old checkpoints restore as before.
• Table-visible: a rotated upsert/CDC checkpoint is committed as several snapshots; readers can observe prefixes.
• Downgrade (see Issue 1): an older binary cannot resume a half-committed split checkpoint.

1.3 Performance / Side-Effect Analysis

• Memory is bounded by N rows per writer; the accumulated List<WriteResult> holds only file metadata.
• A rotation closes the files of all open partitions, so small-file count grows with partition count (documented).
• Upsert mode produces about rows/N snapshots per checkpoint, committed sequentially inside notifyCheckpointComplete. At the documented example value this is bounded, but a very small value on a 10^10-row job means thousands of sequential table commits, each with manifest rewrite and optimistic-retry exposure, before the checkpoint completes. The docs now warn about metadata bloat; the commit-latency side is not mentioned (Issue 3).
• With checkpoint id 0 (no id property written) each group is checked by scanning all manifests, so restore cost is groups x manifests. Rare path, metadata only.
• Orphan files: files closed by a rotation are not committed until the checkpoint completes, so a job failure leaves them as orphans, the same class of leftovers as before, only more numerous. Normal orphan-file cleanup applies.

1.4 Error Handling and Logging

Validation throws OptionValidationException with key and value, IOException from complete() is propagated, and the committer logs splits and skipped groups. No swallowed exception found. Formal issues:

Issue 1 (Low, compatibility documentation): downgrade after a partially committed split checkpoint can lose the remaining groups. (carryover, not raised before; I previously described downgrade as data-safe.)

• Location: seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergFilesCommitter.java (SNAPSHOT_PROPERTY_CHECKPOINT_ID is written on every group), docs/en/introduction/concepts/incompatible-changes.md.
• Problem: every group snapshot carries seatunnel.checkpoint-id. If the job crashes after group 0 of N and is restored by an older binary, the old isAlreadyCommitted(checkpointId) finds that id and skips the whole checkpoint, so groups 1..N-1 are never committed.
• Potential risk: silent loss of the later changes of that one checkpoint, only for a crash mid-commit followed by a downgrade before the restore completes.
• Best improvement: add one sentence to the migration guide (EN/ZH): do not downgrade a job while a split checkpoint may be partially committed; restore with the new version first. Code change is not needed.
• Severity: Low. Raised by another reviewer: No.

Issue 2 (Low, observability): the first planned group can be empty.

• Location: IcebergAggregatedCommitter.java planCommits (the groups.isEmpty() || hasEqualityDeletes branch) and IcebergFilesCommitter.commit ("Nothing to commit" branch).
• Problem: the first round is always opened as a group even if its results contain no records (for example a schema-change flush with no rows). Such a group writes no snapshot, so a table can show commit-group-count = 3 with groups 1 and 2 only.
• Potential risk: no data risk (restore treats the missing group as a no-op, and the early exit simply never triggers), but external tooling or assertions that expect groups 0..n-1 would be confused.
• Best improvement: drop empty rounds in planCommits, or merge an empty first round into the next group. Optional.
• Severity: Low. Raised by another reviewer: No.

Issue 3 (Low, documentation): sequential commit latency of many groups is not mentioned.

• Location: docs/en/connectors/sink/Iceberg.md and docs/zh/connectors/sink/Iceberg.md, flush_row_size section.
• Problem: the text covers metadata bloat but not that all groups of a checkpoint are committed one after another before the checkpoint completes.
• Potential risk: users with very small values see long commit phases and slow checkpoints.
• Best improvement: one sentence next to the "keep it large" advice.
• Severity: Low. Raised by another reviewer: No.

2. Code Quality Assessment

2.1 Coding Standards

New core methods and fields (planCommits, doCommitInOrder, group constants, rowsInWriter, RecordWriter.complete() contract) carry comments that explain the ordering requirement. No wildcard imports. Fine.

2.2 Test Coverage and Test Stability

Rating: Stable. IcebergWriterRotationTest (9 cases on real format-v2 tables) and IcebergSinkConfigTest are deterministic with no sleeps. The new E2E cases use FakeSource with fixed rows, parallelism = 1, BATCH mode, and assertions on the exact final rows plus snapshot-group order (IcebergSinkIT.java, assertFlushRowSizeJob); no fixed waits, and Spark is disabled with a stated reason. Residual gap: E2E does not exercise restore after a crash, which is covered at unit level only. No existing assertion was weakened.

2.3 Documentation Updates

docs/en and docs/zh Iceberg.md plus incompatible-changes.md (EN/ZH) are updated and match the code. See Issues 1 and 3 for two additional sentences.

3. Architectural Soundness

3.1 Elegance

Precise fix, in the connector only, no engine/core/API change. It mirrors Flink's IcebergFilesCommitter.commitDeltaTxn.

3.2 Maintainability

Grouping lives in one static, directly testable planCommits, and restore reuses the same plan.

3.3 Extensibility

The group properties make later optimizations (for example a commit-group size cap) possible without changing state formats.

3.4 Historical-Version Compatibility

Jobs that do not set the option, old checkpoints and old snapshots behave as before; the only exception is the documented downgrade window in Issue 1.

4. Issue Summary

Number	Issue	Location	Severity
1	Downgrade after a partially committed split checkpoint skips remaining groups; not in migration guide	IcebergFilesCommitter.java, incompatible-changes.md	Low
2	First planned group can be empty, leaving a gap in group numbering	IcebergAggregatedCommitter.planCommits	Low
3	Sequential commit latency of many groups not documented	Iceberg.md EN/ZH	Low

5. Merge Recommendation

Conclusion: Ready to merge after fixes

1. Blockers - must be fixed

   • CI only: the Build check on e218a3a93 is red. In the fork run, iceberg-connector-it passed (JDK 11 and 17) and the only failures are the four unit-test jobs, all on LLMBooleanOutputTest.readsLegacyFactoryWrapperWithValidationDisabled in seatunnel-transforms-v2 (InvalidClassException). That is #12929, still open, with a fix PR #12938 also open. This PR does not touch seatunnel-transforms-v2, so it is unrelated to the diff. Once #12938 lands, please sync the latest dev and re-run the failed jobs.

2. Recommended fixes - non-blocking

   • Issues 1 to 3 above (two doc sentences and an optional empty-round cleanup).

Overall assessment. The design is correct, the equality-delete ordering is carefully reasoned, the default path is unchanged, restore is idempotent per group, and the new E2E cases close the previous gap. I found no source, test, or compatibility blocker beyond CI. Thank you again for the careful work and for crediting the original author.

[zhengxiang378928908-code]

Thanks for the second review. I addressed the three suggestions in e89377a:

1. Downgrade: incompatible-changes.md (EN/ZH) now says that a job stopped while a split checkpoint is being committed must be restored with this version or later, because an older version skips the whole checkpoint once it finds any of its snapshots.
2. Results without records: a group that has no records no longer ends a commit group. The following rounds join it, so a leading empty flush (for example right after a schema change) doesn't produce a group that commits no snapshot. Nothing is dropped, and a checkpoint whose results are all empty is planned as before. resultWithoutRecordsDoesNotOpenAGroup covers this. Without the change it gets groups 1/4, 2/4, 3/4, with no snapshot for group 0.
3. Latency: the flush_row_size docs now say that the snapshots of a checkpoint are committed one after another before the checkpoint completes, so a very small value also makes checkpoints take longer.

[DanielLeens]

Thank you @zhengxiang378928908-code for the quick follow-up on all three points from my last round (the downgrade note in incompatible-changes.md, the empty-first-group fix, and the commit-latency sentence in the docs), and for the earlier E2E cases. The branch was rebased (new head 1884e8473), so I re-read the whole diff from scratch rather than only the delta. The only gate I cited last time was the red Build check, and Build is now SUCCESS on this head. My conclusion therefore changes to approve. This review is static analysis only; I did not build or run anything locally.

Status of my previous findings

• Downgrade after a partially committed split checkpoint: resolved. incompatible-changes.md (EN and ZH) now tells users to restore such a job with this version or later, and explains why.
• First planned group can be empty: resolved in IcebergAggregatedCommitter.planCommits. A group without records no longer closes the group, so later rounds join it (hasRecords check), and group numbering has no gap. resultWithoutRecordsDoesNotOpenAGroup covers it.
• Sequential commit latency not documented: resolved in the flush_row_size section.

What Problem Does This PR Solve?

Pain point. With a primary key (upsert, CDC, or iceberg.table.primary-keys) the Iceberg delta task writer keeps every inserted key and buffered position delete until it completes. In SeaTunnel it only completes in IcebergSinkWriter.prepareCommit, so a long checkpoint interval, or a batch job that checkpoints once, grows the heap with the row count and ends in OOM (#10623).

Fix approach. The new optional flush_row_size completes the task writer every N rows, and each completed writer becomes its own WriteResult. Since Iceberg applies an equality delete only to data files with a smaller sequence number, planCommits splits a checkpoint into ordered groups and doCommitInOrder commits them as separate snapshots tagged seatunnel.commit-group / seatunnel.commit-group-count. Restore commits only the missing groups.

One-sentence summary. Opt-in row-count rotation of the Iceberg sink writer to bound memory, plus ordered multi-snapshot commits so upsert/CDC deletes stay correct.

1. Code Change Review

1.1 Core Logic Analysis

IcebergRecordWriter.write: rowsInWriter++ ; >= flushRowSize -> flush() (complete task writer, add WriteResult, writer=null, rowsInWriter=0)
IcebergSinkWriter.prepareCommit -> IcebergRecordWriter.complete() -> all ordered results -> IcebergCommitInfo
IcebergAggregatedCommitter.commit -> commitFiles(restoring=false)
    planCommits: 1 group -> doCommit (unchanged, group = -1) ; N groups -> doCommitInOrder (one snapshot per group)
restoreCommit -> commitFiles(restoring=true)
    1 group -> isAlreadyCommitted (old semantics) ; N groups -> committedGroups(ckpt) (early exit) -> commit only missing groups

Normal path. With flush_row_size unset nothing rotates mid-checkpoint, so planCommits yields one group and the old doCommit runs with no new snapshot properties. The restore branch for a single group keeps the old isAlreadyCommitted behavior. The existing schema-change flush also yields two results in one checkpoint and now takes the ordered path, which fixes the stale-row case the author demonstrated.

Correctness of the new empty-group handling (re-traced). A group is only extended instead of opened when the previous group has no records at all (data or delete files with recordCount > 0). In that case no earlier data file exists in that group, so merging a later equality-delete round into it cannot break the sequence-number ordering. Rounds with records and equality deletes still open a new group, and rounds without equality deletes still merge into the previous one. A checkpoint whose results are all empty is planned as a single group and ends in the existing "nothing to commit" branch, as before. The plan is derived only from the persisted commit infos, so restore stays deterministic.

Restore. committedGroups treats a snapshot with the checkpoint id and no group property as the whole checkpoint (old-version commit), collects group indexes otherwise, and returns early once all groups are found. For checkpoint id 0 it falls back to per-group file-path detection. I found no double-commit window: notifyCheckpointComplete does not retry internally, and a failure restarts the job into restoreCommit, which skips committed groups.

1.2 Compatibility Impact

Partially incompatible, documented and opt-in.

• Config: new optional flush_row_size, no default, nothing renamed or removed.
• State: IcebergCommitInfo, WriteResult, IcebergSinkState unchanged; old checkpoints and old snapshots restore as before.
• Table-visible: a rotated upsert/CDC checkpoint is committed as several snapshots; this and the downgrade caveat are recorded in incompatible-changes.md (EN/ZH).

1.3 Performance / Side-Effect Analysis

• Memory is bounded by N rows per writer; the accumulated List<WriteResult> only holds file metadata.
• A rotation closes files of all open partitions (documented), and upsert mode produces about rows/N snapshots per checkpoint, committed sequentially (documented, including the latency effect).
• hasRecords / hasEqualityDeletes iterate file metadata of one checkpoint only, which is negligible next to the table commits.
• rowsInWriter is a plain long; write and prepareCommit run on the task thread under the checkpoint lock, consistent with the surrounding fields.

1.4 Error Handling and Logging

Option validation throws OptionValidationException with key and value, IOException from complete() propagates, and the committer logs splits and skipped groups. No swallowed exception found. No blocking issues found.

2. Code Quality Assessment

2.1 Coding Standards

New core methods and fields carry comments that explain the ordering requirement and the empty-group rule. No wildcard imports; license headers are present on new files.

2.2 Test Coverage and Test Stability

Rating: Stable. IcebergWriterRotationTest (real format-v2 tables via InMemoryCatalog, no sleeps or containers) and IcebergSinkConfigTest are deterministic. The two E2E cases in IcebergSinkIT use fixed FakeSource rows, parallelism = 1, BATCH mode, exact final-content assertions plus group-order assertions, and Spark is disabled with a stated reason. No existing assertion was weakened. Residual gap (non-blocking): crash-and-restore is covered at unit level only.

2.3 Documentation Updates

docs/en and docs/zh for Iceberg.md and incompatible-changes.md are updated and match the code.

3. Architectural Soundness

3.1 Elegance

Precise fix, confined to the Iceberg connector; no engine/core/API change.

3.2 Maintainability

Grouping is isolated in one static, directly testable planCommits, reused by restore.

3.3 Extensibility

The group properties allow later refinements (for example a group-size cap) without changing state formats.

3.4 Historical-Version Compatibility

Jobs without the option, old checkpoints and old snapshots behave as before; the one documented exception is the downgrade window during a half-committed split checkpoint.

4. Issue Summary

Number	Issue	Location	Severity
-	No blocking or non-blocking issues found on this head	-	-

5. Merge Recommendation

Conclusion: Ready to merge

CI facts for head 1884e8473: Build is SUCCESS, mergeable_state is blocked only because a required human approval is still pending, there are no merge conflicts, and the branch is 12 commits behind dev (metadata only, no action needed).

Blockers: none.
Recommended non-blocking: none beyond a possible future E2E for crash-and-restore with a split checkpoint.

Overall assessment. The design is correct, the equality-delete ordering is carefully reasoned, the default path is unchanged, restore is idempotent per group, docs and the migration note are complete, and tests are deterministic and meaningful. All earlier findings are closed. Thank you for the careful and responsive work, and for crediting the original author of #10755. A maintainer with write access can do the final approval and merge.

[DanielLeens]

Thanks @zhengxiang378928908-code for keeping the branch up to date. Since my approval on 1884e8473, the branch was rebased onto the latest dev, and the new head is fb104a5e123486fec8acb4d5bbb3f3608a5022fd. I ran a range-diff of the three PR commits before and after the rebase. The connector code, the unit tests, the E2E cases and both Iceberg.md pages are identical. The only difference is where the new section sits in incompatible-changes.md (EN/ZH): it is still under ## dev, now after the JSON entry that came in from dev. I also checked that no dev commit since the old base touched connector-iceberg, connector-iceberg-e2e or the sink API, so the rebase cannot change the behavior I reviewed. I still re-read the full diff of this head. This is a static review only.

What Problem Does This PR Solve?

• User pain point: with a primary key (upsert, CDC or iceberg.table.primary-keys), the Iceberg delta task writer keeps every inserted key and every buffered position delete until it completes. In SeaTunnel it only completes in IcebergSinkWriter.prepareCommit, so a long checkpoint interval, or a batch job that checkpoints once, grows the heap with the row count and ends in OOM (#10623).
• Fix approach: the new optional flush_row_size completes the task writer every N rows, and each completed writer becomes its own WriteResult. Iceberg applies an equality delete only to data files with a smaller sequence number. So planCommits splits a checkpoint into ordered groups, and doCommitInOrder commits them as separate snapshots tagged seatunnel.commit-group / seatunnel.commit-group-count. Restore commits only the missing groups.
• One-sentence summary: opt-in row-count rotation of the Iceberg sink writer to bound memory, plus ordered multi-snapshot commits so upsert/CDC deletes stay correct.

1. Code Change Review

1.1 Core Logic Analysis

IcebergRecordWriter.write: rowsInWriter++ ; >= flushRowSize -> flush() (complete task writer, add WriteResult, reset)
IcebergSinkWriter.prepareCommit -> IcebergRecordWriter.complete() -> ordered results -> IcebergCommitInfo
IcebergAggregatedCommitter.commit -> commitFiles(restoring=false)
    planCommits: 1 group -> doCommit (unchanged, no group properties) ; N groups -> doCommitInOrder
restoreCommit -> commitFiles(restoring=true)
    1 group -> isAlreadyCommitted (old semantics) ; N groups -> committedGroups(ckpt) -> commit only missing groups

• Default path unchanged: without flush_row_size, nothing rotates in the middle of a checkpoint, so there is one group and the old doCommit runs. The existing schema-evolution flush also yields two results in one checkpoint, and it now takes the ordered path, which fixes the stale-row case.
• Grouping is correct: a new group opens only for a round with records and equality deletes after a group with records. Empty rounds do not open a group (hasRecords), so numbering has no gaps. The plan is derived only from persisted commit infos, so restore is deterministic.
• Restore is idempotent per group: committedGroups treats an old-version snapshot (checkpoint id without group property) as the whole checkpoint, and returns early once all groups are found.

1.2 Compatibility Impact

Partially incompatible, opt-in and documented. There is one new optional option and no rename or default change. State classes are unchanged. A rotated upsert/CDC checkpoint is committed as several snapshots, and the downgrade caveat for a half-committed split checkpoint is in incompatible-changes.md (EN/ZH).

1.3 Performance / Side-Effect Analysis

Memory is bounded by N rows per writer, and the accumulated results hold only file metadata. A rotation closes the files of all open partitions, and upsert mode produces about rows/N snapshots per checkpoint, committed one after another. Both effects are documented.

1.4 Error Handling and Logging

Option validation throws with key and value, IOException propagates, and the committer logs splits and skipped groups. No swallowed exceptions. No blocking issues found.

2. Code Quality Assessment

2.1 Coding Standards

New core methods and fields explain the ordering requirement and the empty-group rule. License headers are present, and there are no wildcard imports.

2.2 Test Coverage and Test Stability

Rating: Stable. IcebergWriterRotationTest (real format-v2 tables through InMemoryCatalog, no sleeps) and IcebergSinkConfigTest are deterministic. The two IcebergSinkIT cases use fixed rows, parallelism = 1, BATCH mode, and exact final-content and group-order assertions. No existing assertion was weakened. Residual gap (non-blocking): crash-and-restore of a split checkpoint is covered at unit level only.

CI for this exact head: the fork Build run 40073958586 on fb104a5e completed with success, including iceberg-connector-it on JDK 11 and 17, all unit-test legs and both updated-modules-integration-test parts. The branch is 0 behind / 3 ahead of dev and has no conflicts. The merge state is only waiting for a required approval.

2.3 Documentation Updates

docs/en and docs/zh for Iceberg.md and incompatible-changes.md are updated and match the code.

3. Architectural Soundness

3.1 Elegance

Precise fix, confined to the Iceberg connector. No engine, core or API change.

3.2 Maintainability

Grouping is isolated in one static, testable planCommits, and restore reuses it.

3.3 Extensibility

The group properties leave room for later refinements (for example a group-size cap) without a state-format change.

3.4 Historical-Version Compatibility

Jobs without the option, old checkpoints and old snapshots behave as before. The one documented exception is a downgrade during a half-committed split checkpoint.

4. Issue Summary

Number	Issue	Location	Severity
-	No blocking or non-blocking issues found on this head	-	-

5. Merge Recommendation

Conclusion: Ready to merge

1. Blockers: none. No blocking issues found.
2. Recommended (non-blocking): a future E2E for crash-and-restore with a split checkpoint.

Overall assessment: unchanged from my previous approval. The design is correct, the default path is untouched, restore is idempotent per group, docs and the migration note are complete, and CI is green on this exact head. I am re-approving so the approval is bound to the head that will be merged. A maintainer with write access can do the final approval and merge. Thank you for the careful work.

[DanielLeens]

A small correction to the merge-gate wording in my previous review: the exact-head fork Build remains SUCCESS for fb104a5e123486fec8acb4d5bbb3f3608a5022fd, but the Apache PR signal workflow is awaiting approval and no required Build result is attached to the PR's head rollup. So the remaining process is not just reviewer approval: a write-capable maintainer must also approve/inspect that workflow and attach/accept the corresponding exact-head result.

My previous source-level clearance is unchanged; this is not another full source review or a request for a new source push. There is no failed test here to repair by an unrelated dependency change or a speculative sync. Branch drift is recorded separately, and a maintainer's final approval remains distinct from my read-only technical conclusion.

[DanielLeens]

Thanks for the contribution. The CI fact cited in my previous conclusion has changed; this follow-up updates that gate on the unchanged reviewed head, without issuing another full source review.

Gate snapshot (2026-10-28T15:43:21+00:00), head fb104a5e123486fec8acb4d5bbb3f3608a5022fd: Build is SUCCESS.

CI was previously the only remaining technical gate in my conclusion; CI is now passing, so that CI objection is withdrawn and my technical conclusion is clear to proceed. Daniel currently has READ permission. I have not changed the effective GitHub review state to APPROVED; a write-capable maintainer must perform the final approve/dismiss action required by branch protection.

Separately, GitHub reports merge conflicts with dev. Please sync the latest base and resolve the conflicts first; the resulting new head needs checks and re-review. This conflict gate is independent of CI and existing source findings.

[DanielLeens]

Thanks for keeping the branch current and for the detailed E2E evidence in the discussion. I re-read all 16 changed files and the write/prepare/commit/restore path. The earlier grouping, empty-result and downgrade points remain addressed; no blocking source issue was found.

What Problem Does This PR Solve?

User pain point: A long checkpoint keeps the Iceberg delta writer's remembered keys alive, allowing large upsert or CDC loads to exhaust heap.

Fix approach: Optionally rotate task writers after flush_row_size rows, preserve completion order and commit equality-delete groups as ordered snapshots with per-group restore detection.

One-sentence summary: With a limit of two rows, inserts for IDs 1 and 2 complete one writer; a later delete of 1 and update of 2 must receive a later Iceberg sequence number so only the latest rows remain.

Review report time: 2026-10-29T11:51:51.921233-04:00. Current head 50e6f5355ff71b39b0f0411ec1fee8bdb6ee27bc; base dev at 886a3ac6e59aad5e46755761cf9fb8ac2a6d3df9. Checks: Signal pull request update=SUCCESS; Build=SUCCESS. Check references: Signal pull request update Build. Compare: diverged, ahead 3, behind 5; merge state blocked / BLOCKED.

1. Code Change Review

1.1 Core Logic Analysis

seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/writer/IcebergRecordWriter.java:84-105,190-207 counts only successfully written rows and completes/reset writers at the configured limit. Config/options/factory register an optional positive long without changing the default. seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergAggregatedCommitter.java:89-172 plans rounds from each persisted writer result list, opens a later group only for nonempty equality deletes after records and folds empty rounds into existing groups. seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergFilesCommitter.java:88-144,233-290 commits groups in order, tags snapshots and scans the selected branch on restore. The one-group path keeps the existing snapshot commit; schema-change rotations also benefit even when the option is unset.

Before:

writer.write(record);
// task writer completed at checkpoint
filesCommitter.doCommit(flattenResults(commitInfos), checkpointId);

After:

writer.write(record);
if (++rowsInWriter >= flushRowSize) { flush(); } // only when option is set
List<List<WriteResult>> groups = planCommits(commitInfos);
// multiple groups -> doCommitInOrder; restore skips recorded groups

Key Findings

• Normal write/prepareCommit and recovery commit paths reach the changed logic.
• This is a precise key-retention fix plus the ordering correction required by early rotation.
• Append-only results remain one snapshot; equality deletes after rotation require a later group.
• Group planning is deterministic from persisted result order, including unequal writer counts and empty leading results.
• Restore skips existing groups and treats an older ungrouped checkpoint snapshot as complete; crash-level E2E remains a useful follow-up.

Complete runtime path

FactoryUtil -> IcebergSinkFactory optionRule -> IcebergSinkConfig positive limit check
 -> IcebergSink.createWriter -> IcebergSinkWriter.write -> IcebergRecordWriter
 -> task writer write -> optional threshold flush -> ordered WriteResult list
checkpoint -> IcebergSinkWriter.prepareCommit(checkpointId) -> writer.complete -> IcebergCommitInfo
 -> SinkAggregatedCommitterTask aggregation -> IcebergAggregatedCommitter.combine
 -> planCommits -> one snapshot or ordered groups -> Iceberg AppendFiles/RowDelta on selected branch
 -> snapshot checkpoint/group metadata
recovery -> SinkAggregatedCommitterTask.restoreState (:275) -> restoreCommit
 -> same group plan -> committedGroups scans selected branch ancestry
 -> skip durable groups -> commit missing suffix -> checkpoint completion
A schema change also flushes a task writer, so it can activate the ordering fix without the new limit.

The effects depend on the entry and preconditions above. Recovery, batching and connector lifecycle retain their existing contracts unless explicitly described here; source inspection is not runtime proof.

1.2 Compatibility Impact

Partially incompatible. The new optional positive long leaves the default row-count rotation disabled and does not rename options or change commit-info state classes. Rotated upsert/CDC and schema-evolution checkpoints may become multiple visible snapshots instead of one. Both incompatible-change guides explain prefix visibility, group metadata, older ungrouped checkpoints and the prohibition on downgrading during a partially committed grouped checkpoint. No configuration conversion is needed; restore such checkpoints with this or a later version.

1.3 Performance / Side-Effect Analysis

The open delta writer's remembered-key memory is bounded by the limit, while accumulated file metadata still grows until checkpoint. Rotation closes all open partition files; a small limit increases files, serial snapshot commits, table metadata and checkpoint latency. Both guides explicitly explain these costs. No new executor, network retry policy or shared lock is introduced; Iceberg commit I/O increases per group and the recovery scan returns once all groups are found. This is not a constant-total-memory or single-snapshot atomicity promise.

1.4 Error Handling and Logging

Nonpositive limits fail with the option name/value. Writer completion exceptions propagate; ordered commits do not pretend that an interrupted prefix is the entire checkpoint. Logs identify grouped checkpoints and skipped groups. The existing file-presence fallback for nonpositive checkpoint IDs and branch-head selection are retained. No credential logging or swallowed new exception was found.

Issue 1: Document the restore-group detection contract

• Location: seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergFilesCommitter.java:111
• Problem description: The new committedGroups method is the correctness boundary for partial checkpoint recovery, but has no method-level explanation of its three modes: nonpositive-ID file fallback, positive-ID branch ancestry and legacy ungrouped snapshot compatibility.
• Potential risk: Maintenance can accidentally change the legacy/full-checkpoint interpretation or group completeness shortcut without understanding why it exists.
• Best improvement / root-cause and location assessment: Add focused Javadoc describing those modes, deterministic group identity, selected-branch scope and the snapshot-history assumptions. No behavior change or broad refactor is needed. Sorry for omitting this documentation requirement in my earlier review; this is a carryover, and it does not invalidate the inspected algorithm.
• Severity: Medium; non-blocking.
• Raised by another reviewer: No.

2. Code Quality Assessment

2.1 Coding Standards

The complete changed-file set, imports and removed lines were inspected. No new option, protocol field, public API or connector registration is added. The important behavior and its limits are explained above; test-boundary documentation should explain the regression it protects.

2.2 Test Coverage and Test Stability

Stability rating: Stable. IcebergWriterRotationTest.java:85-318 uses new per-test InMemoryCatalog instances, fixed rows, actual format-v2 files/readers, sorted values preserving duplicates, deterministic partial-commit replay and closed read iterables/catalogs. It covers append-only, partitioned/unpartitioned upsert, CDC, unequal writer counts, empty results, schema flush and repeated restore. IcebergSinkIT.java:197-294 resets each target location per invocation, uses parallelism 1 BATCH fixtures with fixed rows, checks exact final content/duplicate IDs and ordered complete snapshot groups. No new sleeps, clock assumptions, host ports or external data dependence appear. Spark changelog cases are explicitly disabled. No process-crash/distributed recovery E2E or executed heap bound is claimed; the author's reported CI results are distinct from local verification.

Local verification results: exact-head git fetch and isolated review branch seatunnel-review-12937-20261029 created; full git diff 886a3ac6e59aad5e46755761cf9fb8ac2a6d3df9...50e6f5355ff71b39b0f0411ec1fee8bdb6ee27bc read (16 files); git diff --check passed; git show, rg and line-numbered source/caller/config/test reads used for the trace above. GitHub changed filenames are reconciled with this local diff before publication. No local compile, Maven verify, unit/IT/E2E, Docker, service or UI execution occurred, as required for this review task. No build/deployment artifact was reused; engine-ui was not built.

2.3 Documentation Updates

Both Iceberg guides and both incompatible-change guides agree with the option, default, ordering and downgrade limits. The snapshot-count/per-partition-file and latency guidance addresses the earlier feedback. Only the internal recovery-helper documentation recommendation remains.

3. Architectural Soundness

3.1 Elegance of the Solution

Precise fix.

3.2 Maintainability

Best change-location assessment: The writer owns key retention and the connector committer owns Iceberg sequence ordering, so these are the correct layers. Modifying shared engine checkpoints or dropping equality deletes would expand scope and lose correctness. Keep the optional threshold and deterministic ordered result plan.

3.3 Extensibility

Keep existing connector identities and contracts. Reuse the current boundary and regression fixtures; no shared-framework redesign or unrelated dependency is needed for the behavior reviewed here.

3.4 Historical-Version Compatibility

The new optional positive long leaves the default row-count rotation disabled and does not rename options or change commit-info state classes. Rotated upsert/CDC and schema-evolution checkpoints may become multiple visible snapshots instead of one. Both incompatible-change guides explain prefix visibility, group metadata, older ungrouped checkpoints and the prohibition on downgrading during a partially committed grouped checkpoint. No configuration conversion is needed; restore such checkpoints with this or a later version. This review does not claim an executed upgrade/recovery validation.

4. Issue Summary

Number	Issue	Location	Severity
1	Document the restore-group detection contract	seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergFilesCommitter.java:111	Medium, non-blocking

Completeness check: full diff and affected entry/call chain, failure/resource paths, compatibility, tests and stability, bilingual docs, configuration/registration and current CI/merge gates were inspected before enumerating these findings. No additional finding is reserved for a later round on this version.

Community review reconciliation: All non-Daniel discussion and available reviews/inline comments were read; no independent reviewer technical finding or maintainer APPROVED/CHANGES_REQUESTED conclusion was present. The author's E2E, empty-group, downgrade and scan-shortcut replies are incorporated in this consolidated review. Previous Daniel code-side blockers are cleared; the small documentation recommendation above is non-blocking.

5. Merge Recommendation

Conclusion: Ready to merge

1. Blockers — must be fixed: No code-side blocker found. Required current-head CI and effective maintainer review/merge gates must be satisfied.
2. Recommended fixes — non-blocking: Issue 1: Document the restore-group detection contract

Overall assessment: The writer owns key retention and the connector committer owns Iceberg sequence ordering, so these are the correct layers. Modifying shared engine checkpoints or dropping equality deletes would expand scope and lose correctness. Keep the optional threshold and deterministic ordered result plan. Current head 50e6f5355ff71b39b0f0411ec1fee8bdb6ee27bc; base dev at 886a3ac6e59aad5e46755761cf9fb8ac2a6d3df9. Checks: Signal pull request update=SUCCESS; Build=SUCCESS. Check references: Signal pull request update Build. Compare: diverged, ahead 3, behind 5; merge state blocked / BLOCKED. Behind/diverged status alone is synchronization metadata and does not justify a sync-and-rerun instruction. Daniel has READ permission: this COMMENT is a technical conclusion, not an effective APPROVED event or dismissal of historical requested changes. A write-capable maintainer owns final approval/dismissal where required. The current-head check links above are the CI snapshot, not a locally executed test. Any historical Daniel requested-changes display is distinct from this source-clear COMMENT; a write-capable maintainer owns effective approval/dismissal. Behind status alone does not require another sync comment. Thanks again; happy to help with the focused next revision.

[DanielLeens]

Hi @zhengxiang378928908-code, thanks for resolving the conflict with dev so quickly. I re-checked the current head e755b1a563c0c0ccab373d39e9a1e08381b8d2f3.

What changed since my last review (50e6f535)

The three commits were rebased onto the current dev (3244c950). I compared the PR's own diff before (886a3ac6...50e6f535) and after (3244c950...e755b1a5) the rebase. The 16 files and all added and removed lines are identical. The only textual difference is context in docs/zh/connectors/sink/Iceberg.md, where dev (#13089) now writes the "required" column as Chinese "no" instead of English no. The new flush_row_size row already uses the same word, so the zh table stays consistent. The two dev Iceberg commits picked up by the rebase (#13067 test polling in IcebergSinkCDCIT, and #13089 docs) do not touch any file of this PR.

What Problem Does This PR Solve?

Unchanged: a long checkpoint keeps the Iceberg delta writer's remembered keys alive, so large upsert or CDC loads can exhaust heap. The optional flush_row_size rotates the task writer early. Completed writers are committed as ordered snapshot groups whenever a later round carries equality deletes, so those deletes still win over earlier rows, and a restore skips groups that are already durable.

1. Code Change Review

1.1 Core logic (re-checked at this head)

IcebergSinkWriter.write -> IcebergRecordWriter.write (IcebergRecordWriter.java:96-105, rotate at limit)
checkpoint -> prepareCommit -> complete() (ordered WriteResult list, RecordWriter.java:35-40)
 -> IcebergAggregatedCommitter.commitFiles (:89) -> planCommits (:124)
 -> 1 group: doCommit (unchanged single snapshot) | n groups: IcebergFilesCommitter.doCommitInOrder (:91)
restore -> restoreCommit -> commitFiles(restoring=true) -> committedGroups (:111) -> commit missing suffix

I re-read planCommits, committedGroups and the group summary properties (IcebergFilesCommitter.java:61-64, :266-267, :281-282). They are the same code I cleared before. When the option is unset the default path is unchanged, and append-only checkpoints still produce one snapshot.

1.2 Compatibility, 1.3 Side effects, 1.4 Error handling

These are unchanged from my previous review. The change is partially incompatible, which is documented in both incompatible-changes.md files: rotated upsert/CDC or schema-evolution checkpoints may become several visible snapshots, and you must not downgrade while a grouped checkpoint is partially committed. There is no change to options or the state format. A non-positive flush_row_size fails fast with a contextual OptionValidationException (IcebergSinkConfig.java:88-92).

Issue 1: Document the restore-group detection contract (still open from my previous review)

• Location: seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergFilesCommitter.java:111
• Problem: committedGroups is the correctness boundary for partial-checkpoint recovery, but it still has no method-level comment. It covers three modes: file-based fallback for a non-positive checkpoint id, branch-ancestry scan for a positive id, and "snapshot with checkpoint id but no group index = whole checkpoint already committed" for older ungrouped commits.
• Risk: A later maintainer could change the legacy interpretation or the early-return shortcut without knowing why it exists.
• Best improvement: A short Javadoc listing those three modes, the selected-branch scope and the assumption that group planning is deterministic from persisted results. No behaviour change.
• Severity: Medium, non-blocking.
• Raised by another reviewer: No.

2. Code Quality

• 2.1 Standards: apart from Issue 1, the new fields and methods carry useful comments.
• 2.2 Tests: IcebergWriterRotationTest, IcebergSinkConfigTest and the extended IcebergSinkIT, with the upsert and CDC flush_row_size configs, are unchanged. Stability: Stable. They use deterministic in-process fixtures and the existing IT polling.
• 2.3 Docs: the English and Chinese Iceberg pages and migration notes agree with the option name, default (unset), validation and ordering limits.

3. Architecture

This remains a precise fix. Key retention is solved in the writer and sequence ordering in the connector committer, without touching shared engine checkpoint code.

4. Issue Summary

#	Issue	Location	Severity
1	Missing Javadoc on the restore-group detection contract	IcebergFilesCommitter.java:111	Medium, non-blocking

5. Merge Recommendation

Conclusion: Ready to merge

• Blockers (non-CI): none.
• CI: the fork Build run 41568751302 for this exact head succeeded: all four unit-test legs, iceberg-connector-it (11/17) and updated-modules-integration-test-part-1/2 (11/17) passed, with the remaining updated-module parts skipped. The apache-side check rollup has no result attached yet, so a maintainer may need to approve the PR signal workflow so that the green Build is linked to the PR.
• Merge state: mergeable, no conflict; 3 commits behind dev, which is not a reason to sync again. An effective write-maintainer approval is still required.
• Non-blocking: Issue 1. A small follow-up commit with the Javadoc would be welcome, but it does not need to hold the merge.

Thank you for the careful work on the ordering and recovery semantics. This is a solid fix for a real production memory problem.