Documentation
¶
Index ¶
- Constants
- func ClientConfig(cfg *internal.Config) pkgkafka.Config
- func ClientID(cfg *internal.Config) string
- func ConsumerConfig(cfg *internal.Config, name, topic string) pkgkafka.ConsumerConfig
- func ConsumerGroup(cfg *internal.Config, name string) string
- func New(params Params) (*pkgkafka.Producer, error)
- type Message
- type Params
- type Producer
Constants ¶
const ( TopicRunnerHeartBeats = "runner_heart_beats" TopicOtelLogRecords = "otel_log_records" TopicOtelTraces = "otel_traces" TopicDLQ = "dlq" )
Topics this service produces to / consumes from. Names mirror the destination ClickHouse tables.
const ( TypeRunnerHeartBeat = "runner_heart_beat" TypeOtelLogRecord = "otel_log_record" TypeOtelTrace = "otel_trace" TypeDLQ = "dlq_record" )
Envelope message types.
Variables ¶
This section is empty.
Functions ¶
func ClientConfig ¶
ClientConfig maps ctl-api config into the generic Kafka client config. Shared by the producer here and the domain consumers.
func ClientID ¶
ClientID identifies this process to the brokers. Kafka reports it in request metrics, broker logs, and consumer group member ids, and — the reason it is worth being specific — applies client quotas per client id, so a shared value would make it impossible to throttle one producer without throttling all of them. It is also stamped into every envelope as the message source.
Derived from the same service_type/service_deployment that tag our pods in Datadog, so a client id here names the deployment the same way the dashboards do: ctl-api/api-runner, ctl-api/worker-installs, ctl-api/consumer-clickhouse-sink.
func ConsumerConfig ¶
func ConsumerConfig(cfg *internal.Config, name, topic string) pkgkafka.ConsumerConfig
ConsumerConfig maps ctl-api config into the generic consumer config for one topic. The fetch tunables are process-wide config, set per deployment, since each consumer runs in its own pod.
func ConsumerGroup ¶
ConsumerGroup names the group for one consumer. Per-consumer rather than one shared group because a group with heterogeneous topic subscriptions rebalances every member whenever any member restarts — so a deploy of one consumer would stall the others for no reason.
Changing a group's name makes it start from the earliest retained offset (pkgkafka sets ConsumeResetOffset to AtStart), which for a topic with history means replaying the whole retention window. Rename only while a topic is cold, or seed the new group from the old one's committed offsets first.
Source Files