Documentation
¶
Index ¶
- Constants
- func DedupToken(topic string, partition int32, first, last int64) string
- func Wrap(source, typ string, payload any) ([]byte, error)
- type Config
- type Consumer
- type ConsumerConfig
- type Envelope
- type Handler
- type Message
- type Producer
- func (p *Producer) Close()
- func (p *Producer) Enabled() bool
- func (p *Producer) Flush(ctx context.Context) error
- func (p *Producer) Ping(ctx context.Context) error
- func (p *Producer) Produce(ctx context.Context, topic, key string, value []byte)
- func (p *Producer) ProduceEnvelope(ctx context.Context, topic, key, typ string, payload any) error
- func (p *Producer) ProduceEnvelopesSync(ctx context.Context, topic, typ string, msgs []Message) []int
Constants ¶
const EnvelopeVersion = 1
Variables ¶
This section is empty.
Functions ¶
func DedupToken ¶
DedupToken builds a stable per-batch token from a partition's offset range, suitable for a ClickHouse insert_deduplication_token.
Types ¶
type Config ¶
type Config struct {
Brokers []string
ClientID string
SecurityProtocol string
// TLSCAPath is the bundle used to verify brokers; empty uses system roots.
TLSCAPath string
// TLSCertPath and TLSKeyPath enable mTLS and must be set together.
TLSCertPath string
TLSKeyPath string
// ProduceTimeout bounds a synchronous produce. franz-go retries a buffered
// record effectively forever by default, so without a bound a broker outage
// would block a caller waiting on the ack indefinitely. Only applies to the
// sync path; async producers never wait.
ProduceTimeout time.Duration
}
Config is the Kafka client configuration shared by the producer and consumer.
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer runs a consumer-group poll loop and dispatches each partition's records to a Handler, committing offsets only after the handler succeeds.
func NewConsumer ¶
func (*Consumer) Stuck ¶
Stuck reports whether a handler call has been running longer than max, and for how long. Intended for a liveness check, not the hot path: a handler call is expected to be bounded by its own timeout (e.g. a ClickHouse write deadline), so this only trips on a genuine hang that bound failed to catch.
type ConsumerConfig ¶
type ConsumerConfig struct {
Group string
Topics []string
FetchMaxWait time.Duration
// FetchMinBytes is the batch floor, the rest are ceilings. Worst-case buffered
// bytes is
//
// min(FetchMaxBytes, partitionsOnBroker*FetchMaxPartitionBytes) * MaxConcurrentFetches
//
// so partitions-per-pod — replica count, not topic size — is usually what
// actually bounds a consumer's memory.
FetchMinBytes int32
FetchMaxBytes int32
FetchMaxPartitionBytes int32
MaxConcurrentFetches int
}
type Envelope ¶
type Envelope struct {
Version int `json:"version"`
Type string `json:"type"`
ProducedAt time.Time `json:"produced_at"`
Source string `json:"source"`
Payload json.RawMessage `json:"payload"`
}
Envelope is the versioned wrapper every Kafka message uses. Payload is left as raw JSON so consumers decode it by Type/Version rather than the producer and consumer sharing a compiled payload type.
func Unwrap ¶
Unwrap parses an envelope. The caller unmarshals Payload based on Type/Version.
Rejects a version newer than EnvelopeVersion, not merely a different one: an older envelope must still pass, since a rolling deploy can have not-yet-restarted producers still writing it while consumers have already upgraded to accept a new one. Bumping EnvelopeVersion is an expand-then-contract change — consumers accept the new version first, then producers start writing it — so an exact-match check would reject the new version everywhere until every consumer had rolled, which defeats the expand step entirely.
type Handler ¶
Handler processes one partition's batch. Returning nil commits the batch; returning an error leaves it uncommitted for redelivery.
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer wraps a franz-go client. A disabled producer is a no-op, letting callers keep a legacy fallback path so downstream writes never depend on Kafka being present.
Two produce modes, chosen by what the caller's current write does:
- Produce / ProduceEnvelope are fire-and-forget. Correct where the existing path is already lossy — heartbeats buffer in memory and flush on a 5s ticker, so a crash already drops a few seconds of them and nothing cares.
- ProduceEnvelopesSync waits for the acks. Correct where the existing path is synchronously durable, as the OTLP log write is: it blocks on a ClickHouse insert before returning 201, so producing fire-and-forget instead would hand the caller a success for records that live only in a process buffer.
func NewProducer ¶
func (*Producer) Produce ¶
Produce sends a single pre-marshaled message. Fire-and-forget: the async callback records metrics and logs on failure, so callers never block on the broker.
func (*Producer) ProduceEnvelope ¶
ProduceEnvelope wraps payload in the versioned envelope and produces it. Fire-and-forget; see ProduceEnvelopesSync when the caller needs the ack.
func (*Producer) ProduceEnvelopesSync ¶
func (p *Producer) ProduceEnvelopesSync(ctx context.Context, topic, typ string, msgs []Message) []int
ProduceEnvelopesSync wraps each message and produces the batch, blocking until every record is acked. Acks are already RequiredAcks(AllISRAcks) with idempotence on, so an ack here means the record is on every in-sync replica.
Returns the indices of messages that were NOT acked, so a caller with a fallback can write exactly those rather than re-writing the whole batch and duplicating the ones that succeeded. An empty return means everything is durable in Kafka.
Produces the whole batch in one call rather than looping: ProduceSync enqueues all records, cancels lingering, and waits once, so this costs one round trip's latency instead of N serialized ones.
On timeout the outcome for an in-flight record is genuinely ambiguous. franz-go will not fail a buffered record that has already been sent while producing idempotently, because it cannot know whether the broker applied it. Such a record is reported as failed here — so the caller writes it via the fallback — and may still land in Kafka afterwards. That is an at-least-once choice: a duplicate row is recoverable, a silently dropped log line is not.
Source Files