Skip to content

Queues

etcd-recipes ships three queue recipes:

Recipe Use it when
DistributedQueue FIFO by arrival
DistributedPriorityQueue Order by priority, not arrival
DistributedWorkQueue The item must survive the consumer processing it

All three extend EtcdConnector, so they share the lifecycle, exception, and connection-state surface described in Core concepts.

The first two lose an item when a consumer dies

DistributedQueue and DistributedPriorityQueue delete the item as part of taking it. The moment dequeue() returns, etcd has no memory of the item — if your process dies on the next line, the work is simply gone. That is the queue's contract, not a bug (Curator's queues behave the same), and it is the right trade when items are cheap or replayable.

When losing an item is not acceptable, use DistributedWorkQueue. It claims instead of deletes, and redelivers what a dead consumer never finished.

The take side

DistributedQueue and DistributedPriorityQueue differ only in how items are ordered. Everything about taking them is shared, and lives in AbstractQueue:

abstract class AbstractQueue : EtcdConnector {
  fun dequeue(): ByteSequence
  fun tryDequeue(): ByteSequence?
  fun poll(timeout: Duration): ByteSequence?
  fun poll(timeout: Long, timeUnit: TimeUnit): ByteSequence?
  val size: Int
}
Call When the queue is empty
dequeue() Blocks until an item arrives
tryDequeue() Returns null immediately
poll(timeout) Blocks up to timeout, then returns null

Every take is a transaction that deletes the head key guarded on its mod revision, so exactly one consumer wins each item no matter how many are racing. A loser does not fail — it re-reads the new head and tries again. When the queue is empty the take parks on a watcher rather than polling, and re-reads the head after waking: the first PUT the watcher happens to see is not necessarily the head by sort order.

size is an RPC, and advisory

etcd cannot push a count, so size issues a range-count on every read. It is true at some recent revision and may be stale by your next line — fine for logging and dashboards, useless for deciding whether a take will succeed. Use tryDequeue/poll for that; they are atomic. The same warning applies to Micrometer's bindQueueDepth gauge, which calls size on every scrape.

DistributedQueue

FIFO by arrival, ordered by mod revision (SortTarget.MOD) — etcd's own commit order, not a client clock.

DistributedQueue(client, "/queues/orders").use { queue ->
  queue.enqueue("order-1")

  // Blocks until an item is available. Exactly one consumer across the cluster
  // wins each item: the take is a CAS delete guarded on the item's mod revision.
  val value = queue.dequeue()
  logger.info { "Dequeued ${value.asString}" }
}

try (DistributedQueue queue = new DistributedQueue(client, "/queues/orders")) {
  queue.enqueue("order-1");

  // Blocks until an item is available. Exactly one consumer across the cluster
  // wins each item: the take is a CAS delete guarded on the item's mod revision.
  ByteSequence value = queue.dequeue();
  System.out.println("Dequeued " + getAsString(value));
}

enqueue is overloaded for String, Int, Long, and ByteSequence. The take side always hands back a ByteSequence; asString / asInt / asLong convert it back (Java: the static ByteSequenceUtils.getAsString(…) and friends).

DistributedQueue(client, "/queues/orders").use { queue ->
  // Non-blocking: null the moment the queue is found empty.
  val immediate = queue.tryDequeue()

  // Bounded: waits under a watcher, then gives up and returns null.
  val waited = queue.poll(5.seconds)

  logger.info { "tryDequeue=${immediate?.asString}, poll=${waited?.asString}" }
}

try (DistributedQueue queue = new DistributedQueue(client, "/queues/orders")) {
  // Non-blocking: null the moment the queue is found empty.
  ByteSequence immediate = queue.tryDequeue();

  // Java uses the (long, TimeUnit) overload; the Duration one is Kotlin-facing.
  ByteSequence waited = queue.poll(5, TimeUnit.SECONDS);

  System.out.println("tryDequeue=" + immediate + ", poll=" + waited);
}

Kotlin takes Duration, Java takes (long, TimeUnit)

Both poll overloads exist. The kotlin.time.Duration one reads better from Kotlin; the (long, TimeUnit) one is the one Java can call at all — a Kotlin Duration parameter mangles the JVM method name, so poll(Duration) is not reachable from Java. Same story for receive on the work queue.

Batch enqueue

enqueueAll writes every value in one transaction:

