Caches¶

A cache turns a watch into a local read. Instead of asking etcd for a value every time you need it, you snapshot once, keep a watch open, and read from memory — so the hot path never blocks on a network round trip.

etcd-recipes ships three:

Recipe	Caches	Typed
`NodeCache<T>`	One key	Always — takes an `EtcdCodec<T>`
`PathChildrenCache`	Every child of a prefix	No — raw `ByteSequence`
`TypedPathChildrenCache<T>`	Every child of a prefix	Yes — decorates the above

NodeCache and PathChildrenCache extend EtcdConnector, so they share the lifecycle, exception, and connection-state surface described in Core concepts. TypedPathChildrenCache does not — see below.

Listeners run on a watch dispatcher¶

Read this before you write a listener.

Never block a cache listener on anything that needs etcd

Watcher callbacks originate on jetcd's Vert.x event loop. Anything that needs another gRPC response — a get, a put, a lock acquisition — or that contends for a lock the caller holds while that caller is waiting on gRPC, deadlocks the event loop. Not "slows it down": stops it, for every watch in the process, permanently.

The library's Client.watcher extension hops callbacks onto a dedicated single-thread executor, which buys back the simple cases. It does not make listeners safe in general: a listener that blocks on a mutex held by a thread that is itself parked on an etcd call still hangs the dispatcher, and once the dispatcher hangs the cache stops converging while continuing to look healthy.

Treat a listener as a notification, not a workflow. Copy what you need, hand it to your own executor, and return.

Each cache gets its own dispatcher thread, and it is single-threaded on purpose: that is what makes events arrive in etcd's order. A slow listener therefore delays every later event for that cache — and the watch-recovery loop, including its resync GET, runs on that same thread. Keep listeners short even when they cannot deadlock.

`NodeCache<T>`¶

One key, kept hot, decoded through an EtcdCodec<T>. This is the "watch my config value" recipe.

NodeCache(client, "/config/greeting", StringCodec).use { cache ->
  // start() snapshots the key, then anchors the watch at the snapshot revision + 1.
  cache.start()

  // Reads are local: no round trip, and no chance of blocking on a slow etcd.
  logger.info { "Current value: ${cache.current}" }
}

NodeCache<T> is generic and takes a codec, so it is presented in Kotlin throughout this section. It is callable from Java — see the Java guide — but the codec generics make it markedly less pleasant there; Java callers usually want PathChildrenCache.

Why `start()` snapshots first, then anchors the watch¶

start() is one-shot: it throws if called twice, or after close(). What it does in between matters more than it looks.

It reads the key, remembers the revision that read was answered at, and then opens the watch at snapshot revision + 1. That ordering is the whole recipe:

Watch first, snapshot second, and a PUT landing in between gets seen by the watch and then overwritten by the older snapshot. The cache silently serves a stale value forever.
Snapshot first, watch from "now", and a PUT landing in between is missed by both. Same outcome, different race.
Snapshot first, watch from the snapshot's revision + 1, and the stream picks up exactly where the snapshot left off — no gap, no overlap.

This is etcd's standard bootstrap, and the same fix is applied everywhere this library establishes a watch. It is also why compaction of the watched revision re-syncs transparently: the resync is the same snapshot-and-re-anchor operation.

Reading¶

current decodes on read and returns null when the key is absent. currentBytes hands back the raw ByteSequence and skips the codec.

// Any EtcdCodec<T> works: StringCodec, ByteSequenceCodec, jsonCodec<T>(), or the
// Jackson codec from etcd-recipes-jackson.
NodeCache(client, "/config/flags", jsonCodec<FeatureFlags>()).use { cache ->
  cache.start()

  // current decodes on read, so a malformed payload throws here, at your call site.
  val flags: FeatureFlags? = cache.current
  logger.info { "Max batch size: ${flags?.maxBatchSize}" }

  // currentBytes hands back the raw value and skips the codec entirely.
  logger.info { "Raw: ${cache.currentBytes}" }
}

current decodes at your call site

Decoding lazily means a malformed payload throws where you read it, not on the dispatcher thread where nobody is listening. Events are the other way round: a payload that fails to decode in the listener path is recorded on exceptions and the event is skipped, because killing the dispatcher over one bad value would stop the cache updating at all.

Listening¶

