Add WorkQueue<T> with MPSC and linked-queue backings

utils/queue-utils/src/main/java/datadog/common/queue/BiContextualProducer.java

@@ -0,0 +1,22 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * A {@link Producer} that derives its element from two caller-supplied contexts.
+ *
+ * <p>Two rather than one because the second context is typically a value the call site hoisted out
+ * of a loop — a schema, a clock reading, a per-batch buffer — that the item alone cannot recover.
+ * Carrying it as a parameter is what keeps the producer a non-capturing bound-once field and keeps
+ * the hoist visible where it happens, instead of a per-iteration capture or a cached binding that
+ * can silently go stale.
+ *
+ * <p>The ladder stops here on purpose. A third context is usually derivable from the item, and a
+ * primitive one has to be boxed to ride a generic parameter, which costs more than re-deriving it.
+ * A call site that genuinely needs more should close over what it needs once per scope.
+ */
+@Strategy
+@FunctionalInterface
+public interface BiContextualProducer<C1, C2, T> {
+  T produce(C1 first, C2 second);
+}

utils/queue-utils/src/main/java/datadog/common/queue/ContextualProducer.java

@@ -0,0 +1,15 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * A {@link Producer} that derives its element from a caller-supplied context.
+ *
+ * <p>The context parameter is what lets the producer stay non-capturing: state the element needs is
+ * passed in at the call site rather than closed over.
+ */
+@Strategy
+@FunctionalInterface
+public interface ContextualProducer<C, T> {
+  T produce(C context);
+}

utils/queue-utils/src/main/java/datadog/common/queue/ExceptionHandler.java

@@ -0,0 +1,24 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Deals with a consumer's failure and lets the item go, for a caller who wants to see what went
+ * wrong without deciding whether to try again. The item is dropped either way, and the failure does
+ * not reach the caller of {@code processOrHandle}.
+ *
+ * <p>The narrow half of {@link RetryStrategy}: reach for that one when the answer to a failure is
+ * sometimes "again", and this one when it is only ever "record it and move on". The item comes
+ * along because the consumer that threw is in no position to say which one died.
+ *
+ * @see WorkQueue#processOrHandle(java.util.function.Consumer, ExceptionHandler)
+ */
+@Strategy
+@FunctionalInterface
+public interface ExceptionHandler<T> {
+  /**
+   * Called on the consuming thread, in place of propagating. A handler that throws propagates in
+   * the failure's stead.
+   */
+  void handle(T item, Throwable failure);
+}

utils/queue-utils/src/main/java/datadog/common/queue/LinkedWorkQueue.java