DistributedQueue(client, "/queues/orders").use { queue ->
  // One transaction: either every value lands or none does. The keys embed the
  // argument index, so within the batch consumers see them in argument order.
  queue.enqueueAll(
    [
      "order-1".asByteSequence,
      "order-2".asByteSequence,
      "order-3".asByteSequence,
    ],
  )

  // size costs a range-count RPC on every read — it is not a cached counter.
  logger.info { "Depth: ${queue.size}" }
}

try (DistributedQueue queue = new DistributedQueue(client, "/queues/orders")) {
  // One transaction: either every value lands or none does. The keys embed the
  // argument index, so within the batch consumers see them in argument order.
  queue.enqueueAll(
    List.of(
      getAsByteSequence("order-1"),
      getAsByteSequence("order-2"),
      getAsByteSequence("order-3")));

  // getSize() costs a range-count RPC on every read — it is not a cached counter.
  System.out.println("Depth: " + queue.getSize());
}

All-or-nothing matters more than it looks. A loop of individual enqueue calls that dies halfway leaves a partial batch that consumers are already draining, and nothing tells them the rest is never coming. One transaction has no such state: either every entry is visible at the same revision, or none is.

Batched entries share that revision, so mod-revision order cannot separate them. Their keys embed the argument index instead, which is what keeps within-batch order equal to argument order.

Scoped usage

// Builds the queue, runs the block with it as receiver, closes it on the way out.
val value = withDistributedQueue(client, "/queues/orders") {
  enqueue("order-1")
  dequeue()
}
logger.info { "Dequeued ${value.asString}" }

withDistributedQueue builds the queue, runs the block against it as receiver, and closes it on the way out. Java has no equivalent — use try-with-resources, as the tabs above do. See the Java guide.

DistributedPriorityQueue

Ordered by priority instead of arrival: entries are filed under a zero-padded priority prefix and taken in key order (SortTarget.KEY). Lower number wins.

// minimumWaitTime has no default on the constructor — pass ZERO for no pacing.
DistributedPriorityQueue(client, "/queues/jobs", minimumWaitTime = Duration.ZERO).use { queue ->
  queue.enqueue("nightly-report", 200)
  queue.enqueue("page-oncall", 1)

  // Lower number wins: "page-oncall" comes out first, regardless of enqueue order.
  logger.info { "Dequeued ${queue.dequeue().asString}" }
}

Note that minimumWaitTime has no default on the constructor — pass Duration.ZERO when you do not want pacing. The scoped function does default it:

// Here minimumWaitTime does default to zero.
withDistributedPriorityQueue(client, "/queues/jobs") {
  enqueue("page-oncall", 1)
  enqueue("nightly-report", 200)
}

val job = withDistributedPriorityQueue(client, "/queues/jobs") { poll(5.seconds) }
logger.info { "Dequeued ${job?.asString}" }

Java cannot construct DistributedPriorityQueue

minimumWaitTime is a kotlin.time.Duration, and it has no default — so every generated constructor carries a Duration parameter, and the Kotlin compiler marks constructors with inline-class parameters synthetic. There is no Java-visible constructor left. The enqueue(value, priority: Int) overloads are perfectly callable; it is only the construction that Java cannot express. Instantiate it from Kotlin, or use DistributedQueue / DistributedWorkQueue, both of which Java constructs normally.

Priorities are 0..65535

withDistributedPriorityQueue(client, "/queues/jobs") {
  // Int priorities are range-checked against 0..65535 and throw
  // IllegalArgumentException outside it, rather than silently wrapping mod 65536
  // and filing the entry in the wrong bucket.
  enqueue("page-oncall", 1)

  // UShort is the underlying priority type, so this overload needs no check.
  enqueue("nightly-report", 200u)
}

Eight enqueue overloads cover four value types (String, Int, Long, ByteSequence) times two priority types (Int, UShort). UShort is the real one: priorities are stored as a five-digit key prefix. The Int overloads exist for convenience and range-check their argument, throwing IllegalArgumentException outside 0..65535 rather than letting toUShort() wrap silently — a wrapped 70000 becomes 4464, which is not an error anywhere, just an item that quietly sorts into the wrong bucket.

minimumWaitTime

// Each enqueue on this instance sleeps until at least 20ms has passed since the
// previous one, thinning out same-priority CAS contention on the sequence key.
// It paces this instance only — it is not a cluster-wide rate limit.
withDistributedPriorityQueue(client, "/queues/jobs", minimumWaitTime = 20.milliseconds) {
  repeat(10) { i -> enqueue("job-$i", 1) }
}

