Help us shape the future of Go! Take the 2026 Go Developer Survey

xkafka

package module
v0.5.1 Latest Latest
Warning

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

 Go to latest Published: Jul 4, 2026 License: MIT Imports: 18 Imported by: 0

README

xkafka

Lightweight Kafka wrapper for Go, built on top of franz-go.

Go Version

xkafka wraps the excellent franz-go client with a compact API for common Kafka workflows: producing messages, consuming records through handlers, committing offsets after successful processing, using Kafka transactions, building Kafka-to-Kafka exactly-once processing with GroupTransactSession, and exposing Kafka observability with OpenTelemetry and Prometheus.

Explore ready-to-run use cases in examples.

Features

  • Client: Unified client for both producing and consuming.
  • Producing: Synchronous, asynchronous, and transactional publishing.
  • Consuming: Batch handlers for regular consumer groups.
  • Share Groups: Batch consumption with flexible Ack management.
  • EOS Processing: Exactly-once Kafka-to-Kafka processing via GroupTransactSession.
  • Commits: Safe offset commits executing only on success.
  • Observability: OpenTelemetry and Prometheus support.
  • Security: Native TLS and SASL (PLAIN, SCRAM-SHA-256/512).
  • Configuration: Setup via Go structs or environment variables.

Installation

This repository contains the core xkafka module. The core package is released from the repository root:

go get github.com/mkbeh/xkafka

Quick start

These example show basic producing and consuming flows. For production workloads, tune retries, offset commits, security, and observability for your needs.

ctx := context.Background()

// Initialize the unified Kafka client.
client, err := xkafka.NewClient(
	xkafka.WithConfig(&xkafka.Config{
		Brokers:             "localhost:9092",
		DefaultProduceTopic: "orders.created",
		Topics:              "orders.created",
		Group:               "orders-worker-group",
	}),
	xkafka.WithConsumerBatchHandler(func(ctx context.Context, records []*kgo.Record) error {
		for _, record := range records {
			fmt.Printf("received: topic=%s key=%s value=%s\n", record.Topic, record.Key, record.Value)
		}

		return nil // Returning nil commits offsets after successful batch processing.
	}),
)
if err != nil {
	log.Fatal(err)
}
defer func() {
	if err := client.Shutdown(ctx); err != nil {
		log.Printf("shutdown failed: %v", err)
	}
}()

// Start the blocking consumer loop in a separate goroutine.
go func() {
	if err := client.HandleFetches(ctx); err != nil {
		log.Printf("consumer stopped: %v", err)
	}
}()

// Publish a synchronous message using DefaultProduceTopic.
if err := client.ProduceSync(ctx, &kgo.Record{
	Key:   []byte("order-1"),
	Value: []byte("created"),
}); err != nil {
	log.Fatal(err)
}

[!IMPORTANT] Records are considered processed only when the handler returns nil. If the handler returns an error, offsets are not committed and processing resumes after SUSPEND_PROCESSING_TIMEOUT.

Transactions

To enable transactional publishing, configure TransactionalID and use RunInTx.

client, err := xkafka.NewClient(
	xkafka.WithConfig(&xkafka.Config{
		Brokers:         "localhost:9092",
		TransactionalID: "orders-tx-producer",
	}),
)
if err != nil {
	log.Fatal(err)
}
defer func() {
	if err := client.Shutdown(context.Background()); err != nil {
		log.Printf("shutdown failed: %v", err)
	}
}()

if err := client.RunInTx(ctx, func(ctx context.Context, tx *xkafka.Tx) error {
	if err := tx.ProduceSync(ctx, &kgo.Record{
		Topic: "orders.created",
		Key:   []byte("order-1"),
		Value: []byte("created"),
	}); err != nil {
		return err
	}

	if err := tx.ProduceSync(ctx, &kgo.Record{
		Topic: "audit.events",
		Key:   []byte("order-1"),
		Value: []byte("audited" ),
	}); err != nil {
		return err
	}

	return nil // Commits automatically. Any returned error or panic aborts the transaction.
}); err != nil {
	log.Fatal(err)
}

