dl/coordinator: bound pending-file memory under commit backlogs - #30958
Merged
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR hardens the Iceberg topics coordinator against OOM scenarios caused by large backlogs of uncommitted pending files, by (1) committing in bounded chunks and (2) shedding load when the aggregate pending-file set grows too large. It also updates translators to treat coordinator backpressure as a deschedule/backoff signal and adds tests for the new bounded-copy/chunked-commit behavior.
Changes:
- Add
topic_state::copy_bounded()to bound the in-memory pending set per commit while preserving coordinator-offset (watermark) ordering semantics. - Update
iceberg_file_committer/coordinator loop to commit pending files in bounded chunks across multiple passes (no sleep between passes when more work remains). - Add coordinator backpressure (
too_many_pending_files) and translator behavior/tests to back off instead of retrying under load shedding; add new tunables for per-commit and aggregate pending-file limits.
Reviewed changes
Copilot reviewed 19 out of 19 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/v/datalake/translation/tests/partition_translator_tests.cc | Adds a translator test ensuring backpressure causes polling+backoff and resumes once load shedding stops. |
| src/v/datalake/translation/partition_translator.cc | Treats too_many_pending_files as a backoff/deschedule signal with rate-limited logging. |
| src/v/datalake/coordinator/types.h | Introduces errc::too_many_pending_files and formatting support. |
| src/v/datalake/coordinator/tests/state_update_test.cc | Adds unit tests validating copy_bounded() ordering/batch semantics. |
| src/v/datalake/coordinator/tests/state_test_utils.h | Updates test committer to new commit_result return type. |
| src/v/datalake/coordinator/tests/iceberg_file_committer_test.cc | Updates tests for chunked commits and commit_result; adds a cross-partition chunking test. |
| src/v/datalake/coordinator/tests/coordinator_test.cc | Adds backpressure and “drain without sleeping” tests; adds a chunked committer test double. |
| src/v/datalake/coordinator/state.h | Declares topic_state::copy_bounded(). |
| src/v/datalake/coordinator/state.cc | Implements copy_bounded() using a merge by added_pending_at to preserve watermark correctness. |
| src/v/datalake/coordinator/iceberg_file_committer.h | Adds a binding for max_files_per_commit and updates interface return type. |
| src/v/datalake/coordinator/iceberg_file_committer.cc | Uses copy_bounded() and returns commit_result{updates, has_more} to drive multi-pass draining. |
| src/v/datalake/coordinator/frontend.cc | Maps coordinator backpressure error to RPC error code. |
| src/v/datalake/coordinator/file_committer.h | Introduces commit_result (updates + has_more) to support chunked draining. |
| src/v/datalake/coordinator/coordinator.h | Wires in max_pending_files config and tracks leader-local backpressure cache state. |
| src/v/datalake/coordinator/coordinator.cc | Implements aggregate backpressure checking + immediate multi-pass commit draining when has_more is set. |
| src/v/datalake/coordinator/coordinator_manager.cc | Wires new config bindings into coordinator + committer construction. |
| src/v/datalake/coordinator/BUILD | Adds chunked_vector dependency needed by the new bounded-copy implementation. |
| src/v/config/configuration.h | Adds two new tunables for per-commit chunk size and coordinator pending-file cap. |
| src/v/config/configuration.cc | Defines defaults/docs/bounds for the new Iceberg tunables. |
Comment on lines
+578
to
+582
| if (has_too_many_pending_files()) { | ||
| vlog( | ||
| datalake_log.debug, | ||
| "Rejecting request to add files for {}: too many pending files", | ||
| tp); |
Comment on lines
+542
to
+546
| if ( | ||
| backpressured_as_of_.has_value() | ||
| && now - *backpressured_as_of_ < commit_interval_()) { | ||
| return true; | ||
| } |
Comment on lines
+191
to
+193
| // Threshold of total pending files across this coordinator's topics above | ||
| // which it sheds load. See should_reject_for_backpressure(). | ||
| config::binding<size_t> max_pending_files_; |
Comment on lines
+206
to
+210
| // Backpressure state (leader-local): when set, we are shedding load and | ||
| // this is when we last confirmed it; we keep rejecting without recomputing | ||
| // until a recheck interval past it. Recomputed by a new leader. See | ||
| // should_reject_for_backpressure(). | ||
| std::optional<ss::lowres_clock::time_point> backpressured_as_of_; |
Comment on lines
72
to
75
| case errc::too_many_pending_files: | ||
| return fmt::format_to(out, "errc::too_many_pending_files"); | ||
| } | ||
| } |
andrwng
force-pushed
the
iceberg-commit-chunk
branch
3 times, most recently
from
June 30, 2026 02:22
89d851b to
1dbd886
Compare
Collaborator
CI test resultstest results on build#86484
test results on build#86631
test results on build#86768
test results on build#86776
|
nvartolomei
reviewed
Jun 30, 2026
nvartolomei
left a comment
nvartolomei
left a comment
Contributor
There was a problem hiding this comment.
overall lgtm;
-
please add a ducktape test where max is set to 2 and add a topic with something like 10 partitions and let's make sure we don't drop files and we make progress in reasonable time too
-
add a metric for coordinator rejections; would be useful to monitor it across fleet i reckon
Contributor
|
On the ducktape test: Previous suggestion is flawed so ignore it. I want a backlog of multiple files spread across pending entries which we then commit slowly to the catalog. |
Introduce a variant of topic_state::copy() that bounds the copy to a given number of pending files, so a large backlog can later be committed in chunks rather than materialized all at once. The chunk includes all files at or below any accepted offset: if an entry at offset O is included, every entry (in any partition) at offset <= O is too. This ensure that any updates made by a given coordinator batch is committed atomically. No caller yet; wired into the committer next.
When a topic accumulates a large backlog of pending files (e.g. if the catalog has been rejecting our commit requests), committing them all at once forces the coordinator to materialize the whole set at once (the per-commit state copy and the accumulated Iceberg file list), which can OOM the broker. This commit bounds each commit via copy_bounded(), draining the backlog in chunks over multiple passes. To ensure the backlog drains promptly rather than one chunk per commit interval, the committer reports via commit_result::has_more whether the bounded copy left files behind, and the coordinator loop immediately retries a commit (skipping its inter-pass sleep) while a topic still has a chunked backlog.
andrwng
force-pushed
the
iceberg-commit-chunk
branch
3 times, most recently
from
July 2, 2026 18:37
d8ed497 to
fa058a8
Compare
nvartolomei
previously approved these changes
Jul 3, 2026
| err_msg="pending file backlog never reached the backpressure threshold", | ||
| ) | ||
|
|
||
| # Relax the limits so the coordinator drains the backlog promptly. |
Contributor
There was a problem hiding this comment.
If we relax limits the test doesn't prove that forward progress is being made.
One other thing test is missing is asserting that translators do respect backpressure signal.
Contributor
Author
There was a problem hiding this comment.
Thanks, made some more tweaks to the test to assert that we still make progress while backpressured and that we don't translate files during backpressure
Adds a new field to the coordinator fetch offsets response, allowing it to signal to translators that the coordinator is under load. The offsets are still returned and used to report lag.
Caps the total pending file count across all topics owned by a coordinator, configurable via iceberg_coordinator_max_pending_files. Once it reaches the threshold the coordinator rejects add-files and fetch-offset requests with too_many_pending_files, which the translator treats as a backoff signal until the committer drains the backlog below the threshold. Counting the pending files walks the whole pending set, so once we trip the threshold we hold the rejecting state for a recheck interval rather than recomputing on every request. The test injects a small threshold via the property binding and asserts add-files and fetch-offset are rejected once the backlog crosses it.
Add a coordinator_probe exposing per-operation counters for requests shed due to too many pending files, split into add-files and fetch-offset rejections. This is the coordinator's first metric and makes load shedding observable to operators.
Spreads a topic across multiple partitions and commits slowly (one file per commit) so pending files pile up faster than they drain, tripping the coordinator's pending-file threshold. Asserts the backpressure metric fires, then relaxes the limits and verifies every record still lands in the table exactly once.
andrwng
force-pushed
the
iceberg-commit-chunk
branch
from
July 6, 2026 18:27
fa058a8 to
08821cc
Compare
nvartolomei
reviewed
Jul 7, 2026
| return before == after | ||
|
|
||
| wait_until( | ||
| quiesced, |
Contributor
There was a problem hiding this comment.
I don't understand this assertion. Why do we expect pending_file_count and number of files created to be exactly the same in a ~5 second window? I'd expect the metrics to vary slightly down and high as both components make slow progress.
hold steady over a window shorter than the commit interval
We don't respect (sleep) the commit interval if there is more work to do though, right?
Contributor
Author
There was a problem hiding this comment.
Ah, you're right, I didn't take into account that we aren't waiting the full commit message. I think what this test was actually showing was that we finish translating and committing still even with the backpressure..
Contributor
Author
There was a problem hiding this comment.
Actually I'm going to merge this and tweak the test in a followup, given it's been sitting without pro code changes for a couple days
nvartolomei
approved these changes
Jul 7, 2026
nvartolomei
left a comment
nvartolomei
left a comment
Contributor
There was a problem hiding this comment.
Generally lgtm but still confused about the test.
Good to merge if I understand incorrectly the test https://github.com/redpanda-data/redpanda/pull/30958/changes#r3536399532
Collaborator
|
/backport v26.1.x |
Collaborator
|
/backport v25.3.x |
Collaborator
|
/backport v25.2.x |
This was referenced Jul 7, 2026
Collaborator
|
Failed to create a backport PR to v25.2.x branch. I tried: |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
When an Iceberg topic accumulates a large backlog of uncommitted data files, e.g. the catalog has been rejecting the coordinator's commits for a while, the coordinator can build up an unbounded set of pending files in its replicated state. Reconciling that backlog forces it to materialize the whole set at once (the per-commit state copy plus the accumulated Iceberg file list), which can OOM the broker.
This series bounds that memory along two independent axes:
High level changes:
The approach taken here is relatively simple, opting to go with tunable count-based backpressure, rather than more rigorous memory-based backpressure. This is mostly because the implementation was easier, though the mechanisms for backpressure should be reusable if we decide to change this in the future.
Backports Required
Release Notes
Improvements