com.worxbend.cobalt

The admin surface's gate: an Authorization header in, a Principal or an AdminReply out.

Separate from JwtVerifier because they answer different questions. The verifier answers "is this token genuine"; this answers "what does cobalt's HTTP surface do about the answer" — which status code, which body, and which WWW-Authenticate challenge. Keeping the second out of the first is what lets the verifier be tested with no notion of HTTP and lets this be tested with no notion of cryptography.

Attributes

Companion: object
Supertypes: class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: AdminAuth.type

The operational answers, computed without a socket.

Cask is cobalt's HTTP surface and nothing more (ADR §1). There are no business endpoints here and there never will be: events arrive over Kafka, and an HTTP write path would be a second, unordered, uncommitted way into the same database.

Splitting the answers from the annotated route class is what makes them testable at all — Cask has no test kit, so the alternative is binding a port in a unit test. Here the routes are three one-line delegations and every decision lives in a pure method.

Attributes

Supertypes: class Object

trait Matchable

class Any

Reads committed and log-end offsets for one consumer group.

Split from ConsumerLagGauge so the gauge can be exercised with hand-written maps, and split from the scheduler so a failed admin call is a returned failure rather than an exception on a timer thread nobody is watching.

Attributes

Supertypes: class Object

trait Matchable

class Any

A status code, a content type, a body and any extra headers — the whole of cobalt's HTTP contract.

headers is empty for every answer a handler produces and non-empty for exactly one thing: the WWW-Authenticate challenge RFC 6750 requires on a 401. Carrying it on the reply rather than setting it at the route is what keeps AdminAuth testable without Undertow.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Supertypes: class Object

trait Matchable

class Any
Self type: AdminRoutes.type

What a route does to the pipeline, which is what decides the scope it demands.

Two, not one per route. The split that matters is between looking and acting: GET /admin/dlq/records returns event payloads, which is a disclosure; POST /admin/consumer:restart?target=latest permanently skips unconsumed events, which is destruction. An on-call engineer who only needs to read the DLQ during triage should not carry a credential that can also empty the pipeline, and that is the only distinction a token can usefully carry.

A token holding the write scope also satisfies Read. The mutating routes already return the state the read routes return — :restart answers with the full consumer status, :replay with the record identities it acted on — so refusing the GET while permitting the POST would be a distinction with no security content and one that produces an inexplicable 403 in the middle of an incident.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

cobalt's HTTP listener: Cask's routing, Undertow's lifecycle, owned explicitly.

Undertow is built here rather than by cask.main.Main.main. Cask's own main binds the port, registers a JVM shutdown hook and returns nothing — which costs two things this service needs. First, the bound port is unreachable, so an integration test cannot bind port 0 and then talk to it; it has to guess a free port and race every other suite for it. Second, the listener's shutdown belongs in CoordinatedShutdown alongside the consumer drain and the pool close, not in a second, independently-ordered JVM hook that may run before or after them.

Cask still does all the routing: cask.main.Main.defaultHandler is its dispatch trie, and this class only decides where it is bound and when it stops.

Attributes

Supertypes: trait StrictLogging

class Main

class Object

trait Matchable

class Any

How this deployment verifies bearer tokens on the admin surface.

enabled defaults to true and there is no key default, matching wolfram: a security layer whose default state is "off" ships off, because nothing fails when it is. Running cobalt's admin API unauthenticated therefore has to be said out loud as AUTH_ENABLED=false, and running it authenticated requires a key — a service configured to verify but given nothing to verify against refuses to boot rather than accepting everything.

Value parameters

algorithm: HS256/HS384/HS512 (symmetric, needs secret) or RS256/RS384/RS512 (asymmetric, needs public-key). Pinned here; the token's own alg header is compared against it and never trusted.
leeway: tolerance applied to exp and nbf, for clock skew between the issuer and this host. Small on purpose: this is the window in which a revoked token still works.
readScope: the scope a token must carry to call a GET /admin/… route. Empty leaves signature verification on and drops the scope check, for a deployment whose issuer mints no scopes — which is a worse posture than scopes and a far better one than AUTH_ENABLED=false, so it is available and spelled out rather than forbidden.
writeScope: the scope a token must carry to call a POST /admin/… route — the ones that move offsets, republish onto the production topic, or destroy the externalised checkpoints.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Why a request was refused before it reached a handler.

Two cases and not one, because they map to different status codes and to different fixes: 401 says "I do not know who you are", 403 says "I know exactly who you are and the answer is still no". Collapsing them sends an operator with an expired token into a permissions investigation.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all
Known subtypes: class Unauthenticated

class Forbidden

Turns one groupedWithin batch into a durable effect, and returns the offsets that effect earned.

This class is the at-least-once guarantee. It returns Committables, and it returns them only on the paths where the batch is genuinely accounted for — every record either written to PostgreSQL, or published to the DLQ. When neither is true the returned Future fails, the stream fails with it, and the offsets are never handed to the committer. ConsumerStream then puts Committer.flow strictly downstream of this stage, so an offset is only ever a receipt for work that already happened.

Poison isolation is a bisection, not a retry loop. ADR §4.3 requires that one bad record cannot wedge the stream. The naïve fix — dead-letter the whole batch on failure — discards up to batchSize perfectly good events for one malformed one. So a failed batch is retried whole (a database blip is not a data problem), then split in half and each half written independently, recursively, until the failure is attributed to a single record. That record is dead-lettered and the batch proceeds. log₂(500) ≈ 9 extra round trips is a cheap price for never losing a good event and never stalling.

The classification in BatchProcessor.isDataError is what keeps the bisection safe. Without it, a database that is merely down would bisect to singletons and dead-letter the entire batch — converting a recoverable outage into permanent data loss, which is strictly worse than stalling. Only a failure the database will deterministically repeat (SQLSTATE class 22, data exception; class 23, integrity constraint) is allowed to dead-letter a record. Everything else is rethrown so the RestartSource backs off and tries again with the offsets uncommitted.

