Skip to content

refactor: Queue unification - #219

Merged
Antony Stubbs (astubbs) merged 37 commits into
confluentinc:masterfrom
astubbs:features/single-queue
Apr 22, 2022
Merged

refactor: Queue unification#219
Antony Stubbs (astubbs) merged 37 commits into
confluentinc:masterfrom
astubbs:features/single-queue

Conversation

@astubbs

@astubbs Antony Stubbs (astubbs) commented Mar 8, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Massive simplification, much improved semantics around poller notifying of work incoming.

Also address:

@astubbs
Antony Stubbs (astubbs) changed the base branch from master to features/batching March 8, 2022 23:30
@astubbs
Antony Stubbs (astubbs) marked this pull request as draft March 8, 2022 23:31
@astubbs
Antony Stubbs (astubbs) force-pushed the features/batching branch 6 times, most recently from b7b631c to 7e26075 Compare March 14, 2022 15:26
@astubbs Antony Stubbs (astubbs) mentioned this pull request Mar 17, 2022
Closed
Antony Stubbs (astubbs) referenced this pull request in astubbs/parallel-consumer Mar 22, 2022
@astubbs
fixes #236: Shard starvation under some circumstances (#237)
8c40436 
The simplest scenario of this issue, is:

using PARTITION ordering
max concurrency set to 2

- consumer is able to poll 10 messages, the first 5 for partition 0, the next 5 for partition 1
- when PC requests work from work manager, it first requests 2 work units
- work manager then ingests 2 units from it's queue, but to shard partition 0
- then shard iteration begins, partition order restriction is set, so only takes a single entry from the ONLY partition shard of zero
- so result is only a single record is returned, even through there are 5 in the queue for the 2nd partition that aren't "realised"
- on next request to WM, 1 more unit is requested to hit concurrency target - and this is also ingested from the 3 remaining for partition 0, and the rest for partition 1 are still undiscovered, and so on and so on - extrapolate out...
- solution:
- ingest the entire work queue into shards every round, which is effectively also what #219

Also fixed by upcoming RR:

refactor: Queue unification #219
@astubbs
Antony Stubbs (astubbs) changed the base branch from features/batching to master April 6, 2022 08:54
@astubbs Antony Stubbs (astubbs) mentioned this pull request Apr 21, 2022
Closed
astubbs
Antony Stubbs (astubbs) commented Apr 21, 2022
View reviewed changes

@astubbs Antony Stubbs (astubbs) left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

clean up all the dead code

Comment thread ...src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java Outdated
Comment thread ...src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java Outdated
Comment thread ...src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java Outdated
Comment thread ...lel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java Outdated
Comment thread ...consumer-core/src/main/java/io/confluent/parallelconsumer/internal/CountingCRLinkedList.java Outdated
Comment thread parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java Outdated
Comment thread ...ntegration/java/io/confluent/parallelconsumer/integrationTests/LargeVolumeInMemoryTests.java Outdated
Comment thread ...gration/java/io/confluent/parallelconsumer/integrationTests/MultiInstanceHighVolumeTest.java Outdated
Comment thread ...core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureTest.java Outdated
Comment thread ...l-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingTests.java Outdated
astubbs
Antony Stubbs (astubbs) commented Apr 22, 2022
View reviewed changes

@astubbs Antony Stubbs (astubbs) left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

...

@astubbs
Antony Stubbs (astubbs) force-pushed the features/single-queue branch 2 times, most recently from 3b51ffe to aa5c0e1 Compare April 22, 2022 11:08
@astubbs
Antony Stubbs (astubbs) marked this pull request as ready for review April 22, 2022 11:26
@astubbs
Antony Stubbs (astubbs) merged commit 9a96686 into confluentinc:master Apr 22, 2022
@astubbs
Antony Stubbs (astubbs) deleted the features/single-queue branch April 22, 2022 11:29
@astubbs Antony Stubbs (astubbs) mentioned this pull request Apr 22, 2022
Open
40 of 64 tasks
@astubbs Antony Stubbs (astubbs) mentioned this pull request May 16, 2022
Closed
Antony Stubbs (astubbs) added a commit to astubbs/parallel-consumer that referenced this pull request Aug 18, 2026
@astubbs
docs(inflight) confluentinc#909: record the epoch-check asymmetry #31 …
29953f4 
…does NOT fix, and the untested retry queue

