Repository navigation
Make PC message buffer size configurable - #682
Merged
Roman Kolesnev (rkolesnev) merged 6 commits intoJan 24, 2024
Merged
Conversation
Edward Vaisman (eddyv)
previously approved these changes
Jan 23, 2024
Roman Kolesnev (rkolesnev)
deleted the
feature/configurable-message-buffer
branch
January 24, 2024 15:39
Martyn Ye (sangreal)
pushed a commit
to sangreal/parallel-consumer
that referenced
this pull request
Feb 8, 2024
* Make PC message buffer size and load factor configurable
Merged
4 of 5 tasks
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>
6 tasks done
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.
Make PC message buffer size configurable.
Two configuration options added - specifying message buffer size (which will calculate loading factor and set loading factor to static value) and configurable initial and max loading factors.
ceiling ( messageBufferSize / maxConcurrency*batchSize)Readme updated to add consideration to tune fetch and buffer size in Partition ordering mode as buffer size can be small by default leading to uneven load per partition and starving processing threads.
Test added to illustrate starved threads scenario and tuning of config to prevent it.
Checklist