The externalised checkpoint and the Kafka commit always describe the same position, and that is the invariant BatchProcessor.Accounted exists to hold. process returns a Committable for every record in the batch, decoded or not, so the checkpoint has to account for every one of them too. Deriving it from the writable records alone — which is what a Vector[PendingWrite] invites — silently drops the offsets of dead letters, and the two consequences are both quiet: consume.checkpoint.divergence, whose only healthy value is zero, reports a gap that means nothing and hides the one that does; and restart?target=stored rewinds the group onto records it has already dead-lettered, re-consuming and re-dead-lettering them on every recovery. The offsets are therefore computed once, in process, from the whole batch, and insert is the single place they can reach the database.

No span is opened around the write. A batch aggregates records from many unrelated traces, so it cannot be a child of any one of them, and a span with batchSize links is not something any backend renders usefully. The per-record CONSUMER span in RecordDecoder is where trace continuation lives; batch health is a metric (ConsumerMetrics.batchWrite), which is the shape that question actually has.

Value parameters

backoff: the pause between whole-batch attempts, as a function so tests do not have to wait. Short on purpose: the real backoff for a sustained outage is the consumer's RestartSource, which also unwinds the Kafka session.

Attributes

Companion: object
Supertypes: trait StrictLogging

class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: BatchProcessor.type

cobalt's admin API, described once and rendered as OpenAPI.

Why tapir endpoint values for a Cask service

cobalt's HTTP surface is Cask and stays Cask (ADR §1) — annotations on a routes class, no second web stack, no server interpreter on this classpath. What tapir contributes here is only the description: a set of endpoint values that a generator turns into a document with real schemas.

The alternative was a hand-written openapi.yaml. That is a second statement of the contract, and the two disagree the first time a parameter is added — silently, because nothing compiles YAML. This arrangement has the same hazard in a smaller and checkable form: the description could drift from the Cask routes. So CobaltApiDocsSuite asserts, in both directions, that every path here is served and every route served is described. That test is the reason this design is acceptable rather than merely convenient.

Why the admin API is documented at all

Two of these routes move a production pipeline: one replays dead letters onto the main topic, the other moves a consumer group's committed offsets. An operator meets them during an incident, from a terminal, having never used them before. A generated document with every parameter's meaning and every failure's cause is the difference between that and reading Scala.

Attributes

Supertypes: class Object

trait Matchable

class Any
Self type: CobaltApiDocs.type

Attributes

Companion: class
Supertypes: trait StrictLogging

class Object

trait Matchable

class Any
Self type: CobaltApp.type

Everything cobalt owns, constructed once and torn down in the reverse order.

A composition root and not a container. The graph is a dozen objects with a construction order that is itself load-bearing — telemetry before anything that meters, migrations before the repository that assumes the schema, the consumer last because it starts doing work the moment it exists. Guice would hide exactly that ordering, and cobalt has no other use for it.

Teardown order is the interesting half, and it is expressed as CoordinatedShutdown phases rather than as a close() sequence. A SIGTERM must:

unbind the admin listener, so an orchestrator's readiness probe starts failing and traffic stops being routed here (PhaseServiceUnbind);
drain the Kafka consumer — finish the in-flight batch, write it, commit it (PhaseServiceRequestsDone, registered by EventConsumer);
only then close the DLQ producer, the admin client, the connection pools and finally telemetry (PhaseBeforeActorSystemTerminate).

Closing telemetry first is the common mistake, and it makes the drain — the one part of shutdown that can lose data — the one part with no metrics and no spans.

Attributes

Companion: object
Supertypes: class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: CobaltConfig.type

cobalt's whole configuration, read once by the composition root.

One aggregate rather than four independent lookups, so a typo in any namespace refuses the boot rather than surfacing the first time that subsystem is exercised — for a consumer, "the first time" can be hours after deploy.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The Cask route table: a one-line delegation to AdminHandlers per route, through one of two doors.

Every route under /admin is authenticated and scoped; the three platform routes are not. Which door a route uses is declared in AdminRoutes.Access and checked from both ends — see that table for what is open and why, and for the two tests that make it impossible for a new route to be added without an answer.

The paths are string literals because Cask's annotations are macros and want a constant; AdminRoutesSuite asserts each one against the corresponding constant in modules/observability or AdminRoutes, so a divergence between this file and the shared vocabulary fails a test rather than silently giving Prometheus a 404.

Every route that changes anything is a @cask.post, because none of them is safe or repeatable-without-effect — the replay appends records to the topic on each call, and the lifecycle routes move a consumer group. dryRun defaults to true on the two destructive ones so the default POST is the one that publishes and moves nothing. Everything the operation needs arrives as query parameters rather than a JSON body: the request has four scalar fields, a body parser would mean either upickle — a second JSON stack alongside circe, which ADR §3.3 keeps off the classpath — or hand-rolled parsing, and a query string is what someone can actually type into curl mid-incident.

Attributes

Companion: object
Supertypes: trait Routes

class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: CobaltRoutes.type

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ConsumerConfig.type

How the stream batches, retries, and gives up.

Every field here is a correctness knob rather than a tuning one, and each is spelled out because the default that would apply in its absence is wrong for this pipeline.

Value parameters

batchSize: the groupedWithin element bound of ADR §4.3. It is also the largest number of rows one failed insert can force the isolation search to bisect, so raising it trades steady-state throughput against poison-pill recovery time.
batchWindow: the groupedWithin time bound. This is the latency floor of the whole pipeline when traffic is sparse: with one event per minute, an event is durable batchWindow after it arrives and not before.
commitMaxBatch: how many offsets Committer.flow aggregates into one commitAsync. Committing per record is a round trip per event; committing too rarely widens the replay window after a crash.
properties: free-form consumer overrides (security, DNS, tuning). bootstrap.servers and group.id are set from the typed fields above and always win, so a stray property cannot silently point the consumer at another cluster.
retryDelay: pause between those attempts. Deliberately short — the real backoff is RestartConfig, which unwinds the whole consumer including its Kafka session.
writeAttempts: how many times a whole batch is retried before the isolation search starts bisecting it. Retries exist to ride out a connection blip; bisecting exists to find a row the database will never accept. Conflating the two is what turns a five-second database outage into a DLQ full of perfectly good events.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Makes a fresh stream. Called once per start or resume; a stream is not restartable once drained.