[!NOTE] RunInTx automatically commits on nil and aborts on errors or panics. To prevent consumers from reading aborted records, enable kgo.ReadCommitted() using WithFetchIsolationLevel.

Share Groups

Use WithShareGroupBatchHandler with ShareGroup to consume through Kafka Share Groups.

client, err := xkafka.NewClient(
	xkafka.WithConfig(&xkafka.Config{
		Brokers: "localhost:9092",
		Topics:  "orders.created",

		ShareGroup:      "orders-share-group",
		ShareMaxRecords: 5,
		MaxPollRecords:  5,

		ShareRejectAfterDeliveries: 3,
		ShareReleaseTimeout:        5 * time.Second,
	}),
	xkafka.WithShareGroupBatchHandler(func(ctx context.Context, records []*kgo.Record) error {
		for _, record := range records {
			fmt.Printf("share record: topic=%s value=%s\n", record.Topic, record.Value)
		}

		return nil
	}),
)
if err != nil {
	log.Fatal(err)
}
defer client.Shutdown(ctx)

if err := client.HandleFetches(ctx); err != nil {
	log.Fatal(err)
}

[!NOTE] Share Group Ack Rules: Successful handlers trigger AckAccept, while errors trigger AckRelease. If the delivery limit is reached, records are marked as AckReject (if ShareRejectAfterDeliveries > 0), otherwise they continue to be released via AckRelease.

Exactly-Once Semantics

Use GroupTransactSession for exactly-once (Kafka-to-Kafka) processing. It commits consumed offsets and produced records atomically in the same Kafka transaction.

session, err := xkafka.NewGroupTransactSession(
	xkafka.WithConfig(&xkafka.Config{
		Brokers:         "localhost:9092",
		Topics:          "orders.input",
		Group:           "orders-eos-group",
		TransactionalID: "orders-eos-session",
	}),
	xkafka.WithGroupTransactSessionBatchHandler(
		func(ctx context.Context, records []*kgo.Record, tx *xkafka.Tx) error {
			for _, record := range records {
				if err := tx.ProduceSync(ctx, &kgo.Record{
					Topic: "orders.output",
					Key:   record.Key,
					Value: record.Value,
				}); err != nil {
					return err
				}
			}

			return nil // Commits consumed offsets and produced records atomically.
		},
	),
)
if err != nil {
	log.Fatal(err)
}
defer func() {
	if err := session.Shutdown(ctx); err != nil {
		log.Printf("shutdown failed: %v", err)
	}
}()

if err := session.HandleFetches(ctx); err != nil {
	log.Fatal(err)
}

[!NOTE] If the handler returns an error, the group transaction is aborted and consumed offsets are not committed.

Observability

xkafka uses franz-go hooks for Kafka client instrumentation and adds wrapper-level metrics for producer, consumer, share group, and transaction workflows.

  • OpenTelemetry: Metrics and distributed tracing via franz-go hooks.
  • Prometheus: Wrapper-level metrics for produce, consume, share group, and transaction operations.
Configuration
xkafka.WithMeterProvider(meterProvider)
xkafka.WithTracerProvider(tracerProvider)
xkafka.WithTracerPropagator(propagator)
xkafka.WithMetricsNamespace("orders")
xkafka.WithMetricLabel("service", "orders-api")

xkafka does not depend on a specific tracing backend. Provide an OpenTelemetry TracerProvider through WithTracerProvider, and export traces using your application or OpenTelemetry Collector pipeline.

A runnable tracing example is available in examples/tracing.

Metric naming

Prometheus metrics follow the [<namespace>_]kafka_<metric_name> layout. For example, using the orders namespace:

orders_kafka_produce_errors_total
orders_kafka_consume_handle_duration_seconds
orders_kafka_consume_errors_total
orders_kafka_transactions_total
orders_kafka_transaction_duration_seconds

[!NOTE] Low-level franz-go client metrics and traces are exported through OpenTelemetry hooks. xkafka adds wrapper-level Prometheus metrics around producer, consumer, share group, and transaction workflows.

For the full list of exported Prometheus metrics, see internal/pkg/kprom/metrics.go.

Configuration

Config can be initialized directly as a Go struct or populated from environment variables by your application configuration layer.

