Barriers¶

A barrier makes clients wait for something other than a lock. etcd-recipes ships three, and they answer three different questions:

The first is a gate: someone holds it shut, everyone else queues, and one removeBarrier() releases the lot. The second is a rendezvous with no gatekeeper — the last arrival releases everyone, including itself. The third is the second one twice, wrapped around a phase of work.

DistributedBarrier and DistributedBarrierWithCount extend EtcdConnector and share the lifecycle, exception, and connection-state surface described in Core concepts. DistributedDoubleBarrier does not — see below.

DistributedBarrier¶

One client arms the barrier, does whatever the others must not race, and removes it:

DistributedBarrier(client, "/barriers/import").use { barrier ->
  // false means another client already holds the barrier — this instance did
  // not arm it, and must not assume it may lift it.
  if (barrier.setBarrier()) {
    logger.info { "Barrier armed; every waiter blocks" }

    // ... do the work that the waiters must not race ...

    // Releases every waiter at once.
    barrier.removeBarrier()
  } else {
    logger.info { "Someone else armed it" }
  }
}

try (DistributedBarrier barrier = new DistributedBarrier(client, "/barriers/import")) {
  // false means another client already holds the barrier — this instance did
  // not arm it, and must not assume it may lift it.
  if (barrier.setBarrier()) {
    System.out.println("Barrier armed; every waiter blocks");

    // ... do the work that the waiters must not race ...

    // Releases every waiter at once.
    barrier.removeBarrier();
  } else {
    System.out.println("Someone else armed it");
  }
}

setBarrier() returning false is not a failure — it means another client got there first and the barrier is already armed by somebody else. That distinction matters: the client that armed it is the one that should remove it. removeBarrier() returns false if this instance already removed it.

Everyone else waits:

DistributedBarrier(client, "/barriers/import").use { barrier ->
  // Advisory: true-at-some-recent-revision, and the barrier may lift a
  // microsecond later. Never branch on it to decide whether to wait.
  if (barrier.isBarrierSet()) logger.info { "Barrier is up; blocking" }

  // Blocks until the barrier key is deleted.
  barrier.waitOnBarrier()
  logger.info { "Released" }
}

try (DistributedBarrier barrier = new DistributedBarrier(client, "/barriers/import")) {
  // Advisory: true-at-some-recent-revision, and the barrier may lift a
  // microsecond later. Never branch on it to decide whether to wait.
  if (barrier.isBarrierSet()) {
    System.out.println("Barrier is up; blocking");
  }

  // Blocks until the barrier key is deleted.
  barrier.waitOnBarrier();
  System.out.println("Released");
}

waitOnBarrier() blocks until the barrier key is deleted, watching for the DELETE rather than polling. The bounded overloads return false on timeout:

DistributedBarrier(client, "/barriers/import").use { barrier ->
  // false means the timeout elapsed with the barrier still up. Waiting again
  // is fine — a waiter holds no state between calls.
  if (barrier.waitOnBarrier(30.seconds)) {
    logger.info { "Released within the timeout" }
  } else {
    logger.info { "Still blocked after 30s" }
  }
}

try (DistributedBarrier barrier = new DistributedBarrier(client, "/barriers/import")) {
  // Java uses the (long, TimeUnit) overload; the Duration one is Kotlin-facing.
  // false means the timeout elapsed with the barrier still up. Waiting again is
  // fine — a waiter holds no state between calls.
  if (barrier.waitOnBarrier(30, TimeUnit.SECONDS)) {
    System.out.println("Released within the timeout");
  } else {
    System.out.println("Still blocked after 30s");
  }
}

A waiter holds no state between calls, so timing out and waiting again is free — that is exactly what a "wait, log progress, wait again" loop does.

Waiting on a barrier nobody set¶

By default (waitOnMissingBarriers = true) a waiter blocks on a barrier that does not exist yet, on the theory that it is about to. That is what you want when waiters may start before the coordinator does — arriving early should not mean sailing straight through.

Set it to false and a missing barrier reads as "already lifted", so waitOnBarrier() returns true immediately:

// waitOnMissingBarriers=false: an unset barrier is treated as already lifted,
// so waitOnBarrier() returns true immediately instead of blocking until some
// client arms and then removes it.
DistributedBarrier(
  client = client,
  barrierPath = "/barriers/import",
  leaseTtlSecs = 5L,
  waitOnMissingBarriers = false,
).use { barrier ->
  barrier.waitOnBarrier()
}

