Documentation
¶
Overview ¶
Package consumer is the Kafka consumer runtime: consumer naming and selection, the shared ClickHouse-writing sink, and the dead-letter path. It is to `consumer` what internal/pkg/worker is to `worker` — none of it knows anything about a particular domain.
Domain consumers live with their domain (app/runners/consumer holds heartbeats, otel-logs and otel-traces) and consist of little more than a Sink plus a decode-and-insert handler. The one consumer that lives here is the DLQ (see dlq.go), because the dead-letter topic is shared infrastructure that every consumer produces to rather than any domain's data.
Index ¶
Constants ¶
const NameAll = "all"
NameAll selects every consumer, mirroring `worker --namespace=all`. Locally one process runs them all.
Variables ¶
This section is empty.
Functions ¶
func Decode ¶
Decode unwraps a partition's records into T, skipping anything undecodable or of the wrong type. A record we can't parse is dropped rather than returned as an error: failing the batch would block the partition forever re-reading the same bad record, so it's counted, collected, and skipped instead — every failure in this fetch is sent to the dead-letter queue in one batch after the loop, not one at a time as they're found. A burst of bad records (a bad deploy, a version-skewed producer) would otherwise mean one synchronous produce round trip per record, and a large enough burst could make this handler call itself look stuck — the exact failure mode the dead-letter queue exists to avoid, self-inflicted.
func Insert ¶
func Insert[T any](ctx context.Context, s *Sink, partition int32, recs []*kgo.Record, rows []T) error
Insert writes one partition's decoded batch to ClickHouse, keyed by the batch's offset range as an insert_deduplication_token. Offsets are only committed after this returns nil, so a failed insert is redelivered; the token is what makes that replay idempotent rather than duplicating rows.
Types ¶
type DLQConsumer ¶
type DLQConsumer struct {
*Sink
}
DLQConsumer drains the dead-letter topic into the same app.DLQRecord table the other consumers write to directly when a produce to this topic fails.
It lives in the runtime package rather than with a domain because the topic isn't any domain's data — every consumer produces to it, and what it carries is a decode failure from some other topic.
Its own decode failures go straight to that direct write rather than back through the topic — a dead-letter-about-a-dead-letter stops here, one level deep.
func NewDLQConsumer ¶
func NewDLQConsumer(params Params) (*DLQConsumer, error)
type Name ¶
type Name string
Name identifies one consumer: it is an element of the `consumer --name` value, the suffix of the consumer group, and the prefix of the consumer's metrics. One Name owns exactly one topic.
A Name is NOT a deployment. A deployment (a pod, its resources, its SERVICE_DEPLOYMENT tag, and therefore its Kafka client id) may host more than one consumer — `--name=otel-logs,otel-traces` runs both in one pod under the deployment named `otel`, each still with its own group and its own client. Consumers sharing a pod share a client id, so they also share a Kafka client quota, and share a liveness probe: if one handler wedges, the restart takes the other down with it. Group them only when scaling and failing together is acceptable.
The names live here, in the runtime, rather than each with its domain: NewSelection runs before the fx graph is built (see cmd/consumer.go), so the set of valid names has to be statically known and can't be collected from the domains at wiring time.
type Selection ¶
type Selection struct {
// contains filtered or unexported fields
}
Selection is the parsed `--name` flag.
func NewSelection ¶
NewSelection parses a comma-separated list of consumer names, or "all".
Unknown names are an error rather than a no-op: a typo in a deployment's flag would otherwise produce a pod that starts, passes health checks, and silently consumes nothing, which is a much harder failure to spot than a crash loop.
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink is the shared half of every consumer that writes to ClickHouse: naming, metrics, the ClickHouse handle, and the poll-loop lifecycle. Each consumer embeds one and adds only its own decode-and-insert handler.
func NewSink ¶
NewSink returns nil when this consumer should not run in this process — either it wasn't selected, or Kafka is disabled entirely. Callers return a nil consumer in that case and whatever inline write path they replaced stays in effect.
func (*Sink) ConsumerName ¶
ConsumerName identifies which consumer this sink is, so a healthcheck covering several sinks in one pod can report which one is stuck rather than just that something is.
Source Files