[dougqh]

What Does This Do

Adds datadog.common.queue.WorkQueue<T> in a new utils/queue-utils module: a bounded handoff between producers and a consumer whose defining property is that admission claims capacity before it invokes any producer, so an element that is going to be rejected is never constructed.

Admission, in the order you should reach for it:

• tryPut(element) — for something already built.
• tryPut(producer), tryPut(context, producer), tryPut(c1, c2, producer) — the queue claims a place, then asks the producer to build. The context parameters exist so a producer can stay a non-capturing, bound-once field instead of a lambda allocated per call; the ladder stops at two arguments deliberately.
• tryPutBatch(source, context, producer) and a RejectHandler overload — the queue walks a source collection and transforms as it goes, returning how many it admitted. A producer returning null declines that source element: its place goes back and nothing is counted, so intended - admitted is the caller's exact shortfall.
• tryPutBatch(elements) / tryPutBatch(T...) — return the elements that were refused.
• tryReserve() → Reservation<T> — the escape hatch for work that will not fold into a callback. A refusal is a reservation that reports granted() == false, never null.

Consumption is synchronous, in the caller's frame: process(consumer), process(limit, consumer), context-carrying variants, and processOrRetry / processOrHandle for callers that want to say what a consumer failure means (RetryStrategy, RetryQueue, RetryStrategy.maxRetries, ExceptionHandler). Plain process propagates; the queue never logs and takes no view on a failure it was not given one for.

Three factories over one bound: WorkQueues.createMpscQueue (jctools MPSC ring, the preferred path), createMpmcQueue (jctools MPMC ring, for multiple consumers) and createUnboundedMpmcQueue (a ConcurrentLinkedQueue, for call sites with no defensible capacity yet). All are bounded by the same permit counter, which also makes size() O(1). Function-object parameters are @Nonnull and not checked; elements are checked before any capacity is claimed.