NodeCache(client, "/config/greeting", StringCodec).use { cache ->
  // Register listeners before start(): the watch begins delivering the moment
  // start() returns, and nothing replays what you were not listening for.
  cache.addListener { event: NodeCacheEvent<String> ->
    when (event.type) {
      NodeCacheEvent.Type.CREATED -> logger.info { "Created: ${event.value}" }

      NodeCacheEvent.Type.UPDATED -> logger.info { "Updated: ${event.value}" }

      // value is null on DELETED: the key is gone.
      NodeCacheEvent.Type.DELETED -> logger.info { "Deleted" }
    }
  }
  cache.start()
}

NodeCacheEvent<T> carries a type — CREATED, UPDATED, or DELETED — and the decoded value, which is null on DELETED. CREATED means "the cache had no value and now does", which is a statement about the cache, not about etcd: a key that already existed when you called start() produces UPDATED on its next change, because the snapshot already put a value in hand.

Register listeners before start(). The watch begins delivering the moment start() returns, and nothing replays what you were not listening for.

Watch recovery¶

NodeCache(client, "/config/greeting", StringCodec).use { cache ->
  cache.addRecoveryListener { event: WatchRecoveryEvent ->
    when (event) {
      is WatchRecoveryEvent.Suspended -> logger.warn { "Watch stream died; retrying" }

      is WatchRecoveryEvent.Resubscribed -> logger.info { "Watch re-established" }

      is WatchRecoveryEvent.Resynced -> logger.info { "Compaction: re-snapshotted the key" }

      // Recovery abandoned: current is frozen at whatever it last saw.
      is WatchRecoveryEvent.Failed -> logger.error { "Watch abandoned; value is stale" }
    }
  }
  cache.start()
}

A WatchRecoveryEvent.Failed is the one that matters: recovery has been abandoned, so current is frozen at whatever it last saw and will never move again. It is also recorded on exceptions and drives connectionState to LOST. See Resilience.

Scoped use¶

val greeting: String? =
  withNodeCache(client, "/config/greeting", StringCodec) {
    start()
    current
  }
logger.info { "Greeting: $greeting" }

`PathChildrenCache`¶

A whole prefix. Every child of cachePath is held in memory and kept current by one prefix watch.

KotlinJava

PathChildrenCache(client, "/cache/workers").use { cache ->
  // buildInitial = true snapshots the prefix before the watch starts, so the cache
  // is already populated by the time start() returns.
  cache.start(buildInitial = true)

  cache.currentData.forEach { child: ChildData ->
    logger.info { "${child.key} -> ${child.value.asString}" }
  }
}

try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  // true = snapshot the prefix before the watch starts, so the cache is already
  // populated by the time start() returns.
  cache.start(true);

  for (ChildData child : cache.getCurrentData()) {
    System.out.println(child.getKey() + " -> " + child.getValue().toString(StandardCharsets.UTF_8));
  }
}

Start modes¶

StartMode decides what the cache knows when start() returns:

Mode	Snapshot	`INITIALIZED` event
`NORMAL`	No — starts empty	No
`BUILD_INITIAL_CACHE`	Yes	No
`POST_INITIALIZED_EVENT`	Yes	Yes, carrying the snapshot

KotlinJava

// NORMAL: no snapshot. The cache starts empty and fills only from live events.
PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.start(PathChildrenCache.StartMode.NORMAL)
}

// BUILD_INITIAL_CACHE: snapshot first, then watch from the snapshot revision + 1.
PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE)
}

// POST_INITIALIZED_EVENT: BUILD_INITIAL_CACHE, plus an INITIALIZED event carrying
// the snapshot in event.initialData.
PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT)
}

// NORMAL: no snapshot. The cache starts empty and fills only from live events.
try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  cache.start(PathChildrenCache.StartMode.NORMAL);
}

// BUILD_INITIAL_CACHE: snapshot first, then watch from the snapshot revision + 1.
try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
}

// POST_INITIALIZED_EVENT: BUILD_INITIAL_CACHE, plus an INITIALIZED event carrying
// the snapshot in getInitialData().
try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
}

There are two start() overloads. start(buildInitial, waitOnStartComplete) is the convenience form — buildInitial = true maps to BUILD_INITIAL_CACHE, false to NORMAL — and start(mode, waitOnStartComplete) is the one that can reach POST_INITIALIZED_EVENT. Both default waitOnStartComplete to true.

NORMAL does not start with what is already there

NORMAL skips the snapshot entirely and watches from now. Children that existed before you started are invisible until something touches them. That is occasionally what you want (you only care about changes), and almost never what people mean when they reach for a cache. Prefer BUILD_INITIAL_CACHE.