An enqueue at a given priority derives its sequence number from the current last child at that priority, then commits under a CAS. Two producers computing the same sequence number is ordinary contention: the loser re-reads and retries, up to 50 attempts before throwing EtcdRecipeRuntimeException. minimumWaitTime spaces out writes from this instance to thin that contention. It is not a cluster-wide rate limit, and it sleeps the calling thread.

Concurrent consumers do not see a global order

A single consumer drains a priority queue in exact priority order. With several consumers, each take returns the head as of that moment, but the order in which items are observed across consumers is not a global ordering — one consumer may be slower to log than another that took a lower-priority item next. The guarantee is per-item: each is delivered to exactly one consumer.

DistributedWorkQueue

The other two queues answer "who gets this item?". This one answers "who finished this item?" — a much stronger question, and the reason this recipe exists.

A received item is not deleted. It moves to claimed/, and a separate claim marker is written under the consumer's lease. The item is only removed when you ack() it.

DistributedWorkQueue(client, "/workqueue/jobs").use { queue ->
  queue.enqueue("job-1")

  // The item is claimed, not deleted: it survives in etcd until it is acked, so
  // a crash here redelivers it instead of losing it.
  val item = queue.receive()
  logger.info { "Processing ${item.value.asString} (attempt ${item.attempt})" }

  // ack() completes the item. false means the claim was already lost, so the
  // work may have been redone elsewhere — do not treat it as success.
  if (!item.ack()) {
    logger.warn { "Lost the claim for ${item.id}" }
  }
}

try (DistributedWorkQueue queue = new DistributedWorkQueue(client, "/workqueue/jobs")) {
  queue.enqueue("job-1");

  // The item is claimed, not deleted: it survives in etcd until it is acked, so
  // a crash here redelivers it instead of losing it.
  DistributedWorkQueue.WorkItem item = queue.receive(30, TimeUnit.SECONDS);
  if (item != null) {
    System.out.println("Processing " + getAsString(item.getValue()) + " (attempt " + item.getAttempt() + ")");

    // ack() completes the item. false means the claim was already lost, so the
    // work may have been redone elsewhere — do not treat it as success.
    if (!item.ack()) {
      System.out.println("Lost the claim for " + item.getId());
    }
  }
}

The payload is deliberately not bound to the consumer's lease; only the marker is. So when a consumer dies, its markers evaporate with its lease while the payloads survive, and any consumer's sweep moves the orphans back to the queue — under the same key, so they return to their original FIFO position rather than the back.

receive() blocks, receive(timeout) bounds the wait, tryReceive() does not wait at all. Each returns a WorkItem:

inner class WorkItem {
  val id: String
  val value: ByteSequence
  val attempt: Int
  fun ack(): Boolean
  fun requeue(): Boolean
}

Configuration

val config =
  WorkQueueConfig(
    // How long after a consumer *dies* its claims become reclaimable. A live
    // consumer renews its lease, so this does not bound processing time.
    visibilityTimeoutSecs = 30,
    // Delivery attempts before an item is dead-lettered instead of redelivered.
    maxDeliveries = 5,
    // How often each consumer sweeps for orphaned claims and matured delays.
    sweepInterval = 30.seconds,
  )

DistributedWorkQueue(client, "/workqueue/jobs", config, clientId = "worker-1").use { queue ->
  logger.info { "Consumer ${queue.clientId} ready" }
}

// Java sees only the (visibilityTimeoutSecs) and (visibilityTimeoutSecs,
// maxDeliveries) constructors: sweepInterval is a kotlin.time.Duration, so its
// constructor is not callable from Java. The default 30s sweep applies.
WorkQueueConfig config = new WorkQueueConfig(30, 5);

try (DistributedWorkQueue queue = new DistributedWorkQueue(client, "/workqueue/jobs", config)) {
  System.out.println("Consumer " + queue.getClientId() + " ready");
}

visibilityTimeoutSecs bounds crash detection, not processing time

This is the single most misread knob on the recipe, because the name is borrowed from SQS, where it does bound processing.

Here it is the TTL of the consumer's lease. A live consumer renews that lease in the background for as long as it is alive, so an item may legitimately take ten minutes to process under a 30-second visibility timeout without ever being redelivered. What the timeout actually bounds is how long the queue waits before concluding a silent consumer is dead and handing its claims to someone else.

So tune it against how fast you want crash recovery, not against how slow your handler is. Lower means faster redelivery after a genuine crash, and less tolerance for a consumer that is merely partitioned.

At-least-once means idempotent