// waitOnMissingBarriers=false: an unset barrier is treated as already lifted,
// so waitOnBarrier() returns true immediately. Java has no named arguments, so
// every preceding parameter must be supplied positionally.
try (DistributedBarrier barrier =
       new DistributedBarrier(
         client,
         "/barriers/import",
         5L,        // leaseTtlSecs
         false)) {  // waitOnMissingBarriers
  barrier.waitOnBarrier();
}

isBarrierSet() is advisory — true at some recent revision, and possibly false by the time you read the answer. Use it for logging; never branch on it to decide whether to wait, because waitOnBarrier() already handles the "it lifted while I was asking" race and your if does not.

The barrier lease heals, and that is a trade-off¶

The barrier key is bound to a lease, and unlike a lock's lease that one is self-healing. The reasoning is the same as for service registration: nobody else is competing to own your barrier key, so re-creating it after an expiry does not race anyone.

But a healed barrier is not the same as a barrier that never lapsed, and the difference is visible to waiters:

1. the lease expires (a partition longer than the TTL),
2. etcd deletes the key — every waiter sees a DELETE and lifts,
3. the healer grants a new lease and re-arms the barrier for future waiters,
4. the expiry is recorded on exceptions and drives connectionState.

Step 2 is a spurious lift, and it is unavoidable: by the time this client can react, the key is already gone and the waiters have already been released. No amount of healing can un-ring that bell. What healing buys is that the barrier is armed again afterwards, so a waiter arriving at step 3 blocks correctly instead of walking through a barrier everyone believes is up.

DistributedBarrier(client, "/barriers/import", leaseTtlSecs = 5L).use { barrier ->
  barrier.addConnectionStateListener { new, prev -> logger.warn { "Connection state: $prev -> $new" } }

  // A lease expiry means etcd already deleted the barrier key, so waiters saw a
  // spurious lift. The healer re-grants a lease and re-arms the barrier for
  // future waiters; the expiry itself lands on the exceptions list.
  barrier.addBackgroundExceptionListener { context, e ->
    logger.warn(e) { "Barrier lease trouble in $context" }
  }

  barrier.setBarrier()
  barrier.removeBarrier()
  barrier.exceptions.forEach { logger.warn(it) { "Background failure" } }
}

The scoped form closes the barrier on exit:

withDistributedBarrier(client, "/barriers/import") {
  setBarrier()
  removeBarrier()
}

DistributedBarrierWithCount¶

No gatekeeper: every party calls the same waitOnBarrier(), and the barrier trips when memberCount of them are parked on it.

// Every one of the five members runs this. Nobody proceeds until all five are
// parked here; then all five leave together.
DistributedBarrierWithCount(client, "/barriers/phase1", 5).use { barrier ->
  logger.info { "${barrier.waiterCount} of ${barrier.memberCount} waiting" }
  barrier.waitOnBarrier()
  logger.info { "All ${barrier.memberCount} arrived" }
}

// Every one of the five members runs this. Nobody proceeds until all five are
// parked here; then all five leave together.
try (DistributedBarrierWithCount barrier =
       new DistributedBarrierWithCount(client, "/barriers/phase1", 5)) {
  System.out.println(barrier.getWaiterCount() + " of " + barrier.getMemberCount() + " waiting");
  barrier.waitOnBarrier();
  System.out.println("All " + barrier.getMemberCount() + " arrived");
}

Each waiter registers a lease-bound key under <path>/waiting, and the first arrival CAS-creates <path>/ready. Every waiter watches the prefix; when the waiter count reaches memberCount, <path>/ready is deleted and everyone leaves together. waiterCount is a live count of the registered waiters — advisory, like every count read over a network, and useful mostly for logging a stuck rendezvous.

waitOnBarrier throws InterruptedException and EtcdRecipeException — the latter when the waiter's own key cannot be established, which means somebody else is already using that exact token and this wait can never be counted.

DistributedBarrierWithCount(client, "/barriers/phase1", 5).use { barrier ->
  // A timeout drops this member out of the count until it waits again, so a
  // straggler cannot be counted twice.
  if (!barrier.waitOnBarrier(30.seconds)) {
    logger.warn { "Only ${barrier.waiterCount} of ${barrier.memberCount} showed up" }
  }
}

try (DistributedBarrierWithCount barrier =
       new DistributedBarrierWithCount(client, "/barriers/phase1", 5)) {
  // A timeout drops this member out of the count until it waits again, so a
  // straggler cannot be counted twice.
  if (!barrier.waitOnBarrier(30, TimeUnit.SECONDS)) {
    System.out.println("Only " + barrier.getWaiterCount() + " of "
      + barrier.getMemberCount() + " showed up");
  }
}