Waiting for the snapshot¶

In the two priming modes the snapshot loads on a background thread. waitOnStartComplete = true (the default) makes start() block until it has finished, so currentData is populated when you get control back. Pass false when start() must not block your caller, then bound the wait yourself:

KotlinJava

PathChildrenCache(client, "/cache/workers").use { cache ->
  // waitOnStartComplete = false returns before the snapshot has loaded, so the
  // cache may still be empty. Useful when start() must not block your caller.
  cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE, waitOnStartComplete = false)

  // ...then bound the wait yourself. Returns false if the snapshot is still loading.
  if (cache.waitOnStartComplete(10.seconds))
    logger.info { "Primed with ${cache.currentData.size} children" }
  else
    logger.warn { "Snapshot still loading after 10s" }
}

try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  // false = return before the snapshot has loaded, so the cache may still be empty.
  cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE, false);

  // ...then bound the wait yourself. Java uses the (long, TimeUnit) overload.
  if (cache.waitOnStartComplete(10, TimeUnit.SECONDS)) {
    System.out.println("Primed with " + cache.getCurrentData().size() + " children");
  } else {
    System.out.println("Snapshot still loading after 10s");
  }
}

waitOnStartComplete() with no argument waits indefinitely; the Duration and (long, TimeUnit) overloads bound it and return false on timeout. All three throw InterruptedException.

Reading¶

KotlinJava

PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.start(buildInitial = true)

  // Sorted by child name, so iteration order is stable across clients.
  cache.currentData.forEach { logger.info { "${it.key} -> ${it.value.asString}" } }

  // getCurrentData() takes the child name RELATIVE to the cache path, not a full
  // path: "/cache/workers/w1" returns null, "w1" is what you want.
  val w1: ByteSequence? = cache.getCurrentData("w1")
  logger.info { "w1 -> ${w1?.asString}" }

  // A point-in-time copy of the whole prefix.
  val snapshot: Map<String, ByteSequence> = cache.currentDataAsMap
  logger.info { "Holding ${snapshot.size} children" }
}

try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  cache.start(true);

  // Sorted by child name, so iteration order is stable across clients.
  List<ChildData> children = cache.getCurrentData();
  System.out.println("Holding " + children.size() + " children");

  // The child name RELATIVE to the cache path, not a full path:
  // "/cache/workers/w1" returns null, "w1" is what you want.
  ByteSequence w1 = cache.getCurrentData("w1");
  System.out.println("w1 -> " + (w1 == null ? "absent" : w1.toString(StandardCharsets.UTF_8)));

  // A point-in-time copy of the whole prefix.
  Map<String, ByteSequence> snapshot = cache.getCurrentDataAsMap();
  System.out.println("Snapshot size: " + snapshot.size());
}

currentData returns List<ChildData> sorted by child name, so iteration order is stable across clients and across restarts. currentDataAsMap is an unsorted point-in-time copy.

getCurrentData() takes a child name, not a path

The keys are relative to cachePath: "w1", not "/cache/workers/w1". Passing a full path returns null — silently, because a missing child is a legitimate answer. This trips people up exactly once.

Listening¶

KotlinJava

PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.addListener { event: PathChildrenCacheEvent ->
    when (event.type) {
      PathChildrenCacheEvent.Type.CHILD_ADDED -> logger.info { "Added ${event.childName}" }

      PathChildrenCacheEvent.Type.CHILD_UPDATED -> logger.info { "Updated ${event.childName}" }

      // data carries the value the child held immediately before removal.
      PathChildrenCacheEvent.Type.CHILD_REMOVED -> logger.info { "Removed ${event.childName}" }

      // Only INITIALIZED events populate initialData; it is empty on the others.
      PathChildrenCacheEvent.Type.INITIALIZED -> logger.info { "Primed: ${event.initialData.size}" }
    }
  }
  cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT)
}

try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  cache.addListener(event -> {
    // A switch expression rather than a statement: the compiler enforces that
    // every Type is handled, so a new event type becomes a compile error here
    // instead of an event this listener silently drops.
    String message = switch (event.getType()) {
      case CHILD_ADDED -> "Added " + event.getChildName();
      case CHILD_UPDATED -> "Updated " + event.getChildName();
      // getData() carries the value the child held immediately before removal.
      case CHILD_REMOVED -> "Removed " + event.getChildName();
      // Only INITIALIZED events populate getInitialData().
      case INITIALIZED -> "Primed: " + event.getInitialData().size();
    };
    System.out.println(message);
  });
  cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
}