Attributes

Supertypes: class Object

trait Matchable

class Any

A running stream, as the supervisor sees it. One method to stop it, one future to watch.

An interface and not EventConsumer directly, so the supervisor's state machine can be tested against a handle that needs no broker — which is the only way the pause/resume/failure transitions get covered at all.

Attributes

Supertypes: class Object

trait Matchable

class Any
Known subtypes: class EventConsumer

The lag arithmetic, separated from every I/O that feeds it.

Lag is the single most important number this service publishes and it is trivially easy to get subtly wrong — off by one, negative during a reset, or fabricated for a partition nobody has ever committed to. Keeping it as a pure function over two maps is what makes those cases testable without a broker.

Attributes

Supertypes: class Object

trait Matchable

class Any
Self type: ConsumerLag.type

Publishes ConsumerLag as a Micrometer MultiGauge, fed from a single reused AdminClient.

An AdminClient and not the consumer's own records-lag-max (ADR §7.1). The client metric only covers partitions the consumer is currently fetching: it reads zero during a rebalance and disappears entirely when the process is down — the two moments at which lag is the only number anyone wants. Asking the broker instead means the measurement survives the thing it is measuring.

One AdminClient, reused. Each one opens its own connections and its own metrics; constructing one per poll turns a monitoring signal into a connection-churn problem on the broker.

MultiGauge.register(rows, overwrite = true) is what makes a partition that stops being assigned disappear from the exposition rather than freezing at its last value — a frozen gauge is indistinguishable from a stuck consumer.

Attributes

Supertypes: class Object

trait Matchable

class Any

cobalt's domain metrics, expressed in the shared vocabulary of modules/observability.

No meter name or tag key is invented here. ADR §7.1 makes a single taxonomy across three services the enforceable rule, because one Grafana dashboard has to work everywhere. This class is a typed façade over Meters whose value is that each meter's tag set is fixed in exactly one place — Micrometer happily registers the same name twice with different tags, and Prometheus then renders two unrelated series.

persisted and duplicate deliberately do not partition the batch. The idempotent insert reports one number, "rows the database actually wrote", and there is no way to attribute the shortfall to a particular type from a single ON CONFLICT DO NOTHING batch. So persisted counts records the database took responsibility for — which is every record in a successful batch, new or already present — tagged by type, and duplicates separately counts the shortfall untagged. Inventing a per-type split would produce a number that looks precise and is not.

-Werror trap (ADR §7.4). Every Micrometer builder returns this and MeterRegistry#counter returns the meter, so a bare registration line trips -Wnonunit-statement. Every call below is part of an expression.

Attributes

Supertypes: class Object

trait Matchable

class Any

The consumer's whole observable state, in one response.

One object rather than four endpoints because these values are only meaningful together: a lag of 40 000 means something different when state is paused than when it is running, and an operator who has to make two requests to find that out will make them minutes apart during an incident.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ConsumerStatus.type

The consumer graph of ADR §4.3, assembled from pieces each of which is testable without a broker.

committableSource ─► decode ─► groupedWithin ─► mapAsync(1) write ─► mapConcat ─► Committer.flow ─► Sink.ignore