Config uses env tags and can be loaded with caarlos0/env. See examples/env.

Environment variables
Variable Default Description
BROKERS Comma-separated seed brokers
SASL_MECHANISM PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512
USER SASL username
PASSWORD SASL password
REQUEST_TIMEOUT_OVERHEAD Request deadline overhead
REQUEST_RETRIES Max request retries
RETRY_TIMEOUT Total retry time limit
DIAL_TIMEOUT Broker dial timeout
CONN_IDLE_TIMEOUT Idle connection timeout
METADATA_MAX_AGE Max age of cached metadata
METADATA_MIN_AGE Min time between metadata refreshes
MAX_WRITE_BYTES Max bytes per connection write
MAX_READ_BYTES Max bytes per broker response
ALWAYS_RETRY_EOF false Retry EOF errors instead of failing connection
DEFAULT_PRODUCE_TOPIC Fallback topic if record topic is empty
PRODUCER_BATCH_MAX_BYTES Max size of a producer batch
MAX_BUFFERED_RECORDS Max buffered records before blocking
MAX_BUFFERED_BYTES Max buffered bytes before blocking
PRODUCE_REQUEST_TIMEOUT Broker response timeout for produce requests
RECORD_RETRIES Max record-level produce retries
RECORD_DELIVERY_TIMEOUT Max record buffering time
PRODUCER_LINGER Linger delay for batch building
TRANSACTIONAL_ID Transactional identifier for EOS
TRANSACTION_TIMEOUT Max transaction duration
ENABLED true Enable consumer loop
TOPICS Comma-separated topics to consume
GROUP Consumer group ID
MAX_POLL_RECORDS 100 Max records per poll
POLL_INTERVAL 1s Interval between polls
SKIP_FATAL_ERRORS true Continue on non-retryable fetch errors
SUSPEND_PROCESSING_TIMEOUT 30s Backoff delay after handler error
SUSPEND_COMMITTING_TIMEOUT 10s Backoff delay after commit/ack error
INSTANCE_ID Static group membership ID
CONSUME_REGEX false Treat topics as regular expressions
DISABLE_FETCH_SESSIONS false Disable fetch sessions
RACK Rack ID for rack-aware fetching
MAX_CONCURRENT_FETCHES Max concurrent fetches buffered by client
SESSION_TIMEOUT Rebalance session timeout
REBALANCE_TIMEOUT Max time for members to rejoin
HEARTBEAT_INTERVAL Heartbeat interval
FETCH_MAX_WAIT Max broker wait time for fetches
FETCH_MIN_BYTES Min bytes broker accumulates before response
FETCH_MAX_BYTES Max bytes per fetch response
FETCH_MAX_PARTITION_BYTES Max bytes per partition fetch
SHARE_GROUP Share group ID
SHARE_MAX_RECORDS Max records per share fetch
SHARE_MAX_RECORDS_STRICT false Strictly cap records per share fetch
SHARE_REJECT_AFTER_DELIVERIES Delivery limit before triggering AckReject
SHARE_RELEASE_TIMEOUT Backoff delay before releasing failed records

License

This project is licensed under the MIT License.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type BatchHandlerFunc 

type BatchHandlerFunc func(ctx context.Context, records []*kgo.Record) error

BatchHandlerFunc handles a batch of consumed Kafka records.

type BatchTxHandlerFunc added in v0.3.1 

type BatchTxHandlerFunc func(ctx context.Context, records []*kgo.Record, tx *Tx) error

BatchTxHandlerFunc handles a batch of consumed Kafka records inside a group transaction.

type Client added in v0.3.1 

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

Client provides Kafka produce and consume operations.

func NewClient added in v0.3.1 

func NewClient(opts ...Opt) (*Client, error)

NewClient creates a Kafka client with the provided options.

func (*Client) HandleFetches added in v0.3.1 

func (c *Client) HandleFetches(ctx context.Context) error

func (*Client) Ping added in v0.4.4 

func (c *Client) Ping(ctx context.Context) error

func (*Client) Produce added in v0.3.1 

func (c *Client) Produce(ctx context.Context, record *kgo.Record, promise PromiseFunc)