DistributedWorkQueue(client, "/workqueue/jobs").use { queue ->
  while (true) {
    val item = queue.receive(30.seconds) ?: break

    // Delivery is at-least-once, so this body must be idempotent: a redelivery
    // is indistinguishable from a first delivery except for item.attempt.
    try {
      handle(item.value.asString)
      item.ack()
    } catch (e: IllegalStateException) {
      // Hand it back now, attempts preserved, rather than making every other
      // consumer wait out the visibility timeout.
      logger.warn(e) { "Attempt ${item.attempt} of ${item.id} failed; requeueing" }
      item.requeue()
    }
  }
}

Delivery is at-least-once, and the duplicate is not hypothetical: a consumer partitioned for longer than its visibility timeout keeps working on an item that has already been redelivered elsewhere. Both consumers run the handler. Only one ack() wins — the other returns false, because the ack is guarded on the claim still belonging to this consumer.

Which is why ack() and requeue() return Boolean rather than Unit, and why false deserves a log line: it does not mean the work failed, it means the work may have been done twice. Your handler has to be safe under that regardless — false arrives after the side effects, not before. Make the handler idempotent (upsert on a key derived from WorkItem.id, or a dedupe table) and treat false as a signal for your metrics rather than something to compensate for.

requeue() hands an item back immediately with its attempt count preserved — the right move when a handler fails fast and you would rather not make every other consumer wait out the visibility timeout for a crash that did not happen.

Dead letters

An item that keeps failing cannot be redelivered forever. Once its attempts reach maxDeliveries, the reclaim sweep routes it to the dead-letter space instead of back to the queue:

DistributedWorkQueue(client, "/workqueue/jobs").use { queue ->
  // Items that exhausted maxDeliveries land here instead of being redelivered
  // forever. Nothing drains this automatically — it is an operator surface.
  for (dead in queue.deadLetters()) {
    logger.warn { "Dead letter ${dead.id} after ${dead.attempts} attempts: ${dead.value.asString}" }
  }

  // Once the underlying bug is fixed, replay one with a fresh attempt count...
  queue.requeueDeadLetter("1700000000000-abc")

  // ...or drop a genuinely poisonous payload for good. Both return false when
  // no such dead letter exists.
  queue.purgeDeadLetter("1700000000001-def")
}

try (DistributedWorkQueue queue = new DistributedWorkQueue(client, "/workqueue/jobs")) {
  // Items that exhausted maxDeliveries land here instead of being redelivered
  // forever. Nothing drains this automatically — it is an operator surface.
  for (DistributedWorkQueue.DeadLetter dead : queue.deadLetters()) {
    System.out.println(
      "Dead letter " + dead.getId()
        + " after " + dead.getAttempts() + " attempts: "
        + getAsString(dead.getValue()));
  }

  // Replay one with a fresh attempt count, or drop it for good. Both return
  // false when no such dead letter exists.
  queue.requeueDeadLetter("1700000000000-abc");
  queue.purgeDeadLetter("1700000000001-def");
}

deadLetters() lists them as DeadLetter(id, value, attempts). Nothing drains that list automatically — that is the point. A dead letter is a poison pill that has already burned maxDeliveries attempts across your fleet, and the decision of whether it is a bug to fix (requeueDeadLetter(id), which returns it with a fresh attempt count) or garbage to drop (purgeDeadLetter(id)) is not one the library can make. Alert on the list being non-empty.

Delayed delivery

DistributedWorkQueue(client, "/workqueue/jobs").use { queue ->
  // Invisible until it matures, then it takes its place among the ready items.
  // Maturity is judged against client clocks, so producer/consumer skew shifts
  // delivery by the skew.
  queue.enqueue("send-reminder", 5.seconds)

  // Deliverable immediately.
  queue.enqueue("send-welcome")
}

An item enqueued with a delay is parked out of sight until it matures, then promoted into the queue in ready-time order, interleaving correctly with items enqueued immediately. A zero or negative delay enqueues immediately.

Maturity is judged against client clocks

The ready time is stamped by the producer's clock and compared against the consumer's. Skew shifts delivery by the skew. At visibility-timeout scale (seconds) that is tolerable; do not build a scheduler that needs millisecond accuracy on top of it.

The delayed enqueue(value, delay) overload takes a kotlin.time.Duration, so like poll(Duration) it is not callable from Java.

The consumer lease

