xkafka

README

xkafka

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

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

go get github.com/mkbeh/xkafka

Quick start

These examples 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{
		Enabled:             true,
		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 KAFKA_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) error {
	if err := client.ProduceSync(ctx, &kgo.Record{
		Topic: "orders.created",
		Key:   []byte("order-1"),
		Value: []byte("created"),
	}); err != nil {
		return err
	}

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

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

[!NOTE] RunInTx executes your function inside a Kafka transaction. It commits automatically on nil and aborts on any returned error or panic.

Consumers that must ignore aborted transactional records should use kgo.ReadCommitted() through WithFetchIsolationLevel.

Share Groups

Use WithConsumerShareBatchHandler with ShareGroup to consume through Kafka Share Groups.

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

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

		ShareRejectAfterDeliveries: 3,
		ShareReleaseTimeout:        5 * time.Second,
	}),
	xkafka.WithConsumerShareBatchHandler(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

handler success - AckAccept

handler error - AckRelease

delivery limit - AckReject if ShareRejectAfterDeliveries > 0; otherwise records keep being released with 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{
		Enabled:         true,
		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.GroupTransactSession) 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 Options

xkafka.WithMeterProvider(meterProvider)
xkafka.WithTracerProvider(tracerProvider)
xkafka.WithTracerPropagator(propagator)
xkafka.WithMetricsNamespace("orders")
xkafka.WithMetricLabel("service", "orders-api")

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
Common
KAFKA_BROKERS	—	Comma-separated seed brokers
KAFKA_SASL_MECHANISM	—	PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512
KAFKA_USER	—	SASL authentication username
KAFKA_PASSWORD	—	SASL authentication password
KAFKA_REQUEST_TIMEOUT_OVERHEAD	—	Extra time added while setting request deadlines
KAFKA_REQUEST_RETRIES	—	Maximum number of request retries
KAFKA_RETRY_TIMEOUT	—	Total retry time limit for requests
KAFKA_DIAL_TIMEOUT	—	Broker dial timeout
KAFKA_CONN_IDLE_TIMEOUT	—	Idle connection timeout
KAFKA_METADATA_MAX_AGE	—	Maximum age of cached metadata
KAFKA_METADATA_MIN_AGE	—	Minimum time between metadata refreshes
KAFKA_MAX_WRITE_BYTES	—	Maximum bytes written per connection write
KAFKA_MAX_READ_BYTES	—	Maximum bytes read from a broker response
KAFKA_ALWAYS_RETRY_EOF	false	Retry EOF errors instead of treating them as terminal connection failures
Producer
KAFKA_DEFAULT_PRODUCE_TOPIC	—	Fallback topic if kgo.Record.Topic is empty
KAFKA_PRODUCER_BATCH_MAX_BYTES	—	Maximum size of a producer batch
KAFKA_MAX_BUFFERED_RECORDS	—	Maximum buffered records before producing blocks
KAFKA_MAX_BUFFERED_BYTES	—	Maximum buffered bytes before producing blocks
KAFKA_PRODUCE_REQUEST_TIMEOUT	—	Broker response timeout for produce requests
KAFKA_RECORD_RETRIES	—	Number of record-level produce retries
KAFKA_RECORD_DELIVERY_TIMEOUT	—	Maximum buffer time for a record before timeout
KAFKA_PRODUCER_LINGER	—	Delay used to wait for more records before building a producer batch
KAFKA_TRANSACTIONAL_ID	—	Unique identifier to enable transactional producing
KAFKA_TRANSACTION_TIMEOUT	—	Maximum allowed transaction duration
Consumer
KAFKA_ENABLED	true	Enables the consumer loop
KAFKA_TOPICS	—	Comma-separated list of topics to consume
KAFKA_GROUP	—	Consumer group ID required for offset commits
KAFKA_MAX_POLL_RECORDS	100	Maximum records handled per poll iteration
KAFKA_POLL_INTERVAL	1s	Interval between poll iterations
KAFKA_SKIP_FATAL_ERRORS	true	Continue after non-retryable fetch errors
KAFKA_SUSPEND_PROCESSING_TIMEOUT	30s	Backoff delay after a handler error
KAFKA_SUSPEND_COMMITTING_TIMEOUT	10s	Backoff delay after a commit or ack error
KAFKA_INSTANCE_ID	—	Static group membership identifier
KAFKA_CONSUME_REGEX	false	Treat configured topics as regular expressions
KAFKA_DISABLE_FETCH_SESSIONS	false	Disable Kafka fetch sessions
KAFKA_RACK	—	Rack identifier for rack-aware fetching
KAFKA_MAX_CONCURRENT_FETCHES	—	Maximum concurrent fetches buffered by the client
KAFKA_SESSION_TIMEOUT	—	Maximum time between heartbeats before rebalance
KAFKA_REBALANCE_TIMEOUT	—	Maximum time for members to rejoin on rebalance
KAFKA_HEARTBEAT_INTERVAL	—	Heartbeat interval
KAFKA_FETCH_MAX_WAIT	—	Maximum broker wait time for incomplete fetches
KAFKA_FETCH_MIN_BYTES	—	Minimum bytes a broker tries to accumulate before responding
KAFKA_FETCH_MAX_BYTES	—	Maximum bytes per fetch response
KAFKA_FETCH_MAX_PARTITION_BYTES	—	Maximum bytes per partition fetch
Share Groups
KAFKA_SHARE_GROUP	—	Share group identifier
KAFKA_SHARE_MAX_RECORDS	—	Maximum records returned per share fetch
KAFKA_SHARE_MAX_RECORDS_STRICT	false	Strictly cap records per share fetch
KAFKA_SHARE_REJECT_AFTER_DELIVERIES	—	Delivery limit before triggering AckReject
KAFKA_SHARE_RELEASE_TIMEOUT	—	Backoff delay before releasing failed records

License

This project is licensed under the MIT License.

Documentation

Index

• type BatchHandlerFunc
• type BatchTxHandlerFunc
• type Client
  • func NewClient(opts ...Opt) (*Client, error)
  • func (c *Client) HandleFetches(ctx context.Context) error
  • func (c *Client) Produce(ctx context.Context, record *kgo.Record, promise func(*kgo.Record, error))
  • func (c *Client) ProduceSync(ctx context.Context, records ...*kgo.Record) error
  • func (c *Client) RunInTx(ctx context.Context, fn TxFunc) (err error)
  • func (c *Client) Shutdown(ctx context.Context) error
  • func (c *Client) TryProduce(ctx context.Context, record *kgo.Record, promise func(*kgo.Record, error))
• type Config
• type ConsumerOpt
  • func WithConsumeResetOffset(offset kgo.Offset) ConsumerOpt
  • func WithConsumerBatchHandler(handler BatchHandlerFunc) ConsumerOpt
  • func WithConsumerShareBatchHandler(handler BatchHandlerFunc) ConsumerOpt
  • func WithFetchIsolationLevel(level kgo.IsolationLevel) ConsumerOpt
  • func WithGroupTransactSessionBatchHandler(handler BatchTxHandlerFunc) ConsumerOpt
• type GroupOpt
  • func WithBalancers(balancers ...kgo.GroupBalancer) GroupOpt
  • func WithShareAckCallback(fn func(*kgo.Client, kgo.ShareAckResults)) GroupOpt
• type GroupTransactSession
  • func NewGroupTransactSession(opts ...Opt) (*GroupTransactSession, error)
  • func (g *GroupTransactSession) HandleFetches(ctx context.Context) error
  • func (g *GroupTransactSession) Produce(ctx context.Context, record *kgo.Record, promise func(*kgo.Record, error))
  • func (g *GroupTransactSession) ProduceSync(ctx context.Context, records ...*kgo.Record) error
  • func (g *GroupTransactSession) Shutdown(_ context.Context) error
  • func (g *GroupTransactSession) TryProduce(ctx context.Context, record *kgo.Record, promise func(*kgo.Record, error))
• type Opt
  • func WithClientID(s string) Opt
  • func WithConfig(config *Config) Opt
  • func WithLogger(logger *slog.Logger) Opt
  • func WithMeterProvider(provider metric.MeterProvider) Opt
  • func WithMetricLabel(key, value string) Opt
  • func WithMetricsNamespace(namespace string) Opt
  • func WithTLS(tlsConfig *tls.Config) Opt
  • func WithTracerPropagator(propagator propagation.TextMapPropagator) Opt
  • func WithTracerProvider(provider trace.TracerProvider) Opt
• type ProducePromiseFunc
• type ProducerOpt
  • func WithAllowIdempotentProduceCancellation() ProducerOpt
  • func WithProducePromise(promiseFunc ProducePromiseFunc) ProducerOpt
  • func WithProducerBatchCompression(preference ...kgo.CompressionCodec) ProducerOpt
  • func WithProducerBatchMaxBytesFn(fn func(topic string) int32) ProducerOpt
  • func WithRecordPartitioner(partitioner kgo.Partitioner) ProducerOpt
  • func WithRequiredAcks(acks kgo.Acks) ProducerOpt
• type TxFunc

Types

type BatchHandlerFunc

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

BatchHandlerFunc handles a batch of consumed Kafka records.

type BatchTxHandlerFunc

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

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

type Client

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

Client is a high-level Kafka client for producing records and consuming records through handlers.

func NewClient

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

NewClient creates a Kafka client with the provided options and verifies broker connectivity.

func (*Client) HandleFetches

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

func (*Client) Produce

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

func (*Client) ProduceSync

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

func (*Client) RunInTx

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

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

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

func (*Client) TryProduce

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

type Config

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 `env:"KAFKA_BROKERS"`

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

	// User is the SASL username.
	User string `env:"KAFKA_USER"`

	// Password is the SASL password.
	Password string `env:"KAFKA_PASSWORD"`

	// RequestTimeoutOverhead adds extra time while setting request deadlines.
	RequestTimeoutOverhead time.Duration `env:"KAFKA_REQUEST_TIMEOUT_OVERHEAD"`

	// ConnIdleTimeout controls how long idle broker connections are kept open.
	ConnIdleTimeout time.Duration `env:"KAFKA_CONN_IDLE_TIMEOUT"`

	// DialTimeout sets the timeout for opening broker connections.
	DialTimeout time.Duration `env:"KAFKA_DIAL_TIMEOUT"`

	// RequestRetries sets how many times retryable requests may be retried.
	RequestRetries *int `env:"KAFKA_REQUEST_RETRIES"`

	// RetryTimeout limits the total time allowed for retrying a request.
	RetryTimeout time.Duration `env:"KAFKA_RETRY_TIMEOUT"`

	// BrokerMaxWriteBytes limits how many bytes may be written to a broker in one write.
	BrokerMaxWriteBytes *int32 `env:"KAFKA_MAX_WRITE_BYTES"`

	// BrokerMaxReadBytes limits the maximum response size read from a broker.
	BrokerMaxReadBytes *int32 `env:"KAFKA_MAX_READ_BYTES"`

	// MetadataMaxAge sets the maximum age of cached Kafka metadata.
	MetadataMaxAge time.Duration `env:"KAFKA_METADATA_MAX_AGE"`

	// MetadataMinAge sets the minimum time between metadata refreshes.
	MetadataMinAge time.Duration `env:"KAFKA_METADATA_MIN_AGE"`

	// AlwaysRetryEOF retries EOF errors instead of treating them as terminal connection failures.
	AlwaysRetryEOF bool `env:"KAFKA_ALWAYS_RETRY_EOF"`

	// DefaultProduceTopic sets the default topic for records without an explicit topic.
	DefaultProduceTopic string `env:"KAFKA_DEFAULT_PRODUCE_TOPIC"`

	// ProducerBatchMaxBytes limits the maximum size of a produced record batch.
	ProducerBatchMaxBytes int32 `env:"KAFKA_PRODUCER_BATCH_MAX_BYTES"`

	// MaxBufferedRecords limits how many records the producer may buffer.
	MaxBufferedRecords int `env:"KAFKA_MAX_BUFFERED_RECORDS"`

	// MaxBufferedBytes limits how many bytes the producer may buffer.
	MaxBufferedBytes int `env:"KAFKA_MAX_BUFFERED_BYTES"`

	// ProduceRequestTimeout limits how long brokers may take to respond to produce requests.
	ProduceRequestTimeout time.Duration `env:"KAFKA_PRODUCE_REQUEST_TIMEOUT"`

	// RecordRetries sets how many times producing records may be retried.
	RecordRetries int `env:"KAFKA_RECORD_RETRIES"`

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

	// RecordDeliveryTimeout limits how long a record may remain buffered before timing out.
	RecordDeliveryTimeout time.Duration `env:"KAFKA_RECORD_DELIVERY_TIMEOUT"`

	// TransactionalID identifies the transactional producer.
	//
	// It is used by producer transactions and group transact sessions.
	TransactionalID string `env:"KAFKA_TRANSACTIONAL_ID"`

	// TransactionTimeout limits how long a Kafka transaction may remain open.
	TransactionTimeout time.Duration `env:"KAFKA_TRANSACTION_TIMEOUT"`

	// ConsumeRegex treats configured topics as regular expressions.
	ConsumeRegex bool `env:"KAFKA_CONSUME_REGEX"`

	// Topics contains Kafka topics consumed by the consumer or group transact session.
	Topics string `env:"KAFKA_TOPICS"`

	// MaxPollRecords limits how many records are handled in one poll iteration.
	MaxPollRecords int `env:"KAFKA_MAX_POLL_RECORDS"`

	// FetchMaxWait limits how long a broker may wait before returning a fetch response.
	FetchMaxWait time.Duration `env:"KAFKA_FETCH_MAX_WAIT"`

	// FetchMaxBytes limits the maximum bytes returned in one fetch response.
	FetchMaxBytes int32 `env:"KAFKA_FETCH_MAX_BYTES"`

	// FetchMinBytes sets the minimum bytes a broker tries to return in one fetch response.
	FetchMinBytes int32 `env:"KAFKA_FETCH_MIN_BYTES"`

	// FetchMaxPartitionBytes limits how many bytes may be fetched from one partition.
	FetchMaxPartitionBytes int32 `env:"KAFKA_FETCH_MAX_PARTITION_BYTES"`

	// DisableFetchSessions disables Kafka fetch sessions.
	DisableFetchSessions bool `env:"KAFKA_DISABLE_FETCH_SESSIONS"`

	// Rack identifies the client rack for rack-aware fetching and consumer assignment.
	Rack string `env:"KAFKA_RACK"`

	// 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 `env:"KAFKA_MAX_CONCURRENT_FETCHES"`

	// Group identifies the Kafka consumer group.
	Group string `env:"KAFKA_GROUP"`

	// InstanceID sets a static consumer group member ID.
	InstanceID string `env:"KAFKA_INSTANCE_ID"`

	// SessionTimeout controls how long a group member may go without heartbeating.
	SessionTimeout time.Duration `env:"KAFKA_SESSION_TIMEOUT"`

	// RebalanceTimeout controls how long group members may take during a rebalance.
	RebalanceTimeout time.Duration `env:"KAFKA_REBALANCE_TIMEOUT"`

	// HeartbeatInterval controls how often a group member sends heartbeats.
	HeartbeatInterval time.Duration `env:"KAFKA_HEARTBEAT_INTERVAL"`

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

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

	// ShareMaxRecordsStrict asks the broker to strictly cap records per ShareFetch.
	ShareMaxRecordsStrict bool `env:"KAFKA_SHARE_MAX_RECORDS_STRICT"`

	// ShareRejectAfterDeliveries controls when failed share group records are rejected.
	//
	// If zero, failed records are released for redelivery.
	ShareRejectAfterDeliveries int32 `env:"KAFKA_SHARE_REJECT_AFTER_DELIVERIES"`

	// ShareReleaseTimeout delays AckRelease after share handler errors.
	//
	// If zero, failed records are released immediately.
	ShareReleaseTimeout time.Duration `env:"KAFKA_SHARE_RELEASE_TIMEOUT"`

	// Enabled enables this Kafka component.
	Enabled bool `env:"KAFKA_ENABLED"`

	// SkipFatalErrors allows the consumer to continue after non-retriable fetch errors.
	SkipFatalErrors bool `env:"KAFKA_SKIP_FATAL_ERRORS"`

	// PollInterval sets the interval between consumer poll iterations.
	PollInterval time.Duration `env:"KAFKA_POLL_INTERVAL"`

	// SuspendProcessingTimeout sets the wait time after handler errors.
	SuspendProcessingTimeout time.Duration `env:"KAFKA_SUSPEND_PROCESSING_TIMEOUT"`

	// SuspendCommittingTimeout sets the wait time after offset commit or ack errors.
	SuspendCommittingTimeout time.Duration `env:"KAFKA_SUSPEND_COMMITTING_TIMEOUT"`
}