func (*Client) ProduceSync added in v0.3.1 

func (c *Client) ProduceSync(ctx context.Context, records ...*kgo.Record) error

func (*Client) RunInTx added in v0.3.1 

func (c *Client) RunInTx(ctx context.Context, fn TxFunc) (err error)

RunInTx executes fn inside a Kafka transaction.

The transaction is committed when fn returns nil. It is aborted when fn returns an error. Panics are recovered only long enough to abort the transaction, then re-thrown.

func (*Client) Shutdown added in v0.3.1 

func (c *Client) Shutdown(ctx context.Context) error

Shutdown stops polling, flushes pending records and acks, and closes the underlying client.

func (*Client) TryProduce added in v0.3.1 

func (c *Client) TryProduce(ctx context.Context, record *kgo.Record, promise PromiseFunc)

type Config added in v0.3.1 

type Config struct {

	// Brokers sets the seed brokers for the client to use.
	//
	// Any seed without a port uses the default Kafka port 9092.
	Brokers string

	// SASLMechanism is the SASL mechanism used for authentication.
	// Supported values: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512.
	SASLMechanism string

	// User is the SASL username.
	User string

	// Password is the SASL password.
	Password string

	// RequestTimeoutOverhead adds extra time while setting request deadlines.
	RequestTimeoutOverhead time.Duration

	// ConnIdleTimeout controls how long idle broker connections are kept open.
	ConnIdleTimeout time.Duration

	// DialTimeout sets the timeout for opening broker connections.
	DialTimeout time.Duration

	// RequestRetries sets how many times retryable requests may be retried.
	RequestRetries *int

	// RetryTimeout limits the total time allowed for retrying a request.
	RetryTimeout time.Duration

	// BrokerMaxWriteBytes limits how many bytes may be written to a broker in one write.
	BrokerMaxWriteBytes *int32

	// BrokerMaxReadBytes limits the maximum response size read from a broker.
	BrokerMaxReadBytes *int32

	// MetadataMaxAge sets the maximum age of cached Kafka metadata.
	MetadataMaxAge time.Duration

	// MetadataMinAge sets the minimum time between metadata refreshes.
	MetadataMinAge time.Duration

	// AlwaysRetryEOF retries EOF errors instead of treating them as terminal connection failures.
	AlwaysRetryEOF bool

	// DefaultProduceTopic sets the default topic for records without an explicit topic.
	DefaultProduceTopic string

	// ProducerBatchMaxBytes limits the maximum size of a produced record batch.
	ProducerBatchMaxBytes int32

	// MaxBufferedRecords limits how many records the producer may buffer.
	MaxBufferedRecords int

	// MaxBufferedBytes limits how many bytes the producer may buffer.
	MaxBufferedBytes int

	// ProduceRequestTimeout limits how long brokers may take to respond to produce requests.
	ProduceRequestTimeout time.Duration

	// RecordRetries sets how many times producing records may be retried.
	RecordRetries int

	// ProducerLinger sets how long records may wait for batching before a produce request is built.
	ProducerLinger *time.Duration

	// RecordDeliveryTimeout limits how long a record may remain buffered before timing out.
	RecordDeliveryTimeout time.Duration

	// TransactionalID identifies the transactional producer.
	//
	// It is used by producer transactions and group transact sessions.
	TransactionalID string

	// TransactionTimeout limits how long a Kafka transaction may remain open.
	TransactionTimeout time.Duration

	// ConsumeRegex treats configured topics as regular expressions.
	ConsumeRegex bool

	// Topics contains Kafka topics consumed by the consumer or group transact session.
	Topics string

	// MaxPollRecords limits how many records are handled in one poll iteration.
	MaxPollRecords int

	// FetchMaxWait limits how long a broker may wait before returning a fetch response.
	FetchMaxWait time.Duration

	// FetchMaxBytes limits the maximum bytes returned in one fetch response.
	FetchMaxBytes int32

	// FetchMinBytes sets the minimum bytes a broker tries to return in one fetch response.
	FetchMinBytes int32

	// FetchMaxPartitionBytes limits how many bytes may be fetched from one partition.
	FetchMaxPartitionBytes int32

	// DisableFetchSessions disables Kafka fetch sessions.
	DisableFetchSessions bool

	// Rack identifies the client rack for rack-aware fetching and consumer assignment.
	Rack string

	// MaxConcurrentFetches limits how many fetch requests may be in flight or buffered at once.
	//
	// It can be used together with FetchMaxBytes to bound consumer memory usage.
	MaxConcurrentFetches *int

	// Group identifies the Kafka consumer group.
	Group string

	// InstanceID sets a static consumer group member ID.
	InstanceID string

	// SessionTimeout controls how long a group member may go without heartbeating.
	SessionTimeout time.Duration

	// RebalanceTimeout controls how long group members may take during a rebalance.
	RebalanceTimeout time.Duration

	// HeartbeatInterval controls how often a group member sends heartbeats.
	HeartbeatInterval time.Duration

	// ShareGroup identifies the Kafka share group.
	//
	// If set, share group consuming is used instead of a regular consumer group.
	ShareGroup string

	// ShareMaxRecords limits how many records ShareFetch may return.
	//
	// If zero, the franz-go default is used.
	ShareMaxRecords int32

	// ShareMaxRecordsStrict asks the broker to strictly cap records per ShareFetch.
	ShareMaxRecordsStrict bool

	// ShareRejectAfterDeliveries controls when failed share group records are rejected.
	//
	// If zero, failed records are released for redelivery.
	ShareRejectAfterDeliveries int32

	// ShareReleaseTimeout delays AckRelease after share handler errors.
	//
	// If zero, failed records are released immediately.
	ShareReleaseTimeout time.Duration

	// Enabled enables this Kafka component.
	Enabled bool

	// SkipFatalErrors allows the consumer to continue after non-retriable fetch errors.
	SkipFatalErrors bool

	// PollInterval sets the interval between consumer poll iterations.
	PollInterval time.Duration

	// SuspendProcessingTimeout sets the wait time after handler errors.
	SuspendProcessingTimeout time.Duration

	// SuspendCommittingTimeout sets the wait time after offset commit or ack errors.
	SuspendCommittingTimeout time.Duration
}