Two open items that were being carried in conversation rather than anywhere a
future session would find them.

bug-epoch-check-is-per-batch-on-add-per-container-on-take.md states the confluentinc#909
defect as the asymmetry rather than the symptom: the add-side guard is evaluated
once per BATCH against a PartitionState the caller already holds, while the take and
complete checks are per CONTAINER against live state. A rebalance landing inside the
registration loop therefore puts old-epoch containers into live shards.

The reason it matters more than a missing re-check: 9a96686 (Queue unification,
confluentinc#219) removed four synchronized (partitionStates) blocks on the stated
grounds that "Synchronisation of PartitionState map no longer needed due to stronger
epoch tracking ... a proactive epoch check before registering the work". That check
was meant to be strong enough to REPLACE the lock. Per-batch and possibly-stale, it
is not, and nothing since has re-established what the lock gave. The note says
plainly that #31 fixes the consequence at the shard and does not repair this,
and points at the filed repair and the two measured reasons it is not a one-liner.

test-retry-queue-behaviour-untested.md records that RetryQueue has three tests, all
covering ordering, and nothing covering the two-map uniqueness/ordering invariant,
last-write-wins re-add with a changed retry-due time, revoke behaviour, or shard/queue
consistency after a stale removal - the last being where the already-filed orphan
lives. Worth a note because a defect there surfaces as records retried late or never,
not as an exception, and the counters it feeds gate the poller throttle.
Antony Stubbs (astubbs) added a commit to astubbs/parallel-consumer that referenced this pull request Aug 18, 2026
@astubbs
docs(core) confluentinc#909: document the three staleness checkpoints…
678f47b 
… where an agent editing the code will see them

Two sessions in a row - both mine - read one checkpoint in isolation and concluded
the epoch scheme was broken. It is not. Each checkpoint is deliberately partial and
they are only sound together, and nothing in the code said so.

PartitionState#epochIsStale now carries the scheme: register (proactive, once per
poll batch), take (couldBeTakenAsWork, per container against LIVE state, and the
authoritative one), complete (handleFutureResult, per container against live state).
couldBeTakenAsWork and handleFutureResult cross-reference it rather than restating
it, per the repo's never-state-a-fact-twice rule.

It states the two things that were repeatedly re-derived as bugs. That checking once
per batch is CORRECT, not a shortcut: RecordsAndEpoch carries one
epochOfPartitionAtPoll for its whole record list, so a per-record check returns the
identical answer. And that the checkpoint is knowingly racy - the caller looks the
state up live and calls straight in, but a rebalance on the broker-poll thread can
land between that lookup and the inserts, so the guard can pass wrongly either
because the epoch bumped after a correct check or because this state object itself
went stale. Checkpoint 2 is what makes that safe.

It also says what NOT to do: re-checking per record or consulting the live epoch here
closes nothing, since both remain check-then-act against a concurrent rebalance, and
the lock that would close it was removed deliberately in 9a96686
(confluentinc#219) on the grounds that epoch tracking replaced it - reasoning that
holds for the scheme as a whole, not for this checkpoint alone.

Deletes docs/inflight/bug-add-side-epoch-check-can-compare-a-stale-state-against-itself.md.
That note described a non-bug: its two "faults" share one window, backstopped by
checkpoint 2. What was genuinely broken was the shard's collision handling, which is
what this PR fixes.

Chose javadoc over an architecture doc deliberately: only agents read this, and an
agent editing the registration path sees javadoc without knowing to look for a doc.
Antony Stubbs (astubbs) added a commit to astubbs/parallel-consumer that referenced this pull request Aug 19, 2026
@astubbs
docs(core) confluentinc#909: document the three staleness checkpoints…
3094281 
… where an agent editing the code will see them

Two review passes in a row read one checkpoint in isolation and concluded the epoch
scheme was broken. It is not. Each checkpoint is deliberately partial and they are
only sound together, and nothing in the code said so.

PartitionState#epochIsStale now carries the scheme: register (proactive, once per poll
batch), take (couldBeTakenAsWork, per container against LIVE state, and the
authoritative one), complete (handleFutureResult, per container against live state).
couldBeTakenAsWork and handleFutureResult cross-reference it rather than restating it.