type ConsumerOpt

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 WithConsumerShareBatchHandler

func WithConsumerShareBatchHandler(handler BatchHandlerFunc) ConsumerOpt

func WithFetchIsolationLevel

func WithFetchIsolationLevel(level kgo.IsolationLevel) ConsumerOpt

func WithGroupTransactSessionBatchHandler

func WithGroupTransactSessionBatchHandler(handler BatchTxHandlerFunc) ConsumerOpt

type GroupOpt

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

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

func WithShareAckCallback

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

type GroupTransactSession

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

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

NewGroupTransactSession creates a Kafka group transaction session and verifies broker connectivity.

func (*GroupTransactSession) HandleFetches

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

func (*GroupTransactSession) Produce

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

func (*GroupTransactSession) ProduceSync

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

func (*GroupTransactSession) Shutdown

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

Shutdown stops polling and closes the group transaction session.

func (*GroupTransactSession) TryProduce

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

type Opt

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

func WithClientID

func WithClientID(s string) Opt

func WithConfig

func WithConfig(config *Config) Opt

func WithLogger

func WithLogger(logger *slog.Logger) Opt

func WithMeterProvider

func WithMeterProvider(provider metric.MeterProvider) Opt

func WithMetricLabel

func WithMetricLabel(key, value string) Opt

func WithMetricsNamespace

func WithMetricsNamespace(namespace string) Opt

func WithTLS

func WithTLS(tlsConfig *tls.Config) Opt

func WithTracerPropagator

func WithTracerPropagator(propagator propagation.TextMapPropagator) Opt

func WithTracerProvider

func WithTracerProvider(provider trace.TracerProvider) Opt

type ProducePromiseFunc

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

ProducePromiseFunc is called when producing a record completes.

type ProducerOpt

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

func WithAllowIdempotentProduceCancellation() ProducerOpt

func WithProducePromise

func WithProducePromise(promiseFunc ProducePromiseFunc) ProducerOpt

func WithProducerBatchCompression

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

func WithProducerBatchMaxBytesFn

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

func WithRecordPartitioner

func WithRecordPartitioner(partitioner kgo.Partitioner) ProducerOpt

func WithRequiredAcks

func WithRequiredAcks(acks kgo.Acks) ProducerOpt

type TxFunc

type TxFunc func(ctx context.Context) error

TxFunc handles work inside a Kafka transaction.