Repository navigation
Fix buffer size include in flight messages in backpressure calculation logic - #836
Merged
Roman Kolesnev (rkolesnev) merged 2 commits intoOct 21, 2024
Merged
Conversation
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
airlock-confluentinc
Bot
force-pushed
the
fix/buffer-size-include-in-flight
branch
from
October 18, 2024 17:18
e27d71e to
a1414b7
Compare
Edward Vaisman (eddyv)
approved these changes
Oct 18, 2024
Roman Kolesnev (rkolesnev)
deleted the
fix/buffer-size-include-in-flight
branch
October 21, 2024 09:47
Merged
4 of 5 tasks
Antony Stubbs (astubbs)
added a commit
to astubbs/parallel-consumer
that referenced
this pull request
Jul 31, 2026
…isfiable shards-drained condition Characterization-first: arrival-syncing the original awaitingSelection==0 await made it time out deterministically on a quiet box (RED, 10.0s at the exact predicted line), confirming the diagnosis - the static messageBufferSize load factor caps take-out at 150 of 200 records, so the condition only ever passed vacuously before partition assignment. Rewritten as brokerPollPausedWhenBlockedInFlightFillsBuffer: awaits the real steady state (150 out / 50 awaiting, from named constants - pinning the take-cap as a regression guard), then asserts the pause, which non-vacuously verifies upstream confluentinc#836's in-flight-counts-toward-backpressure behaviour for the first time. 30s bounds (conditions stronger, only wall-clock allowance grew). Solutions doc records the new 'vacuous await condition' flake class; changelog + README regenerated; plan units ticked; ledger entry DIAGNOSED+FIXED. Validation: RED reproducer deterministic; rewritten class green 3/3; full integration suite 110/110 (forkCount=4) + unit suite green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Antony Stubbs (astubbs)
added a commit
to astubbs/parallel-consumer
that referenced
this pull request
Aug 4, 2026
BrokerPollerBackpressureTest.brokerPollPausedWithEmptyShardsButHighInFlight failed on the highcpu lane (run 30603617471) while the gating hosted run on the same commit was green. The ledger said "diagnose, don't dismiss". It was not contention: the test asserted a state the code cannot reach, and every green run had been passing without testing anything. The first await waited for getNumberOfWorkQueuedInShardsAwaitingSelection() == 0 with all ten workers latch-blocked. messageBufferSize(150) with maxConcurrency(10) makes PCModule.initDynamicLoadFactor pin a STATIC load factor of 15, so the control loop takes at most 10x15 = 150 records out of the shards. The test produces 200, nothing ever completes while the workers are blocked, and the count floors at 50. Zero is unreachable - at any timeout. It passed anyway because the await starts before partition assignment (~1-2s), when the shard map is empty and the condition is vacuously true. The same-sha hosted run proves it rather than merely suggesting it: the whole 2-test class finished in 5.5s including two deliberate 1s sleeps, and draining 200 records through a 150-record cap cannot happen at all, let alone in that budget. Under load the setup ran long, the await opened after the records had landed, and the failure became deterministic. The resemblance to the tight-absolute-timeout contention pattern already recorded in docs/solutions/test-flakiness/ was therefore a trap: same symptom, different mechanism, and "loosen the bound" would have restored the green without restoring the coverage. Not a main-code bug either - the 50-record floor is the static buffer cap working as designed, and the sibling test passed in the same failing run. The diagnosis was proven before anything was rewritten. Keeping the original await but preceding it with an arrival-sync makes it time out on an idle box, deterministically, first try - load had only been inverting a race. That reproducer was scaffolding and is not committed. Rewritten as brokerPollPausedWhenBlockedInFlightFillsBuffer, awaiting the real steady state: 150 out for processing AND 50 awaiting selection, asserted together so no partial state satisfies it, both derived from named constants. That pins the take-cap as a regression guard, and it gives upstream confluentinc#836's behaviour - blocked in-flight records counting toward backpressure - its first non-vacuous coverage here, since the 50 records left in the shards are far below the pause threshold and only the 150 counted in-flight ones can trip the pause. The scenario the old name described is unreachable by construction: the take-cap and the pause threshold are the same number. Await bounds went from 10s to 30s. The conditions got strictly stronger; only the wall-clock allowance grew, which costs nothing when green and absorbs a busy box. The lesson generalises, so it is recorded as its own flake class in docs/solutions/test-flakiness/vacuous-await-condition-brokerpoller-backpressure-2026-07-31.md: an await for "X == 0", where X starts at 0 and only later becomes non-zero, can pass before the system has reached the state under test. The working plan document was collapsed into that write-up rather than kept beside it - it was created and completed inside this PR, so its forward-looking half was inert and its diagnosis narrative duplicated the solutions doc line for line. The inflight ledger entry shrank to a pointer, per its own scope rule. The test is inherited from upstream (confluentinc#682 configurable buffer size, confluentinc#836 in-flight backpressure), so the latent race shipped with it and is still there. Also folds in the duplication this kept surfacing. BrokerPollerBackpressure- Test, RetriesTest and DrainCloseTest each opened with the same 18 lines - consumer/pcOpts/pc fields, then setupTopic() -> createNewConsumer(NEW_GROUP) -> options builder -> new ParallelEoSStreamProcessor -> subscribe - varying only in the options the test cares about. BrokerIntegrationTest.startPcOnNewTopic(options) now does it once, taking the builder pre-seeded with the consumer, so each setUp keeps just its distinctive line. The javadoc is written to be found: usage snippet first, then pointers to the sibling helpers (produceMessages, getTopic, getKcu, setupTopic, ensureTopic), so the next test author reuses the toolkit instead of re-deriving it and getting the setup subtly wrong. It returns the processor rather than assigning inherited fields on purpose: five other subclasses already declare their own consumer/pc fields, which base-class fields of the same name would silently shadow - the exact class of trap the refactor exists to remove. Additive to the base, no lifecycle change, so the other 18 integration tests are untouched. PMD CPD: clones 33 -> 29, duplicated lines 1001 -> 874. Verified: RED reproducer deterministic on an idle box; rewritten test green across repeated runs; the three refactored classes green; full integration suite 110/110 (forkCount=4) plus the unit suites, re-run after merging master so the result includes #100's broker-poll / rebalance-commit change. --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
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
2 participants
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 determination is done if consumer should be paused - the messages in processing queues / worker thread pool queue and threads should be taken into account along with messages in the shards.
Note that pausing can happen sooner now in certain situations - as inflight / queued in thread pool messages were not counted for buffer size previously but now they are counted in.
Checklist