withDistributedBarrierWithCount(client, "/barriers/phase1", 5) {
  waitOnBarrier()
}

DistributedDoubleBarrier¶

The rendezvous problem, twice. enter() blocks until all memberCount members have arrived; leave() blocks until all of them have finished. Between the two is a phase in which every member knows exactly who else is running.

// Two DistributedBarrierWithCount instances under the covers, at
// /barriers/phase1/enter and /barriers/phase1/leave.
DistributedDoubleBarrier(client, "/barriers/phase1", 5).use { barrier ->
  logger.info { "${barrier.enterWaiterCount} peers already at the gate" }

  // Nobody starts the phase until all five have arrived.
  barrier.enter()

  // ... run the phase ...

  // Nobody tears down until all five have finished it.
  barrier.leave()
  logger.info { "${barrier.leaveWaiterCount} peers still leaving" }
}

// Two DistributedBarrierWithCount instances under the covers, at
// /barriers/phase1/enter and /barriers/phase1/leave.
try (DistributedDoubleBarrier barrier =
       new DistributedDoubleBarrier(client, "/barriers/phase1", 5)) {
  System.out.println(barrier.getEnterWaiterCount() + " peers already at the gate");

  // Nobody starts the phase until all five have arrived.
  barrier.enter();

  // ... run the phase ...

  // Nobody tears down until all five have finished it.
  barrier.leave();
  System.out.println(barrier.getLeaveWaiterCount() + " peers still leaving");
}

It is a thin composition: two DistributedBarrierWithCount instances at <path>/enter and <path>/leave. enterWaiterCount and leaveWaiterCount are those barriers' waiterCounts. Everything the count barrier does — lease-bound waiting keys, self-healing while parked, cleanup on every exit path — it does here too.

The leave() half is the one people skip, and it is the one that makes the recipe worth using. Without it, the fastest member tears down its half of a shared fixture while the slowest is still reading from it.

withDistributedDoubleBarrier(client, "/barriers/phase1", 5) {
  if (enter(30.seconds)) {
    // Bound the leave too: a peer that dies mid-phase must not park the rest
    // of the cluster forever.
    if (!leave(30.seconds)) logger.warn { "Only $leaveWaiterCount left cleanly" }
  } else {
    logger.warn { "Only $enterWaiterCount entered; abandoning the phase" }
  }
}

try (DistributedDoubleBarrier barrier =
       new DistributedDoubleBarrier(client, "/barriers/phase1", 5)) {
  if (barrier.enter(30, TimeUnit.SECONDS)) {
    // Bound the leave too: a peer that dies mid-phase must not park the rest
    // of the cluster forever.
    if (!barrier.leave(30, TimeUnit.SECONDS)) {
      System.out.println("Only " + barrier.getLeaveWaiterCount() + " left cleanly");
    }
  } else {
    System.out.println("Only " + barrier.getEnterWaiterCount() + " entered");
  }
}

Choosing one¶

• Gate. A migration must finish before any worker starts consuming. One client arms a DistributedBarrier at startup and removes it when the migration commits. Workers wait. The party count is unknown and irrelevant; what matters is that one party decides.
• Rendezvous. A sharded job where every shard must have loaded its slice before any shard emits, because the output is only consistent across the whole set. DistributedBarrierWithCount(path, shardCount), one waitOnBarrier() per shard.
• Phase sync. The same sharded job, but the slices are also torn down at the end and nobody may tear down early. DistributedDoubleBarrier: enter(), run, leave().

If you find yourself reaching for a barrier to protect a critical section, you want locks instead.

Coroutines¶

Every wait has a suspending twin that releases the thread while parked, and that unblocks on cancellation: DistributedBarrier.await(), awaitSetBarrier(), awaitRemoveBarrier(), DistributedBarrierWithCount.await(), and DistributedDoubleBarrier.awaitEnter() / awaitLeave() — each with the Duration-bounded overload.

DistributedBarrier(client, "/barriers/import").use { barrier ->
  // Releases the thread while parked, and cancellation unblocks the wait.
  barrier.await(30.seconds)
}

DistributedDoubleBarrier(client, "/barriers/phase1", 5).use { barrier ->
  barrier.awaitEnter()
  barrier.awaitLeave()
}

Cancelling a coroutine parked in one of these removes the waiting key on the way out, so a cancelled member stops counting toward a DistributedBarrierWithCount immediately. See Coroutines.

Observability¶

Barrier waits are watch-driven, so what shows up with the Micrometer module wired in is the watch's own health — the etcd.watch.recovery counter — plus the connection-state transitions each recipe reports. Barrier paths never become tags; that would blow up cardinality. See Observability.