PathChildrenCacheEvent has a childName, a type — CHILD_ADDED, CHILD_UPDATED, CHILD_REMOVED, INITIALIZED — and data. On CHILD_REMOVED, data is the value the child held immediately before removal, which is usually what you need to clean up after it. initialData is populated only on INITIALIZED and is empty on every other event.

`rebuild()` and `clear()`¶

KotlinJava

PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.start(buildInitial = true)

  // A coarse manual re-sync against etcd's current children. The watcher already
  // keeps the cache converged, so this is a repair tool, not part of steady state:
  // a live watch event on the same key can race the snapshot, last writer wins.
  cache.rebuild()

  // Drops every entry locally without touching etcd. The next event or rebuild()
  // refills it; until then currentData reports nothing.
  cache.clear()
}

try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/workers")) {
  cache.start(true);

  // A coarse manual re-sync against etcd's current children. The watcher already
  // keeps the cache converged, so this is a repair tool, not part of steady state.
  cache.rebuild();

  // Drops every entry locally without touching etcd.
  cache.clear();
}

rebuild() re-snapshots the prefix and reconciles the live map in place — it does not clear first, so currentData never reports an empty window mid-rebuild. clear() drops every entry locally without touching etcd.

rebuild() is a repair tool, not a refresh

The watcher already keeps the cache converged, including across compaction. A manual rebuild() races the live watch on any key that changes during the snapshot, and last writer wins — so it can momentarily un-apply an event you already saw. Rely on the watcher for ordering; reach for rebuild() when you have reason to believe the cache drifted.

Watch recovery and resilience¶

PathChildrenCache(client, "/cache/workers").use { cache ->
  cache.addRecoveryListener { event: WatchRecoveryEvent ->
    when (event) {
      is WatchRecoveryEvent.Suspended -> logger.warn { "Watch stream died; retrying" }

      is WatchRecoveryEvent.Resubscribed -> logger.info { "Watch re-established" }

      is WatchRecoveryEvent.Resynced -> logger.info { "Compaction: cache re-synced" }

      // Recovery abandoned: the cache is frozen and will never update again.
      is WatchRecoveryEvent.Failed -> logger.error { "Watch abandoned; cache is stale" }
    }
  }
  cache.start(buildInitial = true)
}

addRecoveryListener reports resubscribes after fatal stream deaths, compaction resyncs, and abandonment. Failed means the cache is now frozen: it will keep answering reads with data that no longer reflects etcd. That is the failure mode worth alerting on, and it is why the event is also recorded on exceptions and moves connectionState to LOST. See Resilience and Observability.

Scoped use¶

val names: List<String> =
  withPathChildrenCache(client, "/cache/workers") {
    start(buildInitial = true)
    currentData.map { it.key }
  }
logger.info { "Workers: $names" }

withPathChildrenCache does not expose resilience

Its parameters are (client, cachePath, userExecutor, receiver) — there is no resilience argument on this overload. Construct PathChildrenCache directly when you need a non-default ResilienceConfig:

// withPathChildrenCache does not expose resilience; construct directly for that.
PathChildrenCache(
  client = client,
  cachePath = "/cache/workers",
  userExecutor = null,
  resilience = ResilienceConfig.DEFAULT,
).use { cache ->
  cache.start(buildInitial = true)
  logger.info { "Children: ${cache.currentData.size}" }
}

`TypedPathChildrenCache<T>`¶

The same prefix cache with every child value decoded through an EtcdCodec<T>: currentData yields TypedChildData<T>, getCurrentData() yields T?, and currentDataAsMap yields Map<String, T>.

TypedPathChildrenCache(client, "/cache/workers", jsonCodec<Worker>()).use { cache ->
  cache.start(buildInitial = true)

  cache.currentData.forEach { child: TypedChildData<Worker> ->
    logger.info { "${child.key} -> ${child.value.name} (load ${child.value.load})" }
  }

  // Same relative-child-name rule as the untyped cache, decoded on the way out.
  val w1: Worker? = cache.getCurrentData("w1")
  logger.info { "w1 -> $w1" }
}

See Typed values for the codec options.

It is a decorator, not a connector¶

TypedPathChildrenCache<T> implements Closeable. It does not extend EtcdConnector. That means exceptions, hasExceptions, isHealthy(), connectionState, ping(), and the connection-state listeners are not on it.