It states the two things that were repeatedly re-derived as bugs. That checking once
per batch is CORRECT rather than a shortcut: RecordsAndEpoch carries one
epochOfPartitionAtPoll for its whole record list, so a per-record check returns the
identical answer. And that the checkpoint is knowingly racy, with checkpoint 2 being
what makes that safe.

It also says what NOT to do: re-checking per record or consulting the live epoch here
closes nothing, since both remain check-then-act against a concurrent rebalance, and
the lock that would close it was removed deliberately in 9a96686 (confluentinc#219)
on the grounds that epoch tracking replaced it - reasoning that holds for the scheme as
a whole, not for this checkpoint alone.

Chose javadoc over an architecture doc deliberately: only agents read this, and an
agent editing the registration path sees javadoc without knowing to look for a doc.
@astubbs Antony Stubbs (astubbs) mentioned this pull request Aug 19, 2026
Merged
8 tasks done
Antony Stubbs (astubbs) added a commit to astubbs/parallel-consumer that referenced this pull request Aug 19, 2026
@astubbs
docs(core) confluentinc#909: document the three staleness checkpoints…
e103a86 
… where an agent editing the code will see them

Two review passes in a row read one checkpoint in isolation and concluded the epoch
scheme was broken. It is not. Each checkpoint is deliberately partial and they are
only sound together, and nothing in the code said so.

PartitionState#epochIsStale now carries the scheme: register (proactive, once per poll
batch), take (couldBeTakenAsWork, per container against LIVE state, and the
authoritative one), complete (handleFutureResult, per container against live state).
couldBeTakenAsWork and handleFutureResult cross-reference it rather than restating it.

It states the two things that were repeatedly re-derived as bugs. That checking once
per batch is CORRECT rather than a shortcut: RecordsAndEpoch carries one
epochOfPartitionAtPoll for its whole record list, so a per-record check returns the
identical answer. And that the checkpoint is knowingly racy, with checkpoint 2 being
what makes that safe.

It also says what NOT to do: re-checking per record or consulting the live epoch here
closes nothing, since both remain check-then-act against a concurrent rebalance, and
the lock that would close it was removed deliberately in 9a96686 (confluentinc#219)
on the grounds that epoch tracking replaced it - reasoning that holds for the scheme as
a whole, not for this checkpoint alone.

Chose javadoc over an architecture doc deliberately: only agents read this, and an
agent editing the registration path sees javadoc without knowing to look for a doc.
Antony Stubbs (astubbs) added a commit to astubbs/parallel-consumer that referenced this pull request Aug 19, 2026
@astubbs
docs(core) confluentinc#909: document the three staleness checkpoints…
b42e8be 
… where an agent editing the code will see them

Two review passes in a row read one checkpoint in isolation and concluded the epoch
scheme was broken. It is not. Each checkpoint is deliberately partial and they are
only sound together, and nothing in the code said so.

PartitionState#epochIsStale now carries the scheme: register (proactive, once per poll
batch), take (couldBeTakenAsWork, per container against LIVE state, and the
authoritative one), complete (handleFutureResult, per container against live state).
couldBeTakenAsWork and handleFutureResult cross-reference it rather than restating it.

It states the two things that were repeatedly re-derived as bugs. That checking once
per batch is CORRECT rather than a shortcut: RecordsAndEpoch carries one
epochOfPartitionAtPoll for its whole record list, so a per-record check returns the
identical answer. And that the checkpoint is knowingly racy, with checkpoint 2 being
what makes that safe.

It also says what NOT to do: re-checking per record or consulting the live epoch here
closes nothing, since both remain check-then-act against a concurrent rebalance, and
the lock that would close it was removed deliberately in 9a96686 (confluentinc#219)
on the grounds that epoch tracking replaced it - reasoning that holds for the scheme as
a whole, not for this checkpoint alone.

Chose javadoc over an architecture doc deliberately: only agents read this, and an
agent editing the registration path sees javadoc without knowing to look for a doc.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@astubbs