DistributedWorkQueue(client, "/workqueue/jobs").use { queue ->
  // Every claim marker this consumer makes hangs off one lease. If that lease
  // expires, all its outstanding claims become reclaimable and their acks fail.
  queue.addLeaseListener { event ->
    when (event) {
      is LeaseEvent.Expired -> logger.warn { "Lease expired; claims are reclaimable" }
      is LeaseEvent.Restored -> logger.info { "Lease healed: ${event.newLeaseId}" }
      is LeaseEvent.Failed -> logger.error { "Lease healing abandoned; claims will not hold" }
      is LeaseEvent.Suspended -> logger.debug { "Keep-alive hiccup; jetcd is retrying" }
    }
  }

  queue.receive(30.seconds)?.ack()
}

Every claim this consumer makes hangs off a single self-healing lease, granted lazily — a producer that only ever calls enqueue never takes a lease or starts a sweeper thread.

The lease heals for future claims only. Markers made under a dead lease are deliberately allowed to die with it, because that is exactly the visibility contract: their items become reclaimable. Healing them would resurrect claims on work that another consumer has already been handed. So a LeaseEvent.Expired means your outstanding claims are gone even though healing will succeed. See Leases and loss.

Closing the queue revokes the lease, which makes any unacked claims reclaimable at once rather than after the visibility timeout — a clean shutdown returns work to the fleet immediately.

Two gaps worth knowing

Unlike the other two queues, the work queue has:

  • no scoped function — there is no withDistributedWorkQueue. Use use { } in Kotlin, try-with-resources in Java.
  • no typed variant — there is no TypedDistributedWorkQueue. Marshal values through an EtcdCodec by hand: queue.enqueue(codec.encode(value)) and codec.decode(item.value).

Neither is a design statement; they are simply not built yet.

Typed queues

TypedDistributedQueue<T> and TypedDistributedPriorityQueue<T> marshal values through an EtcdCodec<T>, so callers hand over T instead of ByteSequence:

TypedDistributedQueue(client, "/queues/orders", jsonCodec<Order>()).use { queue ->
  queue.enqueue(Order(1, "widget"))
  queue.enqueueAll([Order(2, "gadget"), Order(3, "gizmo")])

  val order: Order = queue.dequeue()
  logger.info { "Dequeued $order" }

  // The typed wrapper is a Closeable decorator, not an EtcdConnector: the
  // connector API lives on the instance it wraps.
  logger.info { "Healthy: ${queue.untyped.isHealthy()}, exceptions: ${queue.untyped.exceptions.size}" }
}

TypedDistributedPriorityQueue(client, "/queues/jobs", jsonCodec<Order>()).use { queue ->
  queue.enqueue(Order(1, "widget"), priority = 1)
  val order: Order = queue.dequeue()
  logger.info { "Dequeued $order" }
}

The typed wrappers are decorators, not connectors

They implement Closeable — they do not extend EtcdConnector. So exceptions, isHealthy(), connectionState, and the listener registrations are not on the typed instance; they are on the queue it wraps, reachable through the untyped property. close() delegates, so use { } still does the right thing.

Codecs, including jsonCodec<T>(), are covered in Typed values.

Coroutines

The blocking take on a queue costs you a thread for its whole wait, which is a poor trade when the wait is unbounded. The suspending twins release it:

DistributedQueue(client, "/queues/orders").use { queue ->
  queue.awaitEnqueue("order-1")

  // Suspends rather than parking a thread. Cancelling the wait consumes nothing:
  // the queue is left exactly as it was.
  val value = queue.receive()
  logger.info { "Received ${value.asString}" }
}

DistributedWorkQueue(client, "/workqueue/jobs").use { queue ->
  queue.awaitEnqueue("job-1")

  // Suspends until an item can be claimed; cancelling leaves no orphan claim.
  val item = queue.awaitReceive()
  logger.info { "Processing ${item.value.asString}" }
  item.awaitAck()
}

awaitEnqueue, receive (the suspending twin of dequeue), awaitTryDequeue, awaitReceive, awaitTryReceive, WorkItem.awaitAck, and WorkItem.awaitRequeue all exist. Cancelling a wait consumes nothing and leaves no orphan claim. See Coroutines.

Observability

With the Micrometer module wired in, dequeue and poll record to the etcd.queue timer (tagged op), measuring call → item in hand, so the timer includes the wait on an empty queue. Enqueues and the work queue's receive are not instrumented today. Queue paths never become tags — that would blow up cardinality.

bindQueueDepth(queue) binds a gauge to size, at one range-count RPC per scrape; on a hot queue with frequent scrapes, that load is worth a thought. See Observability.