They are all on untyped, which is a public property and the documented escape hatch:

TypedPathChildrenCache(client, "/cache/workers", jsonCodec<Worker>()).use { cache ->
  cache.start(buildInitial = true)

  // TypedPathChildrenCache is a Closeable decorator, NOT an EtcdConnector. The
  // connector surface lives on `untyped`, which is the documented escape hatch.
  val untyped: PathChildrenCache = cache.untyped

  // A payload that fails to decode is recorded here and its event skipped, rather
  // than killing the watch dispatcher. Nothing else tells you it happened.
  untyped.exceptions.forEach { logger.warn(it) { "Cache background failure" } }
  logger.info { "Healthy: ${untyped.isHealthy()}, state: ${untyped.connectionState}" }

  // Raw, undecoded views of the same cache are reachable through it too.
  logger.info { "Raw children: ${untyped.currentDataAsMap.keys}" }
}

This matters more than the usual "there's an escape hatch if you need it", because of how decode failures surface. A payload the codec cannot read is caught in the re-emit path, recorded on untyped.exceptions, and its event is skipped. The typed listener simply never fires for that child. If you are not reading untyped.exceptions, a poison value in your prefix looks exactly like a child that never changed.

Listening¶

TypedPathChildrenCache(client, "/cache/workers", jsonCodec<Worker>()).use { cache ->
  // The event is a TypedPathChildrenCacheEvent<Worker>, but its `type` is the
  // untyped PathChildrenCacheEvent.Type: there is no separate typed enum.
  cache.addListener { event ->
    when (event.type) {
      PathChildrenCacheEvent.Type.CHILD_ADDED -> logger.info { "Added ${event.data?.name}" }
      PathChildrenCacheEvent.Type.CHILD_UPDATED -> logger.info { "Updated ${event.data?.name}" }
      PathChildrenCacheEvent.Type.CHILD_REMOVED -> logger.info { "Removed ${event.childName}" }
      PathChildrenCacheEvent.Type.INITIALIZED -> logger.info { "Primed: ${event.initialData.size}" }
    }
  }
  cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT)
}

TypedPathChildrenCacheEvent<T> reuses the untyped Type enum

There is no TypedPathChildrenCacheEvent.Type. The event's type is a PathChildrenCacheEvent.Type — the same CHILD_ADDED / CHILD_UPDATED / CHILD_REMOVED / INITIALIZED values. Only the payload is typed: data is T? and initialData is List<TypedChildData<T>>.

Scoped use¶

val loads: Map<String, Int> =
  withTypedPathChildrenCache(client, "/cache/workers", jsonCodec<Worker>()) {
    start(buildInitial = true)
    currentDataAsMap.mapValues { (_, worker) -> worker.load }
  }
logger.info { "Loads: $loads" }

Unlike its untyped counterpart, withTypedPathChildrenCache does take a resilience argument, after userExecutor.

Coroutines¶

All three caches expose their events as flows — eventsAsFlow() and recoveryEventsAsFlow() — which is the better answer to the dispatcher problem at the top of this page. A flow buffers (unlimited by default), so a slow collector cannot stall the watch dispatcher the way a slow listener does, and collection happens on your dispatcher rather than the cache's. Collecting registers a listener and cancelling removes it; it never starts or closes the cache, so you still own the lifecycle. See Flows.

Observability¶

With the Micrometer module wired in, every snapshot load — the initial prime, each rebuild(), and each compaction resync — reports an etcd.cache.sync timer and an etcd.cache.size distribution summary. Cache paths never become tags, because prefix cardinality is unbounded. See Observability.

A steadily climbing etcd.cache.sync count on an idle cache means something is resyncing it repeatedly — usually compaction chasing a watch that keeps falling behind.

Caches¶

Listeners run on a watch dispatcher¶

NodeCache<T>¶

Why start() snapshots first, then anchors the watch¶

Reading¶

Listening¶

Watch recovery¶

Scoped use¶

PathChildrenCache¶

Start modes¶

Waiting for the snapshot¶

Reading¶

Listening¶

rebuild() and clear()¶

Watch recovery and resilience¶

Scoped use¶

TypedPathChildrenCache<T>¶

It is a decorator, not a connector¶

Listening¶

Scoped use¶

Coroutines¶

Observability¶

`NodeCache<T>`¶

Why `start()` snapshots first, then anchors the watch¶

`PathChildrenCache`¶

`rebuild()` and `clear()`¶

`TypedPathChildrenCache<T>`¶