Tests: WorkQueueContractTest runs the shared contract against every backing, plus MpscWorkQueueStressTest. Benchmarks: AdmissionBenchmark (admission forms, reservation escape analysis, the third-backing inlining cliff), ContendedAdmissionBenchmark / ContendedBatchAdmissionBenchmark (the capacity boundary under contention), and AdmissionAlternativesBenchmark (against the queues it replaces and the ring's own fill).

Motivation

Queue users in the tree hand-roll the same few things directly against jctools — a capacity check, a drop counter, a drain loop — and each one answers the same questions slightly differently. But the variation is not the interesting part. The interesting part is a failure that lives in the gap between two decisions that each look fine on their own.

The seam. A telemetry collector does two things: it reads a counter destructively, and it hands the result to a bounded queue that may refuse. The counter design is sound. The queue usage is sound. The bug is the ordering — and the ordering belongs to neither half, so nothing in a review of either half catches it. WafMetricCollector has it fourteen times:

long counter = wafInputTruncatedCounter.getAndSet(i, 0);   // state destroyed
if (counter > 0) {
  if (!rawMetricsQueue.offer(new WafInputTruncated(counter, i))) {
    return;                                                 // ...and now lost for good
  }
}

The counter is already zeroed when the offer fails, so the value is gone permanently — and this fires precisely when telemetry is backing up because the agent is unreachable, i.e. when the data is worth the most. Nothing about getAndSet or offer is wrong. Their sequence is.

What the callback does about it. Because admission claims a place before it invokes the producer, the destructive read moves inside the producer, and the producer only ever runs on a place that has already been granted. You cannot destroy state you have not yet been granted somewhere to put it. The two decisions can no longer be made separately, because the API only offers them joined:

if (!queue.tryPut(i, this, (idx, self) ->
        new WafInputTruncated(self.wafInputTruncatedCounter.getAndSet(idx, 0), idx))) {
  break;   // no place was claimed, so the counter was never read and nothing was lost
}

That is the design's central claim, and it is a correctness argument rather than a performance one. It is why the contextual and batch producer forms are the shapes the API pushes you toward, and why a plain tryPut(element) — while supported — gives up the guarantee: an element built before the call is state already committed.

Two honest limits. It is available, not enforced — capturing the read outside the lambda puts the ordering back the way it was. What changes is which version is the natural one to write, which is why the recipe belongs in the primitive's own documentation rather than being left for each adopter to rediscover from the composition. And the guarantee is about ordering, not about the drain as a whole: a caller that stops on the first refusal still leaves the rest of its counters for the next cycle. That is the correct outcome — they are untouched — but it is the batch form, which reports how many of the intended elements it admitted, that makes the shortfall visible rather than implicit.

The allocation story is the same property, seen from the other side. Building work you then throw away is the benign case of the same ordering. Client-side stats is the clearest example: it builds a SpanSnapshot per eligible span, with peer-tag and additional-tag arrays, and discards it when the inbox turns out to be full — again at the moment of most pressure. Reserve-before-build means a full queue costs a read and nothing else.

Everything else the permit counter buys follows from wanting one bound that means the same thing everywhere: an exact capacity over every backing, and an O(1) size(). Those are worth most under backpressure, which is exactly where a hand-rolled capacity check is least likely to have been thought through. Read that way, the counter's cost — about 13ns per admit-and-drain over the ring's own claim-before-build (AdmissionAlternativesBenchmark, JDK 17: 35.8 vs 22.9ns, no extra allocation), and ~5ns on refusal — is the price of the guarantee rather than an overhead to be removed — which is why the cheaper options that trade the guarantee away (per-backing admission, an approximate bound from batched thread-local permits) are not pursued here.

Possible applications

• Telemetry collectors that read-and-reset counters before offering (WafMetricCollector, CoreMetricCollector): claiming first means a delta is never destroyed for an element the queue then refuses.
• Call sites that ignore offer's return (RumInjectorMetrics): batch admission turns a per-element dropped return into one count.
• Hand-rolled capacity checks (CoreMetricCollector's remainingCapacity() == 0): the claim is the check, so there's no check-then-act race.
• Unbounded LinkedBlockingQueues (ProductChangeCollector, IntegrationsCollector): a bound becomes the default, and unbounded becomes an explicit choice.
• Client-side stats (Run client-side stats through WorkQueue (perf toolbox adoption) #12339): the first real adopter.

Additional Notes

There is no in-tree caller on this branch — the API is exercised only by its own tests and benchmark. #12339 stacks the client-side-stats adoption on top as a trial, so the surface can be reviewed against real use. Reviewing the two together is more informative than reviewing this one alone.

Known gaps, in case they are load-bearing for your read:

• Contention at the boundary is measured, and it is no longer the dominant cost. ContendedAdmissionBenchmark (8 threads, gc profiler, JDK 25, loaded machine — directional) first priced a refused admission at ~960ns against ~3.4ns for the same rejection on jctools' own producer-index CAS: two read-modify-writes on one shared line, taken by every thread at the capacity boundary. claimPlace() now reads the count before spending from it, so a full or closed queue refuses with a load. Refusal is ~7.9ns against ~2.9ns, putting the permit counter at roughly 5ns over a bound the MPSC ring already enforced — an accepted cost for what the API does, not an open question. The decrement stays authoritative, so the bound is unchanged.
• Reserve-before-build now wins on both axes. refusedProducer vs refusedBuildThenOffer is ~8ns / 0 B/op against ~422ns / 32 B/op. Before the read it was the awkward result — 0 B/op but slower in ns/op, because the counter cost more than the allocation it avoided. The two arms are not one variable (no counter in one, no allocation in the other); what reversed is the ordering.
• Trust the direction more than the ratio: 960ns is too expensive for two contended RMWs on a quiet machine, so some of that baseline was this machine's other work amplifying contention. A quiet run should show a smaller multiple against a smaller before. The ~7.9ns is tight (±0.35).
• The closed flag is gone, folded into the permit count as a large negative bias — so the check disappears from all seven admission sites rather than getting cheaper, and closed and capacity can no longer be read out of step. A producer can no longer see an open flag and then claim a place close() already revoked. The count is a long because createUnboundedMpmcQueue seeds it with Integer.MAX_VALUE, which leaves an int no headroom for the bias — close() there would have silently done nothing.
• AdmissionBenchmark is @Threads(1) and Scope.Thread: it prices the admission forms and the third-backing cliff uncontended, and leaves contention to ContendedAdmissionBenchmark. Its reservation arms answered their own question: building a fresh refusal measures 0 B/op where a shared refusal singleton costs 12, because the allocation-merged-with-a-static phi defeats escape analysis (JDK 17).
• Writing that benchmark turned up an API hazard worth a reviewer's eye: createMpscQueue's "Single Consumer" is a requirement, and a second consumer neither throws nor is rejected — the two spin in the ring's gap-wait, which presents as a hang. A JMH @Group with @GroupThreads(1) silently produced two consumers under -Pjmh.threads=8 and wedged a run for 28 minutes. The factory javadoc now states the consequence; nothing enforces it.
• Two behaviour fixes from a @codex review pass, both narrow: size() clamped to the capacity it reports against (it could exceed it while a claimant was backing out of an overspend), and MaxRetries(n) (now RetryStrategy.maxRetries(n)) performs n retries rather than n - 1. A third finding — retry transiently-empty reads in the MPMC backing — did not hold: jctools' poll spins for a pending publish and returns null only when the producer and consumer indices agree, so there is no false empty to ride out. Three of our own comments claimed otherwise and are corrected; discardAll's bounded re-read stays, for the reason that is actually true (size counts places claimed by producers that have not stored yet).
• There is no drop counter, on purpose. The queue owns no dropped(); counting and publishing a refusal stays the caller's, which is the division FlagEvaluationWriterImpl already chose by hand. It was removed rather than shipped because ten of its eleven increment sites were redundant with a return the caller already gets — tryPut's boolean, tryPutBatch's admitted count, Reservation.granted(), the RejectHandler — and nothing in the tree read it, so it was a safety net for a caller who ignores returns rather than new information. It can come back in its own PR if an adopter needs it. The one thing that went with it and should not have: an item a RetryStrategy abandons is now lost with no report to the caller. processOrRetry returns only whether there was an item, so RetryStrategy.onFailure's true/false — resubmitted versus gave up — is the only signal, and nothing reads it; with a library strategy like RetryStrategy.maxRetries the caller never learns. That is the single loss in this class with no synchronous channel back, and it wants either a changed processOrRetry return or the counter back. Both are follow-ups.
• A producer that returns null declines: its place goes back, and no RejectHandler is told, because nothing was lost. The boolean on the single-element tryPut forms still collapses a decline and a refusal into the same false, so a drain loop that skips zero-valued counters cannot tell "nothing to send" from "no room" without the batch form. The javadoc gap that went with it is fixed — all three producer overloads now say so at the point of use — but whether that return should distinguish the two is a design question worth a reviewer's opinion.
• Unexercised by any caller anywhere: tryPutBatch(Collection), tryPutBatch(T...), the RejectHandler overload, tryReserve / Reservation (a client-side stats use is planned), RetryStrategy / RetryQueue / maxRetries, processOrRetry / processOrHandle. That is a lot of surface per adopter and a reasonable thing to push back on.
• shutdown() is not atomic, and its javadoc now says so rather than claiming otherwise. It is close(); discardAll(); — producers, reservations, and retries that claimed capacity before the close can still store an element after the discard, and that element then sits in a queue nothing will drain. Ordering the flag first bounds the survivors to those already in flight rather than eliminating them. Making it genuinely atomic means re-reading closed after every producer returns and before its element is stored — four admission sites, a per-admission cost, and a behaviour change that deserves its own PR and its own measurement. Correcting the doc is this PR; the guard is not.
• The permit counter's bound is exact — the queue never holds more than capacity elements and open reservations — but who gets turned away is approximate at the boundary, where claimants racing can back out together and refuse an admission while the queue is a place or two short of full.
• From review: the MPMC backing's bounded store retry is gone. JCTools' strict offer refuses only at index-full, and a producer holding a permit can never see that, so the retry could never run; a stress run with the retry path made to throw (capacity 2, 8×8 threads) never reached it. A Reservation is confined to the reserving thread — completion is a plain flag, and the rule is now in its public contract rather than a private comment.

🤖 Generated with Claude Code

[dougqh]

Update: two backings added

Both are package-private and reachable only through Queues factories — mpscQueue(capacity), mpmcQueue(capacity), unboundedMpmcQueue(). Shared admission/lifecycle/retry logic is in BaseQueue. 28 tests, all passing.

Reserve-first is real, not emulated

Worth recording, because it decides whether the central guarantee is achievable by wrapping JCTools rather than forking it. I decompiled fill(Supplier, int) in 4.0.6:

• MpscArrayQueue (Unsafe, Java 8 path) — casProducerIndex at bytecode 141, Supplier.get() at 180.
• MpscVarHandleArrayQueue (Java 25+ path, what Queues selects) — casProducerIndex at 128, Supplier.get() at 167.

Both CAS-claim the slot first, and both return early without calling the supplier at all when there's no capacity. So fill(supplier, 1) == 1 is exact reserve-first admission, on both paths. The stress test asserts this directly: 8 threads × 20k producing admissions against a permanently full queue invoke the producer zero times.

A consequence the ticket didn't note

The claim is published (producer index advanced) before the element is stored. A consumer that reaches that slot waits for the element to appear. So reserve-first converts "producer allocates" into "producer allocates while holding a slot the consumer may be blocked on" — a slow producer now stalls the consumer, where before it only stalled itself. Fine for SpanSnapshot; not fine for anything that blocks or does I/O in produce(). I documented this on Queue, but it's a contract the use-case cards should be checked against — OkHttpSink (APMLP-1652) builds a Request including a full buffer copy inside what would become the producer, which is the largest candidate.

Two API defects found by the compiler

1. RetryStrategy<? super T> doesn't typecheck. A strategy over a supertype S would receive a RetryQueue<S>, and the queue can't satisfy that — it only accepts T. Changed process to take an invariant RetryStrategy<T>. The variance in the ticket's converged signature isn't merely too permissive, it's uninhabitable.
2. The overload ambiguity is real, not theoretical. process(consumer, null) doesn't compile — ambiguous between process(Consumer, RetryStrategy) and process(C, BiConsumer). Currently worked around with a cast at the internal call site, but every caller passing a literal null hits it too. This strengthens the case for renaming the context-taking forms.

Deviations from the ticket, deliberate

• The retry "lease" is not honoured. The ticket says a single-item retry "cannot fail on capacity" because the consumer still owns the slot. poll() releases the slot before the consumer runs, so there is no slot left to reuse — a retry is an ordinary re-admission and can be rejected if producers refilled the queue meanwhile (counted as a drop). Honouring the ticket would need a claim-based consumer, i.e. forking the backing rather than wrapping it, or reserving a hidden retry slot. Flagging rather than quietly weakening it.
• Batch admission is element-wise, not one atomic reservation. tryPutBatch / tryPut(Collection) / RetryQueue.retry(T...) can therefore partially admit. fill(supplier, n) would make this atomic for the MPSC backing; left out of the draft.
• shutdown() is not atomic — it sets closed, then discards. A producer already past the closed check can still land an element. Narrowing the window needs backing-level support.
• put(BatchProducer) is lossless by construction, which I think is the intended reading: an element is pulled only once a slot is claimed, so whatever doesn't fit is still held by the producer. It stops early rather than blocking, and the caller checks hasNext(). No drops are counted for stopping at capacity.

On LinkedQueue

It keeps the per-element node, so it doesn't deliver the allocation win — it exists so multi-consumer or as-yet-unbounded call sites can adopt the interface first and be re-backed later. Its size counter is what makes the bound enforceable and size() O(1), which is precisely what DependencyResolverQueue (APMLP-1654) wants in place of a hand-rolled cap plus an O(n) ConcurrentLinkedQueue.size() walk on every admission.

Still open from the original description

The static-routine bypass for hot call sites — and with it the interface-vs-final-class dispatch question — is still not addressed. With two backings live behind Queue<T>, admission call sites that see both are now genuinely bimorphic.

[dd-octo-sts]

🟢 Java Benchmark SLOs — All performance SLOs passed

Suite	Status
Startup	🟢 pass

SLO thresholds are defined here based on automatically generated metrics. A warning is raised when results are within 5% of the threshold.

PR vs. master results

Scenario	Candidate	master	Δ (95% CI of mean)
startup:insecure-bank:iast:Agent	13.94 s	13.88 s	[-0.2%; +1.1%] (no difference)
startup:insecure-bank:tracing:Agent	12.87 s	12.93 s	[-1.3%; +0.3%] (no difference)
startup:petclinic:appsec:Agent	17.20 s	17.11 s	[-0.3%; +1.3%] (no difference)
startup:petclinic:iast:Agent	16.94 s	17.09 s	[-1.7%; -0.0%] (maybe better)
startup:petclinic:profiling:Agent	16.56 s	16.68 s	[-1.8%; +0.4%] (no difference)
startup:petclinic:sca:Agent	17.11 s	16.98 s	[-0.1%; +1.6%] (no difference)
startup:petclinic:tracing:Agent	16.12 s	16.09 s	[-1.0%; +1.4%] (no difference)

Commit: 9fc6cb72 · CI Pipeline · Benchmarking Platform UI

---

Load and DaCapo benchmarks can be triggered manually in the GitLab pipeline. Results will appear in the Benchmarking Platform UI after completion.

[dougqh]

Renamed: Queue<T> → WorkQueue<T>, factories split out

Set apart from the raw JCTools factory, and the java.util.Queue collision (open question 1) is gone — no caller has to qualify an import now.

Queue            -> WorkQueue          (+ WorkQueues factory)
BaseQueue        -> BaseWorkQueue
MpscBoundedQueue -> MpscWorkQueue
LinkedQueue      -> LinkedWorkQueue

The two factory surfaces are now separate classes, because they answer different questions for the caller:

	returns	caller's job
WorkQueues.createMpscQueue(n)	WorkQueue<T>	hand work over; backing is hidden and re-backable
Queues.mpscArrayQueue(n)	MessagePassingQueue<T>	drive the raw queue yourself

Queues keeps only the raw factories and is otherwise untouched, so its eleven existing callers — which include OkHttpSink and ClientStatsAggregator, two of the ticket's own use cases — are unaffected until they migrate deliberately.

Usage now reads:

WorkQueue<SpanSnapshot> inbox = WorkQueues.createMpscQueue(1024);
inbox.tryPut(ctx, SNAPSHOT);
inbox.process(this::publish, new MaxRetries<>(3));

28 tests still green.

Open questions, updated

1. Queue collides with java.util.Queue — resolved by the rename.
2. Overload ambiguity still stands, and is still real rather than theoretical: process(consumer, null) does not compile. Worth deciding whether the context-taking forms get distinct names.
3. RetryQueue.retry(T...) unchecked warning — unchanged. (Note it kept its name: it is the retry capability, not a WorkQueue.)
4. Module placement — settled: utils/queue-utils, alongside Queues.
5. Static-routine bypass / dispatch — still open, and now the sharper question of the two, since two backings sit behind WorkQueue<T>.

[dougqh]

Dropped BatchProducer and put() — deferred to the SCA follow-on

Removed from this PR. The reasoning, so the follow-on doesn't relitigate it:

Why it goes now. Across all five use cases on APMLP-1642 — client-side stats, PendingTrace, OkHttpSink, health metrics, DependencyResolverQueue — every admission is a single element. Batch admission had no caller, so it was a public type, a method, and a lossless-admission contract that every future backing would have to honour, carried on spec.

Why it comes back with SCA. SCA's partition-on-failure (the #11977 Reachability case) is the genuine caller. Landing it there means its real access pattern picks the shape, instead of us guessing between pull, push, and reservation.

The name was wrong too, which is what started this. BatchProducer sat in a family with Producer.produce() and ContextualProducer.produce(ctx) but had hasNext()/next() — so it read as "produces batches" while actually being pulled one element at a time, advertising the opposite of the mechanism. It was also structurally java.util.Iterator<T> with no remove(), which has had a throwing default since Java 8.

Agreed shape for when it returns

Not a callback taking the WorkQueue itself. Handing out the full interface exposes close(), shutdown(), clear() and process() to arbitrary caller code, and it lets the filler stash the reference past the call. It also inverts loop ownership: the filler learns about capacity only from tryPut returning false, by which point the eager tryPut(T) form has already built the element — reintroducing build-then-drop inside the one method meant to prevent it.

Instead, a scoped admission-only capability in the manner of RetryQueue — obtainable only inside the call, dead on return:

interface Admission<T> {                 // no lifecycle, no consumption
  boolean tryPut(T element);
  <C> boolean tryPut(C context, ContextualProducer<? super C, ? extends T> producer);
}

Note this is close to the reserve(int) / Reservation<T> that the ticket deliberately kept private — "only genuinely needed when caller-owned work must happen between claiming capacity and filling it." SCA bisection may be exactly that case, in which case the follow-on is really the argument for making a bounded, scoped form of reservation public. Worth deciding there with the use case in hand.

24 tests, all passing. tryPutBatch and tryPut(Collection) are still present and also have no caller today; happy to trim those too, though they cost no new type.

[bric3]

This is only a first pass review.

---

I used these definitions:

An item is the value enqueued. A producer is a callback that creates or obtains that value; it may also read and reset a counter. A permit represents capacity for one item. Claiming a permit reduces the capacity available to other producers before the item is enqueued.

In these diagrams, storage means enqueueing an item in the underlying queue. An item described as stored has been enqueued. Processing means invoking the consumer with an item removed from the queue.

I asked Codex to draw these diagrams to help me understand the implementation.

The colors mean:

• Blue: public operations.
• Amber: permits.
• Purple: enqueue operations.
• Green: processing and resubmission.
• Red: refusal or abandonment.

Type hierarchy

classDiagram
    direction TB
    class WorkQueues
    class WorkQueue
    class BaseWorkQueue
    class MpscWorkQueue
    class MpmcWorkQueue
    class Reservation
    class RetryStrategy
    class RetryQueue
    class MaxRetries

    WorkQueues ..> WorkQueue : creates
    WorkQueue <|.. BaseWorkQueue
    BaseWorkQueue <|-- MpscWorkQueue
    BaseWorkQueue <|-- MpmcWorkQueue
    WorkQueue ..> Reservation : returns
    WorkQueue ..> RetryStrategy : accepts
    RetryStrategy ..> RetryQueue : receives
    RetryStrategy <|.. MaxRetries

    style WorkQueues fill:#DBEAFE,stroke:#2563EB,color:#111827,font-family:monospace
    style WorkQueue fill:#DBEAFE,stroke:#2563EB,color:#111827,font-family:monospace
    style BaseWorkQueue fill:#FEF3C7,stroke:#D97706,color:#111827,font-family:monospace
    style Reservation fill:#FEF3C7,stroke:#D97706,color:#111827,font-family:monospace
    style MpscWorkQueue fill:#EDE9FE,stroke:#7C3AED,color:#111827,font-family:monospace
    style MpmcWorkQueue fill:#EDE9FE,stroke:#7C3AED,color:#111827,font-family:monospace
    style RetryStrategy fill:#DCFCE7,stroke:#16A34A,color:#111827,font-family:monospace
    style RetryQueue fill:#DCFCE7,stroke:#16A34A,color:#111827,font-family:monospace
    style MaxRetries fill:#DCFCE7,stroke:#16A34A,color:#111827,font-family:monospace

Loading

BaseWorkQueue manages permits, calls producers, and invokes consumers. Its subclasses implement store() and retrieve() using these underlying queues:

Implementation	Producer threads	Threads removing items	Underlying queue
Bounded MPSC	Multiple	One	JCTools circular array
Bounded MPMC	Multiple	Multiple	JCTools circular array
Unbounded MPMC	Multiple	Multiple	ConcurrentLinkedQueue

The unbounded version starts with Integer.MAX_VALUE permits.

Factory methods.

Accepting an item

When a permit is obtained, tryPut(producer) calls the producer immediately. tryReserve() returns a reservation that the caller can later fill or close.

Dotted arrows represent later calls made by the caller.

flowchart TD
    subgraph ProducerPath["tryPut(producer)"]
        Put["tryPut(producer)"] --> ClaimPut["claimPlace()"]
        ClaimPut --> GrantedPut{"Permit obtained?"}
        GrantedPut -->|No| Refused["Return false; producer not called"]
        GrantedPut -->|Yes| Produce["producer.produce()"]
        Produce -->|throws| Throw["Return permit; propagate exception"]
    end

    subgraph ReservationPath["tryReserve() and reservation methods"]
        Reserve["tryReserve()"] --> ClaimReserve["claimPlace()"]
        ClaimReserve --> Reservation["Return reservation; done = !granted"]

        Reservation -.->|later caller invocation| Fill["fill(nonNullItem)"]
        Reservation -.->|later caller invocation| Close["close()"]

        Fill --> FillDone{"done?"}
        FillDone -->|Yes| FillReturn["Return without storing"]
        FillDone -->|No| Complete["Set done = true"]

        Close --> CloseDone{"done?"}
        CloseDone -->|Yes| CloseReturn["Return without changing permits"]
        CloseDone -->|No| Cancel["Set done = true; return permit"]
    end

    subgraph StoragePath["Shared storeOrRelease(item) implementation"]
        Produce -->|returns item or null| NonNull{"Item non-null?"}
        Complete --> NonNull
        NonNull -->|No| ReturnPermit["Return permit; result is false"]
        NonNull -->|Yes| Store["store(item)"]
        Store -->|accepted| KeepPermit["Keep permit claimed; result is true"]
        Store -->|refused| ReturnPermit
    end

    classDef api fill:#DBEAFE,stroke:#2563EB,color:#111827,font-family:monospace
    classDef permit fill:#FEF3C7,stroke:#D97706,color:#111827,font-family:monospace
    classDef storage fill:#EDE9FE,stroke:#7C3AED,color:#111827,font-family:monospace
    classDef refusal fill:#FEE2E2,stroke:#DC2626,color:#111827,font-family:monospace

    class Put,Reserve,Produce,Fill,Close api
    class ClaimPut,ClaimReserve,GrantedPut,Reservation,FillDone,CloseDone,Complete,Cancel,NonNull,KeepPermit permit
    class Store storage
    class Refused,Throw,FillReturn,CloseReturn,ReturnPermit refusal

Loading

Retries

The implementation retries two operations:

• storeRetrying() repeats offer(item) when the underlying queue refuses the item.
• RetryStrategy.onFailure() can resubmit an item after its consumer throws.

They can occur in this order:

1. The queue enqueues an item, repeating offer() if needed.
2. A caller removes the item and invokes its consumer.
3. If the consumer throws, the retry strategy may resubmit the item.
4. Resubmission obtains a new permit and calls the same enqueue implementation, which may repeat offer() again.

The counters measure different events. storeRetrying() counts additional calls to offer(). RetryStrategy.onFailure() receives the number of consumer failures for that item, including the current failure.

Enqueue retry

An enqueue retry calls the underlying queue's offer(item) again with the same item and the permit already claimed.

flowchart TD
    Store["store(item)"] --> Implementation{"Queue implementation"}
    Implementation -->|MPSC| Mpsc["JCTools MPSC offer(item)"]
    Implementation -->|bounded MPMC| Mpmc["JCTools strict MPMC offer(item)"]
    Implementation -->|unbounded MPMC| Linked["ConcurrentLinkedQueue.offer(item)"]

    Mpsc -->|true| Stored["Item stored"]
    Mpsc -->|false| Refused["Caller returns permit; reports failure"]
    Linked --> Stored
    Mpmc -->|true| Stored
    Mpmc -->|false| Retry["storeRetrying(): up to 63 additional offer calls"]
    Retry -->|one returns true| Stored
    Retry -->|all return false| Refused

    classDef storage fill:#EDE9FE,stroke:#7C3AED,color:#111827,font-family:monospace
    classDef refusal fill:#FEE2E2,stroke:#DC2626,color:#111827,font-family:monospace
    class Store,Implementation,Mpsc,Mpmc,Linked,Retry,Stored storage
    class Refused refusal

Loading

The MPMC implementation makes at most 64 calls to offer() in total.

JCTools offer().

Resubmitting after a consumer failure

A processing retry enqueues an item again after its consumer throws. Resubmission requires a new permit. A caller must later invoke a process...() method to process the item again.

The diagram separates the queue's call to onFailure() from the strategy's optional call to retry(item).

The dotted arrow represents optional call.

flowchart TD
    Process["processOrRetry(consumer, strategy)"] --> Take["take(): remove item and return its permit"]
    Take --> Item{"Item found?"}
    Item -->|No| Empty["Return false"]
    Item -->|Yes| Consume["consumer.accept(item)"]
    Consume -->|returns normally| Done["Return true"]
    Consume -->|throws| Strategy["onFailure(item, failureCount + 1, failure, retryQueue)"]
    Strategy --> Retry{"Strategy calls retry(item)?"}
    Retry -->|No| GiveUp["Strategy abandons item or handles it elsewhere"]
    Retry -->|Yes| Readmit["Attach failure count; request a new permit"]
    Readmit --> Granted{"Permit obtained and item stored?"}
    Granted -->|Yes| Requeued["Item waits in queue"]
    Granted -->|No| RetryRefused["retry(item) returns false"]
    RetryRefused --> Decide["Strategy decides what to do next"]
    GiveUp --> Done
    Decide --> Done
    Requeued --> Done
    Requeued -.->|later process call| Take

    classDef api fill:#DBEAFE,stroke:#2563EB,color:#111827,font-family:monospace
    classDef permit fill:#FEF3C7,stroke:#D97706,color:#111827,font-family:monospace
    classDef retry fill:#DCFCE7,stroke:#16A34A,color:#111827,font-family:monospace
    classDef refusal fill:#FEE2E2,stroke:#DC2626,color:#111827,font-family:monospace
    class Process,Empty,Done api
    class Take,Item,Readmit,Granted permit
    class Consume,Strategy,Retry,Requeued,Decide retry
    class GiveUp,RetryRefused refusal

Loading

MaxRetries(3) allows up to three resubmissions. If all three are accepted and later processed with that strategy, the consumer can be invoked four times. The first consumer failure is reported to the strategy as failure count 1.

The processing methods handle consumer failures as follows:

Method	When the consumer throws
process()	Propagates the exception.
processOrHandle()	Calls the exception handler.
processOrRetry()	Calls the retry strategy, which can resubmit the item.

In every case, the item has already been removed from the queue.

[bric3]

suggestion: Could we validate required callbacks before take()? I reproduced the swallowed-error case on all three backings at the current head:

queue.tryPut("item");
queue.processOrHandle((Consumer<String>) null, (item, failure) -> {});

This returns true, leaves size() == 0, and passes an internal NullPointerException to the handler. The caller receives neither the promised validation failure nor the item back. A null retry strategy is also accepted as the plain-processing mode. Public argument validation would keep invalid calls from consuming work; the internal null sentinels can stay private.

[datadog-prod-us1-5]

More details

Admission claims capacity before invoking producers, and cleanup returns unused permits on decline or failure. Reservations require thread confinement, and shutdown requires caller coordination to guarantee an empty queue.

Was this helpful? React 👍 or 👎

_{Open Bits AI session}

_{🤖 Bits Code Review · Commit 47038d9}

[AlexeyKuznetsov-DD]

--- Found with Codex

polish: Could we update the reservation comments at lines 635, 663–665, and 697 to match the current implementation?

They describe an array reservation holding a ring position, preserving its reservation order, and returning capacity only when the consumer drains it. PlaceReservation now reserves capacity without holding a backing position: close() immediately releases an unfilled claim, and the element joins the queue when fill() stores it. The assertions here already exercise that behavior correctly; the comments describe the older design.

[dougqh]

Good catch, those comments were left over from the position-holding design. Fixed in a4bef25: all three now say a reservation claims capacity, not a position. close() gives an unfilled claim back at once and the element joins at fill(). No assertions changed, and the contract tests pass on all backings.

Written by Claude, reviewed by @dougqh

[AlexeyKuznetsov-DD]

LGTM, left minor comment found by Codex.

[dd-octo-sts]

❌ Pull Request Too Large

This pull request adds 3192 lines while the limit is 1000 lines.
Deleted lines, moved code, test code and resources (src/test, src/testFixtures, src/latestDepTest, and other test source sets), and gradle.lockfile files are not counted.

Please split it into smaller stacked pull requests, each one targeting the branch of the previous one. See the pull request size guidelines for more details.

If this PR cannot be split, exceptionnaly add the tag: override pr size check label to bypass this workflow.