The ordering of the last two stages is the whole at-least-once guarantee, and it is the reason this is a Flow and not a fold. An offset is a receipt for a durable effect. Committer.flow sits strictly downstream of the write, so an offset physically cannot reach the committer until BatchProcessor has already made its batch durable — either in PostgreSQL or in the DLQ. Put the committer upstream (or run it in parallel, or commit inside the write stage's andThen) and a crash in the window between the two loses every event in flight, silently, with the consumer group reporting zero lag. At-least-once redelivery plus the (occurred_at, ce_source, ce_id) unique index then makes the pipeline observationally exactly-once at the database.

mapAsync(1) and not a larger parallelism. Offsets must reach the committer monotonically per partition; mapAsync(n) preserves emission order but starts n batches concurrently, so two batches from one partition would be writing at the same time and a failure in the older one would already have had its successor's offset committed past it. One in flight is also enough: the batch is 500 rows in a single executeBatch, so the database, not the stream, is the bottleneck.

The value deserializer is ByteArrayDeserializer, never CloudEventDeserializer (ADR §4.3). Decoding happens in the decode stage, where a failure is an ordinary Left rather than an exception thrown inside KafkaConsumer.poll — which would kill the stream before the connector ever saw the record and make every restart replay the same poison pill forever.

Attributes

Supertypes: class Object

trait Matchable

class Any
Self type: ConsumerStream.type

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: ConsumerSupervisor.type

The consumer's lifecycle and observable state, behind one object.

Why a supervisor at all

Before this, the consumer was a stream started once at boot and drained once at shutdown, and the only way to stop it was to stop the process. That is a poor answer to every real incident: a poison partition that needs skipping, a database under maintenance, a backfill that must not be consumed while it runs, a group whose offsets are wrong. All four end in docker compose restart plus a kafka-consumer-groups.sh invocation nobody has memorised, on a container that has to be given a shell.

The state machine, and what "pause" actually does

Pause tears the stream down; it does not idle it. Pekko Connectors' Consumer.Control has no pause — it can stop the fetcher, drain, or shut down, and nothing in between. Two implementations were available:

hold the stream open and stop demand downstream, which stops committing but leaves the group's session alive and its partitions assigned. The broker keeps them assigned, so no other replica can take over, and max.poll.interval.ms eventually expires and forces the rebalance anyway — after a delay, unpredictably.
drain and stop, which commits everything in flight, leaves the group cleanly, and lets the partitions move to another replica immediately.

The second is what an operator means by "pause this consumer", so that is what it does. resume materialises a fresh stream, which rejoins the group and continues from the committed offsets. The cost is that a pause/resume cycle is a rebalance; the benefit is that a paused replica is genuinely out of the way.

Why one mutex and not an actor

Every transition here is a slow, blocking, at-most-one-at-a-time operation: drain the stream, wait for the group to leave, alter offsets, materialise a new stream. An actor would serialise them and add a mailbox, a protocol and a set of ask timeouts to reason about. A single lock over an AtomicReference serialises them too, in eleven lines, and the resulting code says plainly that a transition holds the consumer for its duration — which is the property that matters and the one an ask-timeout would obscure.

What this does not do

It does not restart the stream on failure — ConsumerStream.restarting still owns that, with backoff. The supervisor observes the outcome so it can report failed with the cause once the restart policy gives up. Two restart mechanisms would fight.

Attributes

Companion: object
Supertypes: trait StrictLogging

class Object

trait Matchable

class Any

The DLQ's operator surface: look at it, and put things back.

Why this exists. Until it did, docs/operations.md told an operator to "re-publish the original events through POST /events on wolfram" — an instruction that assumes the events are still somewhere other than the DLQ, that their bytes can be reconstructed by hand, and that there is time to do it. Kafka retention on the DLQ topic is seven days, so the real situation is a seven-day window in which the only tool is a console consumer. Every decision below follows from that: an operator under time pressure needs to see what is there, see what a replay would do, and then do it, without composing a single record by hand.

The shape mirrors AdminHandlers: answers are computed here and returned as AdminReply values, so every status code and every body in this file is asserted by a unit test that binds no socket and starts no broker.

Idempotence, stated exactly. A replay is not idempotent at the broker: each one appends new records to the main topic, and running the same replay twice produces two copies on the log. It is idempotent at the database, because DeadLetterReplay.producerRecord copies the original bytes verbatim, so the CloudEvents id and source survive and ON CONFLICT (occurred_at, ce_source, ce_id) DO NOTHING absorbs the second copy — which is exactly what makes retrying a replay that failed halfway a safe thing to do rather than a data-duplication risk.

The poison loop, stated exactly. A record that fails again after being replayed lands back on the DLQ, and under a new key: the origin coordinates are the coordinates of the replayed record, which is at a new offset. The DLQ's compaction key therefore does nothing to stop one defect accumulating one dead letter per replay. What does stop it is ReplayHeaders.Attempt — a counter carried in the record's own headers, preserved into each successive dead letter because a dead letter records its record's headers verbatim, and checked by the planner against cobalt.replay.max-attempts. The loop is bounded at that many generations, it is visible in every listing as replayAttempts, and each generation names its predecessor through ReplayHeaders.Of.

Attributes

Supertypes: trait StrictLogging

class Object

trait Matchable

class Any

Where a record goes when the consumer gives up on it.

A trait rather than a concrete producer for one reason that matters: this is the only side effect in the batch path that is not the database, and the unit tests of BatchProcessor are exactly the tests that need to assert what was dead-lettered without a broker in the room.

Attributes

Supertypes: class Object

trait Matchable

class Any
Known subtypes: class KafkaDeadLetterPublisher

Selection, bounding and record reconstruction — every decision a replay makes, with no broker in sight.

The split from DeadLetterAdmin is the one AdminHandlers already makes and for the same reason: the interesting behaviour is which records and in what order, and asserting a Vector should not cost a container.

Attributes

Supertypes: class Object

trait Matchable

class Any
Self type: DeadLetterReplay.type

Read and write access to the DLQ topic, as the two operations an operator needs and nothing else.

A trait because DeadLetterAdmin's decisions — bounding, refusal, the dry-run split, the meters — are worth testing without a broker, and because the Kafka implementation is a lock around two clients that a unit test would learn nothing from.

Attributes

Supertypes: class Object

trait Matchable

class Any
Known subtypes: class KafkaDeadLetterStore

One consumed record, decoded, with its offset still attached.

outcome is an Either and never an exception, which is the whole point of decoding in a stream stage rather than in a Deserializer (ADR §4.3): a throwing deserializer throws inside KafkaConsumer.poll, before the connector sees the record, so the stream dies with the offset uncommitted and every restart replays the same poison record forever.

The committable travels with the record all the way to the commit stage. It is what makes "commit only after the write" expressible as stream topology instead of as a convention.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The last thing observed about one dependency, as a value readiness can read without blocking.

Readiness must never probe inline. A /health/ready handler that opens a JDBC connection or asks a broker for metadata turns the readiness endpoint into a load generator against the very dependency that is struggling, and its own latency into the thing that fails the probe. Instead a background poller writes here and the handler reads a field — so the answer is at most one poll interval stale, and stale-but-instant beats fresh-but-hanging every time.

Starts in the down state deliberately: a process that has not yet proved it can reach its dependencies is not ready, and optimistic initialisation is how a replica joins the load balancer seconds before it starts erroring.

Attributes

Supertypes: class Object

trait Matchable

class Any

How much is on the DLQ right now.

An upper bound, and named as one. latest - earliest counts offsets, not distinct dead letters: the DLQ is keyed on origin coordinates so a compacted topic can hold fewer live records than its offset range suggests, and a transactional marker occupies an offset without being a record at all. Reporting it as exact would be the more comfortable lie; what an operator needs from this number is "is the DLQ empty, a handful, or a flood", and an upper bound answers all three without pretending to a precision the log cannot give.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Offsets for one DLQ partition.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

One record as it currently sits on the DLQ topic.

entry is an Either rather than a DeadLetter because the DLQ is a topic like any other, and nothing stops a misconfigured producer, an older build, or a hand-run kafka-console-producer from putting something else on it. An inspection tool that throws on the first unreadable record is useless precisely when the DLQ is the thing that has gone wrong, so an unreadable record is a value with a stated reason and is listed beside the readable ones.

Value parameters

key: the DLQ record key, which by construction is the origin topic/partition/offset of the record that died.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

How much of the DLQ a request got to look at, and what it found there.

The scan is reported because a filtered answer can be exact and still read as the wrong thing. A reason filter only ever matches inside the records the store fetched, so "there are no unconvertible dead letters" and "there are none among the newest two hundred" are different statements, and an operator who reads the first when the second is true stops looking. scanned says how far the answer reaches; truncated says the window ran out first.

Value parameters

missing: named refs the window did not contain. Non-empty refuses a named replay — see ReplayPlan.refusal.
selected: the records the request acts on: newest-first for a ReplayScope.Recent page, named order for a ReplayScope.Named set.
truncated: the window filled up and the request was still short, so there may be older matches this answer excludes. Never true of a full page, because a full page is a complete answer to "the newest N".

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The running consumer: the graph, its restart supervision, and its shutdown contract.

Shutdown is the reason this is a class and not four lines in Main. A SIGTERM arrives while a batch of up to batchSize records is mid-insert. Killing the stream there loses nothing durable — the offsets were not committed — but it does mean those records are re-consumed and re-inserted on the next boot, and it means every rolling deploy replays a batch. drainAndShutdown instead stops the Kafka fetcher, lets the in-flight batch finish and commit, and only then completes. Wired to CoordinatedShutdown's PhaseServiceRequestsDone, that is what makes a rolling deploy invisible in the data.

The Consumer.Control lives behind an AtomicReference that ConsumerStream.restarting re-sets on every restart attempt; see that method for the failure mode a single capture produces.

Attributes

Companion: object
Supertypes: trait StrictLogging

trait ConsumerHandle

class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: trait StrictLogging

class Object

trait Matchable

class Any
Self type: EventConsumer.type

Everything readiness consults.

Exactly two dependencies, because cobalt has exactly two: without Kafka there is nothing to consume, without PostgreSQL there is nowhere to put it. Anything else — the DLQ producer, the admin client — shares a fate with one of these two and adding it would only make the probe flap for reasons that do not change the answer.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: HealthChecks.type

Verifies JWS bearer tokens against a pinned algorithm, or verifies nothing and says so.

Why this is not wolfram's `JwtVerifier`

It is the same contract, deliberately: the same AuthConfig fields, the same AuthProblem split, the same pinned algorithm, the same refusal messages. It is a second implementation only because the shared module it belongs in does not exist yet. wolfram's verifier is built on jwt-scala, which is on wolfram's classpath and not on cobalt's; putting the type where both services can reach it means either a new modules/security project or a line in build.sbt adding Dependencies.jwt to cobalt — and neither was in this change's remit. The honest options were therefore "leave the admin API open" or "verify with the JDK". docs/services/cobalt.md records the promotion this file is waiting for; when it happens, this file is deleted rather than merged, and CobaltAuthSuite is the conformance suite the survivor has to pass.

What "verify with the JDK" means here

No cryptography is implemented in this file. The signature check is javax.crypto.Mac for HMAC and java.security.Signature for RSA, both from the JCA — exactly the primitives jwt-scala itself calls. What this file adds is the JOSE part: splitting three base64url segments, refusing anything that is not a three-part JWS, reading alg only to compare it, and comparing MACs with MessageDigest.isEqual rather than sameElements so the comparison does not leak the expected signature one byte at a time.

Two places are deliberately stricter than wolfram, because the operations behind this door are irreversible and there is no revocation list anywhere in this system:

exp is mandatory. jwt-scala validates exp when present and accepts a token without one; here a credential that never expires is one that stays valid until somebody rotates the signing secret, which nobody does on a schedule. Every issuer worth using sets exp.
A crit header is refused. RFC 7515 §4.1.11 says a recipient must reject a JWS carrying critical extensions it does not understand. This implementation understands none, so the correct answer is always "no".

Constructed through JwtVerifier.from, which returns an Either. Every way a verifier can be misconfigured — an unknown algorithm, an HMAC algorithm with no secret, an RSA algorithm with an unparseable key — is a boot failure naming the field, not a 500 on the first authenticated request.

Attributes

Companion: object
Supertypes: class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: JwtVerifier.type

The real one: a plain KafkaProducer writing structured-mode CloudEvents to the DLQ topic.

No Pekko producer sink here, on purpose. The DLQ write is a dependency of one batch's completion, not a stream of its own: the offset of a poison record may only be committed once its dead letter is durable, so the write has to be something the batch can flatMap on. Splicing a Producer.flow into the graph would either reorder that dependency or need a second commit path, and both are ways to lose a poison record while committing past it.

acks=all and idempotence come from KafkaCodecs.producerConfig and cannot be turned off by configuration. A DLQ producer that acknowledges before the record is replicated makes the dead letter less durable than the record it replaces, which defeats the entire mechanism.

Attributes

Companion: object
Supertypes: trait StrictLogging

trait DeadLetterPublisher

class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: KafkaDeadLetterPublisher.type

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: KafkaDeadLetterStore.type

The real one: one KafkaConsumer and one KafkaProducer, serialised behind a single lock.

The lock is a feature, not an implementation detail. KafkaConsumer is not thread-safe and Undertow hands every request to a different worker thread, so some serialisation is mandatory; making it one lock over the whole store additionally means two operators cannot replay at the same time. A replay is a deliberate burst of produces at a broker that is, by the nature of the situation, already having a bad day — two of them interleaved is how a bad moment becomes an outage, and the second operator waiting is the correct outcome.

assign, never subscribe. This consumer joins no group: it seeks to explicit offsets and reads a bounded range. Subscribing would give it group membership, a rebalance it can trigger by being slow, and — worst — committed offsets that could be confused with the real consumer's. Reading the DLQ must not be able to move anything.

Its own producer, not the dead-letter publisher's. Sharing one would tie two lifetimes together for the sake of one TCP connection, and the failure mode of getting that wrong (a replay against a producer the shutdown sequence has already closed) is silent. Each owner closes what it opened.

Attributes

Companion: object
Supertypes: trait StrictLogging

trait DeadLetterStore

class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: LagConfig.type

Consumer-lag polling (ADR §7.1).

Value parameters

refreshInterval: how often the AdminClient is asked for committed and end offsets. Two admin round trips per interval per replica, so this is deliberately measured in tens of seconds and not in seconds: lag is a trend, and sampling it faster than Prometheus scrapes it buys nothing but broker load.
requestTimeout: bound on each admin call. It doubles as the broker-reachability probe's timeout, so a hung controller marks the consumer not-ready instead of parking the poller thread forever.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The outcome of a lifecycle command.

Carries the state before as well as after, because "pause" applied to an already-paused consumer and "pause" applied to a running one are both successes and an operator needs to know which one happened — particularly when the command was issued twice because the first response was slow.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: LifecycleResult.type

The process entry point.

Attributes

Supertypes: trait StrictLogging

class Object

trait Matchable

class Any
Self type: Main.type

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: MaintenanceConfig.type

The two scheduled database maintenance jobs (ADR §5).

Both intervals are here rather than derived from anything, because they answer unrelated questions. The partition interval is "how stale may our knowledge of the calendar be", which is measured in hours because months are long; the refresh interval is "how stale may a dashboard be", which is measured in minutes because that is what a person looking at a chart will tolerate.

Value parameters

detachLockTimeout: how long a detach may wait for ACCESS EXCLUSIVE on the fact table before giving up until the next pass. The waiting is the dangerous part: while the detach queues, every new ingest insert queues behind it.
enabled: the escape hatch for a deployment whose partitions are managed elsewhere. Defaults to true: the failure mode of forgetting to switch it on is a total ingest outage at a month boundary.
monthsAhead: months beyond the current one to keep partitioned — ADR §5's N+3. Also the number of consecutive runs the job may miss before an event has nowhere to go, which is the reading that makes 3 generous rather than arbitrary.
partitionInterval: how often to re-check the calendar. Hours, not minutes: creating a partition is a CREATE TABLE with thirteen indexes and there is nothing to gain from discovering the new month sooner than a few hours in.
refreshInterval: how often to rebuild the hourly rollup. The floor on how stale every dashboard number is, and simultaneously the budget the refresh must fit inside: REFRESH ... CONCURRENTLY is a full recompute, so when its duration approaches this value the materialized-view design is finished and ADR §12.4's incremental merge tables are the replacement. That is what maintenance.job.duration is watched for.
retainMonths: months to keep attached, counting back from and including the current one. None — the default — disables retention entirely. Detaching is the one thing here that makes data vanish from queries, so it happens because somebody configured it and never because a default did.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The two database maintenance jobs on cobalt's schedule.

Why cobalt and not ferrite. ADR §5 assigned these to ferrite, and this build puts them here for the reason that outranks it: ferrite is the read side. cobalt already owns the Flyway migrations (see CobaltApp.start), so it is the one service that writes schema, and a partition is schema. Splitting them would mean two services able to change the shape of events.cloud_event, and the deploy-ordering question that follows — may ferrite start before cobalt has migrated? — has no good answer. The narrower argument is just as decisive: partitions exist so that inserts have somewhere to go, and ferrite never inserts. A deployment that scaled ferrite to zero overnight to save money would stop creating partitions for a write path that never stopped writing. Recorded as a deliberate deviation from ADR §5, not as an oversight.

Why the first partition pass is synchronous. A scheduled first run, even at zero delay, is a race with EventConsumer.start, and the loser is ingest: a fresh deployment of this build months after V1__events.sql was written has partitions for July and August 2026 and nothing else, so the first batch through the consumer fails with no partition of relation "cloud_event" found for row — or, worse, succeeds into cloud_event_default, where it is invisible to partition pruning and where its presence then blocks the partition the job is about to try to create. Running it to completion before the consumer exists removes the race entirely, and it costs one pass over a handful of CREATE TABLEs. This mirrors Probes.start, which probes once synchronously for the same class of reason.

A failure never stops the boot and never stops the schedule. runQuietly catches, meters and logs, because the two alternatives are both worse: an exception escaping a scheduled task cancels every subsequent run of it — the job would die silently on one transient connection error and nothing would say so — and an exception escaping the synchronous first pass would take down a consumer that is otherwise perfectly able to run, over a database hiccup, in a CrashLoopBackOff. The failure is not swallowed: it lands on maintenance.job.duration with outcome=failure, which is the channel an operator can actually be paged from.

Attributes

Supertypes: trait StrictLogging

class Object

trait Matchable

class Any
Self type: MaintenanceJobs.type

The maintenance jobs' telemetry, in the shared vocabulary of modules/observability.

Why this exists at all. Both jobs fail silently by construction. The partition job stops creating months and absolutely nothing happens — until a month boundary, at which point every insert fails with no partition of relation "cloud_event" found for row and ingest stops completely. The rollup refresh stops running and the dashboards keep answering, correctly formatted and increasingly wrong. Neither produces a request error, a failed health probe, or a Kafka symptom. A log line exists in both cases and is read by nobody, because nothing prompts anyone to look. Metrics are the only channel through which either failure reaches an operator before its consequence does.

Why the gauges are separate AtomicLongs rather than a callback into the job. Micrometer polls a gauge at scrape time on the scrape thread; a gauge whose supplier ran a query would put a database round trip — with the pool's timeout, and no statement_timeout of its own — on the path of every Prometheus scrape, and a slow database would then also break the metrics that were supposed to tell you the database is slow. Publishing last-known values into an AtomicLong decouples the two: the scrape is a field read, and the value is as fresh as the last job run. A stale gauge is exactly the right reading when the job has stopped, and maintenance.job.duration's count is what says so.

-Werror trap (ADR §7.4). Micrometer's builders return this and its registration methods return the meter, so every registration below is bound with val _ = or is part of an expression.

Attributes

Supertypes: class Object

trait Matchable

class Any

One partition's position, as the admin API reports it.

The four numbers side by side are the point. committed is where the group says it is, stored is where the checkpoint says it is, endOffset is where the log actually ends, and lag is the gap. A disagreement between committed and stored is the single most useful diagnostic this API produces — it means one of the two commits did not happen, and which one tells you whether events were lost or will be replayed.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

A record that decoded, together with everything the write path needs.

The original ConsumerRecord is kept even though the Envelope is fully decoded, because a record that the database later rejects still has to become a DeadLetter, and a dead letter is only replayable if it carries the original bytes and headers.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Who made the request, once the token has been verified.

Deliberately small. Everything on it was proved by the signature check, so a Principal in scope is the authorisation decision — nothing downstream re-checks and the raw token never travels past JwtVerifier.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: Probes.type

The background poller behind both the lag gauge and the readiness probe.

One timer thread doing both, on purpose: the two questions have the same period, the same failure modes, and the same consequence when the answer is "no". Splitting them would give two schedules to keep in sync and a window in which the gauge says the broker is fine while readiness says it is not.

A failure here is recorded, never thrown. An exception escaping a ScheduledExecutorService task silently cancels all subsequent runs — the gauge would freeze at its last value and readiness would freeze at its last answer, which is the exact opposite of what a health poller is for.

Attributes

Companion: object
Supertypes: trait StrictLogging

class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: RecordDecoder.type

Turns a committable Kafka message into a DecodedRecord, inside a CONSUMER span.

Total by construction. Every failure path — an unrecognised content mode, malformed JSON, a missing required attribute, an envelope with no time and therefore no partition to land in, and even an unexpected runtime exception — becomes a Left(DeadLetter). Nothing here throws, because the only thing a throw could achieve is to kill a stream whose restart would meet the same record again.

Trace continuation happens here (ADR §7.2). KafkaTrace.withConsumerSpan extracts the W3C traceparent the producer injected and opens a CONSUMER span parented to it, so a trace that starts at wolfram's HTTP ingress continues into this consumer rather than starting a second, unrelated root trace. The span is made current for the duration of the decode, which is what puts trace_id/span_id into the MDC of anything logged from it.

Known limitation, stated rather than hidden: the span covers the decode, not the batched insert. The insert is a batch operation spanning many traces at once, so it cannot be a child of any one of them; it gets its own span in BatchProcessor. Holding every record's span open across the batch boundary was the alternative and it makes the consumer hold one live span per in-flight record — up to batchSize of them — with no way to end them if the stream is torn down mid-batch.

Attributes

Companion: object
Supertypes: class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ReplayConfig.type

The bounds on the dead-letter admin surface (AdminRoutes.DlqPath).

Every field here exists to make an operator action bounded, which is the whole difference between a replay tool and a way to turn a poison message into a poison-message storm.

Value parameters

enabled: whether a replay may be committed. false still permits dry runs and the inspection endpoints — a deployment that has switched replay off gains nothing from also blinding its operators, and would only push them back to a console consumer. Defaults to true: the endpoint is on the admin listener, which is not a public port, and a tool nobody can use during an incident is the failure this whole surface exists to prevent.
maxAttempts: how many times any one record may be replayed, ever. The bound on the poison loop: a record that fails again comes back to the DLQ under new origin coordinates, so nothing in Kafka limits the cycle; the counter in ReplayHeaders.Attempt does. Three is generous — a defect that survives three deploys is not going to be fixed by a fourth replay.
maxRecords: the ceiling on how many dead letters one request may list or replay. A request above it is refused, not clamped — see ReplayRequest.parse — and it bounds the fetch as well as the produce, so a listing on a topic with a million dead letters costs the same as one on a topic with ten.
pollTimeout: how long one read of the DLQ may take before it answers with what it has. A listing that hangs is worse than a listing that is short, and this endpoint is consulted when the broker is the suspect.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

What the planner decided about one candidate.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Where a partially-completed replay stopped.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The replay markers this build adds to a republished record.

Neither name carries the ce_ prefix, and that is load-bearing. A ce_-prefixed header is a CloudEvents context attribute under the Kafka binding, so spelling these ce_replayattempt would change the event: a different extension set, a different raw jsonb, a different row. As plain transport headers they are invisible to the decoder and to everything downstream of it, which is exactly what "indistinguishable from the original" needs.

Attributes

Supertypes: class Object

trait Matchable

class Any
Self type: ReplayHeaders.type

The two replay meters, in the shared vocabulary of modules/observability.

Same rules as ConsumerMetrics and for the same reason: no name and no tag key is invented here, every tag value comes from Meters.Outcomes, and nothing derived from a request — a ref, a topic, a skip reason — is ever a tag. The DLQ is where the unbounded content in this system ends up, so it is the last place to relax that rule.

A counter per record and a counter per operation, not one or the other. One operation replaying two hundred records and two hundred operations replaying one each produce the same record count and describe very different situations: the first is a recovery, the second is somebody in a loop. Only the operation counter distinguishes them, and only the record counter says how much went back on the topic.

-Werror trap (ADR §7.4). MeterRegistry#counter returns the meter, so every call below is part of an expression rather than a bare statement.

Attributes

Supertypes: class Object

trait Matchable

class Any

What a replay actually did, as distinct from what it planned to do.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Everything a replay would do, computed before anything is published.

The plan is the product; committing it is an afterthought. Splitting the two is what makes the dry run honest: one function produces the plan whether or not it will be executed, so a dry run is not an approximation of the replay — it is the replay, minus the produce loop. An operator who cannot see what a replay would do will not run one during an incident, which makes the tool worthless exactly when it is needed.

Value parameters

scan: how much of the DLQ the plan was computed from. Carried through to the response so a short plan can say whether it is short because the DLQ is, or because the fetch window ran out.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ReplayRequest.type

A parsed, bounded replay request. Build it with ReplayRequest.parse; the fields are validated by then.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

What a replay operation was asked to act on.

Two shapes, because an operator means two different things and they need different failure behaviour.

Recent is "whatever is in there, up to N". The set is discovered, so a record the planner declines is reported and the rest still go.
Named is "these exact records". The set is stated, so anything the planner cannot find or cannot replay refuses the whole operation — publishing three of the five records somebody named is the half-success this endpoint exists not to produce.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Why the planner declined to replay a dead letter.

A closed set, and it stays closed: these strings appear in the response body an operator reads under time pressure and in the log line that outlives the request. They are deliberately not a meter tag — see com.worxbend.observability.Meters.DlqReplayRecords for why the skip count is untagged.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: ReplaySkip.type

Bounded exponential backoff for the consumer's RestartSource (ADR §4.3).

Bounded on purpose. An unbounded restart loop against a broker that is never coming back looks identical, from the outside, to a healthy consumer with no traffic: the process stays up, the pod stays Ready, and the only symptom is lag. maxRestarts within maxRestartsWithin makes the stream fail, which fails readiness, which is a signal an operator actually receives.

Value parameters

randomFactor: jitter. Without it every replica of a service retries in lockstep and the recovering broker is hit by a synchronised thundering herd at each backoff step.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: RestartConfig.type

What a restart did.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: RunState.type

What the consumer is doing right now.

A closed enum and not a boolean pair. "running" and "paused" as two flags admits four states, two of which are nonsense, and every reader then has to know which combination means what. One value with six cases makes the illegal states unrepresentable and makes the admin response a single word an operator can read.

The distinction that matters most is Stopped versus Failed. Both mean "not consuming"; only one of them is somebody's fault. A supervisor that reported a crashed stream as stopped would look exactly like one an operator had paused on purpose, and lag would grow while the dashboard said everything was fine.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: SeekOffset.type

An explicit seek, as an operator supplies it: topic/partition/offset.

Named SeekOffset and not OffsetSpec because Kafka's admin API already owns that name in this package (org.apache.kafka.clients.admin.OffsetSpec, used by AdminOffsets), and two OffsetSpecs one import apart is a compile error at best and the wrong type at worst.

Parsed rather than taken as JSON so the same value works as a query parameter and in a request body, and so a malformed one is refused with a sentence naming the expected shape instead of a decoder's field path.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Where a restart should begin.

Stored is the default and the interesting one. Kafka's auto.offset.reset only applies when the group has no committed offset at all, so it cannot express "resume from the position I durably recorded" — which is what an operator almost always means after an incident. The externalised checkpoint can.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: SeekTarget.type

Where the operational HTTP surface binds.

cobalt serves no business endpoints (ADR §1): this listener exists only for /metrics and the two health probes, so it is configured separately from everything that touches Kafka — a platform team owns the port, the service team owns the stream.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ServerConfig.type

The consumer lifecycle, over HTTP.

Why this exists. A consumer that can only be stopped by stopping the process is a consumer whose every incident response is docker compose restart plus a kafka-consumer-groups.sh invocation nobody has memorised, run inside a container that has to be given a shell first. Every operation here replaces one of those.

Every mutating route is a POST, and one of them defaults to refusing. restart moves a consumer group's committed offsets, which is the single most destructive thing an operator can do to this pipeline: latest skips unconsumed events outright. So it plans by default and commits only when told to, in the same shape the DLQ replay uses — one convention for both dangerous operations rather than two to remember at 3am.

The blocking is deliberate and bounded. These handlers Await on the supervisor. Cask serves each request on its own Undertow worker thread, a lifecycle transition is inherently one-at-a-time, and an operator issuing a stop wants the response to mean stopped — a 202 with an opaque promise would put them straight back into polling the status endpoint. The timeout is what keeps a hung drain from holding the worker forever.

Attributes

Supertypes: trait StrictLogging

class Object

trait Matchable

class Any

The consumer supervisor's own state, as metrics.

The gauge that matters most is consume.running, and it exists to make a deliberate pause distinguishable from an outage. Consumer lag rising with the consumer running is a consumer that cannot keep up; lag rising with it paused is somebody's maintenance window. Alerting on lag alone cannot tell those apart, which is exactly how a planned pause pages the on-call at 2am — and, worse, how everyone learns to acknowledge that page without reading it.

Gauges, not counters, and fed from a poller rather than from the transition. A gauge set only when the state changes reports the truth right up until the process restarts, at which point it reports whatever the constructor left there. Micrometer's gauge holds a weak reference to the state object and reads it at scrape time, so a value updated by the same poller that already computes lag is both cheaper and more honest: it converges on the real value within one interval no matter what happened before.

An unread gauge reports NaN, never its last value and never zero. This is the other half of the same idea and it is a correction: these gauges used to hold their reading indefinitely, so a poller whose Await was timing out left consume.running frozen at 1 for the whole incident — the exact number an operator uses to rule the consumer out as the cause. Prometheus renders NaN and every aggregation over it drops out, so "the probe has stopped reporting" looks like absence rather than like a healthy zero. The cost is that the first scrape after a boot shows gaps until the first probe lands, which is the truth.

Value parameters

freshFor: how long a reading stays believable. Comfortably more than the probe interval, because one slow tick is not an incident; SupervisorProbe's caller sizes it from that interval.

Attributes

Companion: object
Supertypes: class Object

trait Matchable

class Any

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: SupervisorMetrics.type

The supervisor's readings, taken together on the probe's tick.

This type exists so that a registered gauge cannot go unwritten. dlq.depth shipped registered and with no writer at all: /metrics reported a flat 0 while the DLQ filled, the dashboard panel stayed at zero, and docs/operations.md told the operator that gauge was how you check whether a fix worked. Nothing in the code said the wiring was missing, because a gauge with no writer looks exactly like a gauge reporting good news. Both readings now live behind one method with one caller, and SupervisorMetricsSuite asserts that a single tick leaves no registered gauge unread — which is the assertion whose absence let that ship.

Each reading is taken independently and neither can take the other down. They come from different systems and the moment one of them fails is precisely the moment somebody is reading the other.

Value parameters

budget: the bound on ConsumerSupervisor.status, and it has to be able to contain the calls it wraps. status is a checkpoint read followed by three sequential admin round trips, each already bounded by the admin request timeout; a budget of one request timeout expires before the call it wraps can possibly return, so under a slow broker every tick threw and both gauges froze — the failure the poller was introduced to prevent, reintroduced one layer up.

Attributes

Supertypes: trait StrictLogging

class Object

trait Matchable

class Any

com.worxbend.cobalt

Members list

Type members

Classlikes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters

Attributes

Attributes

Value parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters

Attributes

Value parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters