kafka

package
v1.153.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 15, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package kafka provides a pure-Go API for producing and consuming Apache Kafka messages. It requires no CGO and no system librdkafka installation.

Built on github.com/segmentio/kafka-go, it exposes a producer/consumer interface:

Functional options tune the session timeout, start offset, required acks, batching, and custom codecs.

Delivery Semantics

Producer writes wait for broker acknowledgment from the full in-sync replica set by default (kafka.RequireAll); tune this with WithRequiredAcks.

On the consumer side, when a consumer group is configured:

  • Consumer.Receive and Consumer.ReceiveData are at-most-once: the offset is committed as soon as the message is read, before the caller processes it, so a crash or decode failure after the read permanently skips the message.
  • Consumer.FetchMessage + Consumer.CommitMessages are at-least-once: the offset is committed only when the caller explicitly acknowledges the message after successful processing.

When no consumer group is configured (empty groupID), offsets are never committed: Consumer.Receive and Consumer.FetchMessage behave identically, reading always starts from the earliest available offset, and Consumer.CommitMessages returns an error.

Message Encoding and Decoding

Typed payload methods use configurable codec hooks:

Both defaults use github.com/tecnickcom/nurago/pkg/encode. Replace them via WithMessageEncodeFunc and WithMessageDecodeFunc to add custom wire formats, encryption, compression, or schema validation.

Errors

Configuration problems are reported at construction time with errors matching the exported sentinels ErrInvalidOptions, ErrNilEncodeFunc, and ErrNilDecodeFunc. After Consumer.Close, the receive methods return errors matching ErrConsumerClosed. Match the sentinels with errors.Is.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrInvalidOptions is returned by NewConsumer and NewProducer when the
	// broker list, the topic, or a configured option value is missing or out
	// of range.
	ErrInvalidOptions = errors.New("kafka: missing or invalid client options")

	// ErrNilEncodeFunc is returned by NewProducer when the message encode function is nil.
	ErrNilEncodeFunc = errors.New("kafka: nil message encode function")

	// ErrNilDecodeFunc is returned by NewConsumer when the message decode function is nil.
	ErrNilDecodeFunc = errors.New("kafka: nil message decode function")

	// ErrConsumerClosed is returned by Receive, FetchMessage, and ReceiveData
	// after the consumer has been closed with Close.
	ErrConsumerClosed = errors.New("kafka: consumer closed")
)

Sentinel errors returned by the package. They can be matched with errors.Is so callers can distinguish configuration problems from runtime failures.

Functions

func DefaultMessageDecodeFunc 

func DefaultMessageDecodeFunc(_ context.Context, msg []byte, data any) error

DefaultMessageDecodeFunc is the default ReceiveData() deserializer, using encode.ByteDecode. The data argument must be a pointer matching the type of the decoded message.

func DefaultMessageEncodeFunc 

func DefaultMessageEncodeFunc(_ context.Context, data any) ([]byte, error)

DefaultMessageEncodeFunc is the default SendData() serializer, using encode.ByteEncode.

Types

type Consumer 

type Consumer struct {
	// contains filtered or unexported fields
}

Consumer reads messages from a Kafka topic with pluggable decoding.

func NewConsumer 

func NewConsumer(brokers []string, topic, groupID string, opts ...Option) (*Consumer, error)

NewConsumer constructs a Kafka consumer for a topic and consumer group with optional tuning. Call HealthCheck() to verify broker/topic connectivity before beginning receives.

groupID may be empty: without a consumer group the reader always starts from the earliest available offset (WithFirstOffset has no effect), offsets are never committed, and CommitMessages returns an error.

Configuration problems are reported with errors matching ErrInvalidOptions or ErrNilDecodeFunc; match them with errors.Is.

func (*Consumer) Close 

func (c *Consumer) Close() error

Close releases Consumer's resources and closes the broker connection. It is safe to call multiple times; receive methods called after Close return an error matching ErrConsumerClosed.

func (*Consumer) CommitMessages 

func (c *Consumer) CommitMessages(ctx context.Context, msgs ...kafka.Message) error

CommitMessages acknowledges (commits the offsets of) the messages returned by FetchMessage. It must be called only after the messages have been successfully processed, to obtain at-least-once delivery semantics. When no consumer group is configured, offset commits are unavailable and CommitMessages returns an error.

func (*Consumer) FetchMessage 

func (c *Consumer) FetchMessage(ctx context.Context) (kafka.Message, error)

FetchMessage reads one message from the Kafka broker without committing its offset, blocking until a message arrives or ctx ends.

Delivery semantics: at-least-once. When a consumer group is configured, the caller must explicitly acknowledge the message by passing it to CommitMessages after successful processing; uncommitted messages are redelivered after a rebalance or restart. When no consumer group is configured, FetchMessage behaves like Receive.

After Close the returned error matches ErrConsumerClosed; when ctx is canceled or its deadline expires the returned error matches ctx.Err(). Match both with errors.Is.

func (*Consumer) HealthCheck 

func (c *Consumer) HealthCheck(ctx context.Context) error

HealthCheck verifies broker reachability by probing each configured broker until one succeeds. When every broker fails, the individual probe errors are joined into the returned error.

The default probe performs a partition lookup over plaintext TCP with the default kafka-go dialer, so it does not exercise TLS or SASL and always performs real network I/O even when a reader is injected via WithKafkaReader. Override it with WithBrokerCheckFunc to customize the dialer or to make HealthCheck deterministic in tests.

func (*Consumer) Receive 

func (c *Consumer) Receive(ctx context.Context) ([]byte, error)

Receive reads one message from the Kafka broker, blocking until a message arrives or ctx ends.

Delivery semantics: at-most-once. When a consumer group is configured, the message offset is committed as soon as the message is read, before the caller processes it; if the process crashes (or decoding fails in ReceiveData) after Receive returns, the message is permanently skipped. For at-least-once semantics use FetchMessage and commit explicitly with CommitMessages after successful processing. Without a consumer group no offset is ever committed.

After Close the returned error matches ErrConsumerClosed; when ctx is canceled or its deadline expires the returned error matches ctx.Err(). Match both with errors.Is.

func (*Consumer) ReceiveData 

func (c *Consumer) ReceiveData(ctx context.Context, data any) error

ReceiveData reads a message and decodes it into the provided data argument using the configured decoder.

Delivery semantics: at-most-once, see Receive. In particular, when a consumer group is configured the offset is already committed when decoding happens, so a message failing to decode is permanently skipped. For at-least-once semantics use FetchMessage + CommitMessages and decode the payload manually.

After Close the returned error matches ErrConsumerClosed; see Receive.

type KReader 

type KReader interface {
	ReadMessage(ctx context.Context) (kafka.Message, error)
	FetchMessage(ctx context.Context) (kafka.Message, error)
	CommitMessages(ctx context.Context, msgs ...kafka.Message) error
	Close() error
}

KReader defines the kafka-go reader methods used by Consumer.

type KWriter 

type KWriter interface {
	WriteMessages(ctx context.Context, msgs ...kafka.Message) error
	Close() error
}

KWriter defines the kafka-go writer methods used by Producer.

type Option 

type Option func(*config)

Option configures Kafka producer/consumer behavior.

func WithBalancer 

func WithBalancer(b kafka.Balancer) Option

WithBalancer sets the partitioning strategy (kafka.Balancer) used by the Producer to assign messages to partitions. If not set, the default is a kafka.Hash balancer.

NOTE: the default kafka.Hash balancer partitions by message Key. Because Send() publishes messages without a Key, the default concentrates all messages on a single partition. Use WithBalancer (for example with &kafka.RoundRobin{} or &kafka.LeastBytes{}) to distribute keyless messages across partitions.

This option is producer-only and has no effect on a Consumer; passing it to NewConsumer is silently ignored.

func WithBatchSize 

func WithBatchSize(size int) Option

WithBatchSize sets the maximum number of messages the Producer buffers into a single batch before sending it to the brokers. If not set (or set to 0), the kafka-go library default (100) is used. Negative values are rejected by NewProducer with an error matching ErrInvalidOptions.

This option is producer-only and has no effect on a Consumer; passing it to NewConsumer is silently ignored.

func WithBatchTimeout 

func WithBatchTimeout(t time.Duration) Option

WithBatchTimeout sets the maximum time an incomplete message batch is buffered before being flushed to the brokers. Because Send() and SendData() publish one message synchronously per call, each call can block up to this duration; it defaults to 10ms to keep per-message latency low.

The value must be positive: the kafka-go Writer silently replaces a non-positive timeout with its own 1s default, so NewProducer rejects such values with an error matching ErrInvalidOptions.

This option is producer-only and has no effect on a Consumer; passing it to NewConsumer is silently ignored.

func WithBrokerCheckFunc 

func WithBrokerCheckFunc(fn func(ctx context.Context, address string) error) Option

WithBrokerCheckFunc overrides the per-broker reachability probe used by Consumer.HealthCheck and Producer.HealthCheck. fn is called with each configured broker address until one returns nil.

The default probe performs a partition lookup over plaintext TCP with the default kafka-go dialer. Override it to use a custom dialer (for example with TLS or SASL), or to make HealthCheck deterministic and network-free in tests when a client is injected via WithKafkaReader / WithKafkaWriter.

func WithFirstOffset 

func WithFirstOffset() Option

WithFirstOffset makes a consumer group without a committed offset start consumption from the earliest available offset instead of the default latest. It only applies when a consumer group is configured: without a group ID this option has no effect and reading always starts from the earliest available offset.

This option is consumer-only and has no effect on a Producer; passing it to NewProducer is silently ignored.

func WithKafkaReader 

func WithKafkaReader(r KReader) Option

WithKafkaReader injects an existing kafka-go reader (or a compatible implementation), primarily for testing.

When a reader is injected, NewConsumer does not build a kafka.ReaderConfig: the groupID argument is not used for reading, and the session timeout and start offset settings have no effect (their values are still validated). The brokers and topic arguments remain required (and validated) because HealthCheck probes them; injecting a reader does NOT mock HealthCheck, which still performs real network I/O unless WithBrokerCheckFunc is also supplied.

