SupervisorAdmin

com.worxbend.cobalt.SupervisorAdmin
final class SupervisorAdmin(supervisor: ConsumerSupervisor, timeout: FiniteDuration, metrics: SupervisorMetrics) extends StrictLogging

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

Graph
Supertypes
trait StrictLogging
class Object
trait Matchable
class Any

Members list

Value members

Concrete methods

POST /admin/consumer:clearCheckpoints — forget the externalised offsets.

POST /admin/consumer:clearCheckpoints — forget the externalised offsets.

Does not touch Kafka's __consumer_offsets. This is the "the stored position is wrong, use the broker's answer instead" escape hatch, and conflating it with a Kafka offset reset would make one word mean two irreversible things.

Attributes

def pause(): AdminReply

POST /admin/consumer:pause — drain, commit, leave the group, and stay out.

POST /admin/consumer:pause — drain, commit, leave the group, and stay out.

Attributes

def restart(target: String, offsets: String, dryRun: Boolean): AdminReply

POST /admin/consumer:restart — stop, move the group's offsets, start.

POST /admin/consumer:restart — stop, move the group's offsets, start.

target selects where to resume from; offsets supplies the coordinates when it is explicit. dryRun defaults to true, so the first call answers "here is what I would set" and changes nothing.

Attributes

POST /admin/consumer:resume — materialise a fresh stream and rejoin the group.

POST /admin/consumer:resume — materialise a fresh stream and rejoin the group.

Attributes

def start(): AdminReply

POST /admin/consumer:start

POST /admin/consumer:start

Attributes

GET /admin/consumer — the state machine plus committed, stored and end offsets per partition.

GET /admin/consumer — the state machine plus committed, stored and end offsets per partition.

Attributes

def stop(): AdminReply

POST /admin/consumer:stop — the same as pause, reported as an outage rather than as intent.

POST /admin/consumer:stop — the same as pause, reported as an outage rather than as intent.

Attributes

Inherited fields

protected val logger: Logger

Attributes

Inherited from:
StrictLogging