type ConsumerOpt added in v0.3.1 

type ConsumerOpt interface {
	Opt
	// contains filtered or unexported methods
}

ConsumerOpt is a consumer-specific option to configure a client.

This is simply a namespaced Opt.

func WithConsumeResetOffset 

func WithConsumeResetOffset(offset kgo.Offset) ConsumerOpt

func WithConsumerBatchHandler 

func WithConsumerBatchHandler(handler BatchHandlerFunc) ConsumerOpt

func WithFetchIsolationLevel 

func WithFetchIsolationLevel(level kgo.IsolationLevel) ConsumerOpt

func WithGroupTransactSessionBatchHandler added in v0.3.1 

func WithGroupTransactSessionBatchHandler(handler BatchTxHandlerFunc) ConsumerOpt

func WithShareGroupBatchHandler added in v0.4.2 

func WithShareGroupBatchHandler(handler BatchHandlerFunc) ConsumerOpt

type GroupOpt added in v0.3.1 

type GroupOpt interface {
	Opt
	// contains filtered or unexported methods
}

GroupOpt is a consumer group-specific option to configure a client.

This is simply a namespaced Opt.

func WithBalancers added in v0.3.1 

func WithBalancers(balancers ...kgo.GroupBalancer) GroupOpt

func WithShareAckCallback added in v0.3.1 

func WithShareAckCallback(fn func(*kgo.Client, kgo.ShareAckResults)) GroupOpt

type GroupTransactSession added in v0.3.1 

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

GroupTransactSession consumes records and produces records in the same Kafka transaction.

It is intended for Kafka-to-Kafka consume-process-produce flows where consumed offsets and produced records must be committed atomically.

func NewGroupTransactSession added in v0.3.1 

func NewGroupTransactSession(opts ...Opt) (*GroupTransactSession, error)

NewGroupTransactSession creates a Kafka group transaction session.

func (*GroupTransactSession) HandleFetches added in v0.3.1 

func (g *GroupTransactSession) HandleFetches(ctx context.Context) error