This option is consumer-only and has no effect on a Producer; passing it to NewProducer is silently ignored.

func WithKafkaWriter 

func WithKafkaWriter(w KWriter) Option

WithKafkaWriter injects an existing kafka-go writer (or a compatible implementation), primarily for testing.

When a writer is injected, NewProducer does not construct a kafka.Writer: the balancer, required acks, and batch settings have no effect (their values are still validated). The brokers and topic arguments remain required (and validated) because HealthCheck probes them and the topic appears in encode error messages; injecting a writer does NOT mock HealthCheck, which still performs real network I/O unless WithBrokerCheckFunc is also supplied.

This option is producer-only and has no effect on a Consumer; passing it to NewConsumer is silently ignored.

func WithMessageDecodeFunc 

func WithMessageDecodeFunc(f TDecodeFunc) Option

WithMessageDecodeFunc overrides DefaultMessageDecodeFunc for ReceiveData() deserialization. The data argument to ReceiveData() must be a pointer to the correct type.

func WithMessageEncodeFunc 

func WithMessageEncodeFunc(f TEncodeFunc) Option

WithMessageEncodeFunc overrides DefaultMessageEncodeFunc for SendData() serialization.

func WithRequiredAcks 

func WithRequiredAcks(acks kafka.RequiredAcks) Option

WithRequiredAcks sets the number of broker acknowledgments required before a Producer write is considered successful:

  • kafka.RequireNone (0): fire-and-forget, broker-side failures are silently lost;
  • kafka.RequireOne (1): wait for the partition leader to acknowledge the write;
  • kafka.RequireAll (-1): wait for the full in-sync replica set (default).

Any other value is rejected by NewProducer with an error matching ErrInvalidOptions.

This option is producer-only and has no effect on a Consumer; passing it to NewConsumer is silently ignored.

func WithSessionTimeout 

func WithSessionTimeout(t time.Duration) Option

WithSessionTimeout customizes the heartbeat timeout for broker group-management failure detection. Defaults to 10 seconds if not set.

The value must be positive and below math.MaxInt32 milliseconds (~24.8 days); NewConsumer rejects out-of-range values with an error matching ErrInvalidOptions. It only takes effect when a consumer group is configured; without a group ID the value is validated but unused.

This option is consumer-only and has no effect on a Producer; passing it to NewProducer is silently ignored.

type Producer 

type Producer struct {
	// contains filtered or unexported fields
}

Producer publishes messages to a Kafka topic with pluggable encoding.

func NewProducer 

func NewProducer(brokers []string, topic string, opts ...Option) (*Producer, error)

NewProducer constructs a Kafka producer for a topic with optional tuning (encoding, balancing).

The broker list and topic are validated at construction time: an empty or blank-entry broker list, an empty topic, a negative batch size, a non-positive batch timeout, and an unknown required-acks value are all rejected with errors matching ErrInvalidOptions (or ErrNilEncodeFunc for a nil encoder); match them with errors.Is.

The partitioning strategy defaults to a kafka.Hash balancer. Because Send() publishes messages without a Key, the default Hash balancer concentrates all messages on a single partition; use WithBalancer to choose a different strategy (for example a round-robin balancer) to spread keyless messages across partitions.

Broker acknowledgment defaults to kafka.RequireAll, so Send()/SendData() return an error when the write is not acknowledged by the full in-sync replica set; use WithRequiredAcks to trade durability for throughput.

The batch flush timeout defaults to 10ms (instead of the kafka-go 1s default) to keep the latency of synchronous per-message Send() calls low; use WithBatchTimeout and WithBatchSize to tune batching behavior.

The consumer-only options WithSessionTimeout and WithFirstOffset are accepted for API compatibility on the shared Option type but have no effect on a Producer.

func (*Producer) Close 

func (p *Producer) Close() error

Close releases Producer's resources and closes the broker connection. It is safe to call multiple times.

func (*Producer) HealthCheck 

func (p *Producer) HealthCheck(ctx context.Context) error

HealthCheck verifies broker reachability by probing each configured broker until one succeeds. When every broker fails, the individual probe errors are joined into the returned error.

The default probe performs a partition lookup over plaintext TCP with the default kafka-go dialer, so it does not exercise TLS or SASL and always performs real network I/O even when a writer is injected via WithKafkaWriter. Override it with WithBrokerCheckFunc to customize the dialer or to make HealthCheck deterministic in tests.

func (*Producer) Send 

func (p *Producer) Send(ctx context.Context, msg []byte) error

Send publishes a raw byte message to the Kafka topic.

func (*Producer) SendData 

func (p *Producer) SendData(ctx context.Context, data any) error

SendData encodes the data argument and publishes the result to the Kafka topic using the configured encoder.

type TDecodeFunc 

type TDecodeFunc func(ctx context.Context, msg []byte, data any) error

TDecodeFunc is the type of function used to replace the default message decoding function used by ReceiveData().

type TEncodeFunc 

type TEncodeFunc func(ctx context.Context, data any) ([]byte, error)

TEncodeFunc is the type of function used to replace the default message encoding function used by SendData().

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.