@@ -0,0 +1,39 @@
+package datadog.common.queue;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * A {@link WorkQueue} over a {@link ConcurrentLinkedQueue}: multi-producer, multi-consumer,
+ * optionally bounded.
+ *
+ * <p>This backing exists to give call sites that cannot yet take an MPSC ring — because they have
+ * several consumers, or no defensible capacity — the admission and lifecycle contract anyway, so
+ * they can be migrated behind {@link WorkQueue} first and re-backed later. It keeps the linked
+ * queue's per-element node, so it does not deliver the allocation win; prefer {@link
+ * MpscWorkQueue}.
+ *
+ * <p>Storage only: the bound lives in {@link BaseWorkQueue}, which is what replaces the hand-rolled
+ * cap plus O(n) {@code ConcurrentLinkedQueue.size()} walk such a call site otherwise pays on every
+ * admission, and makes {@link #size()} constant-time.
+ */
+final class LinkedWorkQueue<T> extends BaseWorkQueue<T> {
+
+  private final ConcurrentLinkedQueue<Object> queue = new ConcurrentLinkedQueue<>();
+
+  /**
+   * @param capacity the bound, or {@link Integer#MAX_VALUE} to leave the queue unbounded
+   */
+  LinkedWorkQueue(int capacity) {
+    super(capacity);
+  }
+
+  @Override
+  boolean store(Object element) {
+    return queue.offer(element);
+  }
+
+  @Override
+  Object retrieve() {
+    return queue.poll();
+  }
+}

utils/queue-utils/src/main/java/datadog/common/queue/MaxRetries.java

@@ -0,0 +1,15 @@
+package datadog.common.queue;
+
+/** A {@link RetryStrategy} that resubmits an item until a fixed attempt count is reached. */
+public final class MaxRetries<T> implements RetryStrategy<T> {
+  private final int maxRetries;
+
+  public MaxRetries(int maxRetries) {
+    this.maxRetries = maxRetries;
+  }
+
+  @Override
+  public boolean onFailure(T item, int attempt, Throwable failure, RetryQueue<T> retryQueue) {
+    return attempt < maxRetries && retryQueue.retry(item);
+  }
+}

utils/queue-utils/src/main/java/datadog/common/queue/MpscWorkQueue.java

@@ -0,0 +1,43 @@
+package datadog.common.queue;
+
+import org.jctools.queues.MessagePassingQueue;
+
+/**
+ * A {@link WorkQueue} over a JCTools MPSC array queue: many producers, one consumer, no per-element
+ * node. The preferred backing.
+ *
+ * <p>Storage only. The bound and the reserve-before-construct guarantee both live in {@link
+ * BaseWorkQueue}, which spends a place before it calls any producer, so by the time an element
+ * reaches {@link #store} the ring is known to have room for it.
+ *
+ * <p>That the ring could have enforced its own bound, inside a CAS it was performing anyway, is the
+ * cost of this arrangement — see {@link BaseWorkQueue} for what it buys. What it avoids is holding
+ * a ring position open across a caller-controlled gap: the ring reports a claimed-but-unfilled
+ * position as empty, so a reservation that held one would stall the consumer, and would need a
+ * placeholder object per reservation for the consumer to tell an abandoned position from a pending
+ * one.
+ */
+final class MpscWorkQueue<T> extends BaseWorkQueue<T> {
+
+  private final MessagePassingQueue<Object> queue;
+
+  MpscWorkQueue(int requestedCapacity) {
+    this(Queues.<Object>mpscArrayQueue(requestedCapacity));
+  }
+
+  /** Takes the queue already built, so the bound can be the capacity it actually rounded up to. */
+  private MpscWorkQueue(MessagePassingQueue<Object> queue) {
+    super(queue.capacity());
+    this.queue = queue;
+  }
+
+  @Override
+  boolean store(Object element) {
+    return queue.offer(element);
+  }
+
+  @Override
+  Object retrieve() {
+    return queue.poll();
+  }
+}

utils/queue-utils/src/main/java/datadog/common/queue/Producer.java

@@ -0,0 +1,22 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Produces an element for admission into a {@link WorkQueue}.
+ *
+ * <p>A producer is only invoked once a place has been claimed, so it is never called for an element
+ * that will be rejected. Implementations must be non-capturing — a {@code static final} constant of
+ * the concrete type, or a lambda that closes over nothing — which is what {@link Strategy} marks.
+ *
+ * <p>That is not a preference, it is the whole reason this form exists. A capturing lambda
+ * allocates once per call, and so does a {@link Reservation}; the reservation is straight-line code
+ * that keeps whatever the call site had hoisted and needs no context parameters. So a producer that
+ * captures is strictly worse than the reserve form it was meant to improve on. If the state will
+ * not fit the context parameters, use {@link WorkQueue#tryReserve} rather than closing over it.
+ */
+@Strategy
+@FunctionalInterface
+public interface Producer<T> {
+  T produce();
+}

utils/queue-utils/src/main/java/datadog/common/queue/Reservation.java

@@ -0,0 +1,57 @@
+package datadog.common.queue;
+
+/**
+ * A claimed place in a {@link WorkQueue}, for a caller whose work between claiming and filling
+ * cannot be expressed as a {@link Producer}.
+ *
+ * <p>What is claimed is capacity, never a position: the element joins the queue where it is filled,
+ * not where it was claimed, so an open reservation holds no place a consumer could be waiting on
+ * and cannot stall one. A reservation that is neither filled nor closed does leak its capacity,
+ * quietly and permanently, which is why this is an {@link AutoCloseable} meant for
+ * try-with-resources.
+ *
+ * <p>A refused claim is a reservation too, rather than a {@code null}, and one that quietly
+ * discards whatever is filled into it. Nothing about the failed path throws, so the shortest
+ * correct call site is also the obvious one:
+ *
+ * <pre>{@code
+ * try (Reservation<Task> place = queue.tryReserve()) {
+ *   place.fill(buildTask());
+ * }
+ * }</pre>
+ *
+ * <p>Consulting {@link #granted} first is what buys the reserve-first guarantee — skip the build
+ * and nothing is allocated for a queue that had no room for it:
+ *
+ * <pre>{@code
+ * try (Reservation<Task> place = queue.tryReserve()) {
+ *   if (place.granted()) {
+ *     place.fill(buildTask());
+ *   }
+ * }
+ * }</pre>
+ */
+public interface Reservation<T> extends AutoCloseable {
+
+  /**
+   * Whether a place was actually claimed. Worth asking before building anything expensive: a
+   * refused reservation accepts a fill and throws it away, so checking is what turns
+   * allocate-then-drop into never-allocate.
+   *
+   * @return whether a fill will be kept
+   */
+  boolean granted();
+
+  /**
+   * Publishes {@code element} into the claimed place. A granted place is already paid for, so this
+   * cannot be rejected; a refused one discards the element, having already counted the drop.
+   */
+  void fill(T element);
+
+  /**
+   * Gives the place back if it was never filled, immediately. Filling first makes this a no-op, and
+   * nothing is ever consumed for a released place.
+   */
+  @Override
+  void close();
+}

utils/queue-utils/src/main/java/datadog/common/queue/RetryQueue.java

@@ -0,0 +1,32 @@
+package datadog.common.queue;
+
+/**
+ * The capability to resubmit work after a consumer failure.
+ *
+ * <p>Only obtainable inside {@link RetryStrategy#onFailure}, never from a plain consumer, so
+ * re-enqueue-after-failure stays visibly distinct from ordinary admission.
+ */
+public interface RetryQueue<T> {
+  /**
+   * Resubmits the failed item.
+   *
+   * <p>The failed item's place was given back when it was consumed, so this claims a place like any
+   * other admission and can be rejected if the queue filled up behind it; a rejected retry counts
+   * as a drop. This is the overload every ordinary strategy wants: it resubmits without allocating
+   * the array the varargs form needs.
+   *
+   * @return whether the item was resubmitted
+   */
+  boolean retry(T item);
+
+  /**
+   * Resubmits several items in place of the failed item.
+   *
+   * <p>Each piece claims its own place, so a partition can be admitted only in part; the return
+   * value reports whether all of them made it, and each rejection counts as a drop.
+   *
+   * @return whether every item was resubmitted
+   */
+  @SuppressWarnings("unchecked")
+  boolean retry(T... items);
+}

utils/queue-utils/src/main/java/datadog/common/queue/RetryStrategy.java

@@ -0,0 +1,21 @@
+package datadog.common.queue;
+
+import datadog.trace.api.function.Strategy;
+
+/**
+ * Decides what happens to an item whose consumer threw.
+ *
+ * <p>Invoked only on failure — a successful consumption needs no callback. The return value reports
+ * the decision; it does not report whether the item will eventually succeed. Logging and counting
+ * are the caller's to compose here: this API performs neither.
+ */
+@Strategy
+@FunctionalInterface
+public interface RetryStrategy<T> {
+  /**
+   * @param attempt how many times this item has been consumed unsuccessfully, including now, so the
+   *     first failure reports {@code 1}
+   * @return {@code true} if the item was resubmitted, {@code false} if the strategy gave up
+   */
+  boolean onFailure(T item, int attempt, Throwable failure, RetryQueue<T> retryQueue);
+}