func (*GroupTransactSession) Ping added in v0.4.4 

func (g *GroupTransactSession) Ping(ctx context.Context) error

func (*GroupTransactSession) Produce added in v0.3.1 

func (g *GroupTransactSession) Produce(ctx context.Context, record *kgo.Record, promise PromiseFunc)

func (*GroupTransactSession) ProduceSync added in v0.3.1 

func (g *GroupTransactSession) ProduceSync(ctx context.Context, records ...*kgo.Record) error

func (*GroupTransactSession) Shutdown added in v0.3.1 

func (g *GroupTransactSession) Shutdown(_ context.Context) error

Shutdown stops polling and closes the group transaction session.

func (*GroupTransactSession) TryProduce added in v0.3.1 

func (g *GroupTransactSession) TryProduce(ctx context.Context, record *kgo.Record, promise PromiseFunc)

type Opt added in v0.3.1 

type Opt interface {
	// contains filtered or unexported methods
}

func WithClientID added in v0.3.1 

func WithClientID(s string) Opt

func WithConfig added in v0.3.1 

func WithConfig(config *Config) Opt

func WithLogger added in v0.3.1 

func WithLogger(logger kgo.Logger) Opt

func WithMeterProvider added in v0.3.1 

func WithMeterProvider(provider metric.MeterProvider) Opt

func WithMetricLabel added in v0.3.1 

func WithMetricLabel(key, value string) Opt

func WithMetricsNamespace added in v0.3.1 

func WithMetricsNamespace(namespace string) Opt

func WithTLS added in v0.3.1 

func WithTLS(tlsConfig *tls.Config) Opt

func WithTracerPropagator added in v0.3.1 

func WithTracerPropagator(propagator propagation.TextMapPropagator) Opt

func WithTracerProvider added in v0.3.1 

func WithTracerProvider(provider trace.TracerProvider) Opt

type ProducerOpt added in v0.3.1 

type ProducerOpt interface {
	Opt
	// contains filtered or unexported methods
}

ProducerOpt is a producer-specific option to configure a client.

This is simply a namespaced Opt.

func WithAllowIdempotentProduceCancellation added in v0.3.1 

func WithAllowIdempotentProduceCancellation() ProducerOpt

func WithProducePromise added in v0.3.1 

func WithProducePromise(promiseFunc PromiseFunc) ProducerOpt

func WithProducerBatchCompression 

func WithProducerBatchCompression(preference ...kgo.CompressionCodec) ProducerOpt

func WithProducerBatchMaxBytesFn added in v0.3.1 

func WithProducerBatchMaxBytesFn(fn func(topic string) int32) ProducerOpt

func WithRecordPartitioner added in v0.3.1 

func WithRecordPartitioner(partitioner kgo.Partitioner) ProducerOpt

func WithRequiredAcks added in v0.3.1 

func WithRequiredAcks(acks kgo.Acks) ProducerOpt

type PromiseFunc added in v0.4.2 

type PromiseFunc func(record *kgo.Record, err error)

PromiseFunc is called when an asynchronous produce request completes.

type Tx added in v0.4.0 

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

Tx exposes producing operations inside a Kafka transaction.

func (*Tx) Produce added in v0.4.0 

func (tx *Tx) Produce(ctx context.Context, record *kgo.Record, promise PromiseFunc)

Produce asynchronously produces a record inside the current Kafka transaction.

func (*Tx) ProduceSync added in v0.4.0 

func (tx *Tx) ProduceSync(ctx context.Context, records ...*kgo.Record) error

ProduceSync synchronously produces records inside the current Kafka transaction.

func (*Tx) TryProduce added in v0.4.0 

func (tx *Tx) TryProduce(ctx context.Context, record *kgo.Record, promise PromiseFunc)

TryProduce attempts to enqueue a record inside the current Kafka transaction without blocking on producer backpressure.

type TxFunc added in v0.3.1 

type TxFunc func(ctx context.Context, tx *Tx) error

TxFunc handles work inside a Kafka transaction.

Source Files

View all Source files

Directories

Path Synopsis
extra
otelxkafka module
internal
kprom

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.