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

xkafka

package module
v0.4.3 Latest Latest
Warning

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

 Go to latest Published: Jul 1, 2026 License: MIT Imports: 19 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

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 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, 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
KAFKA_BROKERS Comma-separated seed brokers
KAFKA_SASL_MECHANISM PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512
KAFKA_USER SASL username
KAFKA_PASSWORD SASL password
KAFKA_REQUEST_TIMEOUT_OVERHEAD Request deadline overhead
KAFKA_REQUEST_RETRIES Max request retries
KAFKA_RETRY_TIMEOUT Total retry time limit
KAFKA_DIAL_TIMEOUT Broker dial timeout
KAFKA_CONN_IDLE_TIMEOUT Idle connection timeout
KAFKA_METADATA_MAX_AGE Max age of cached metadata
KAFKA_METADATA_MIN_AGE Min time between metadata refreshes
KAFKA_MAX_WRITE_BYTES Max bytes per connection write
KAFKA_MAX_READ_BYTES Max bytes per broker response
KAFKA_ALWAYS_RETRY_EOF false Retry EOF errors instead of failing connection
KAFKA_DEFAULT_PRODUCE_TOPIC Fallback topic if record topic is empty
KAFKA_PRODUCER_BATCH_MAX_BYTES Max size of a producer batch
KAFKA_MAX_BUFFERED_RECORDS Max buffered records before blocking
KAFKA_MAX_BUFFERED_BYTES Max buffered bytes before blocking
KAFKA_PRODUCE_REQUEST_TIMEOUT Broker response timeout for produce requests
KAFKA_RECORD_RETRIES Max record-level produce retries
KAFKA_RECORD_DELIVERY_TIMEOUT Max record buffering time
KAFKA_PRODUCER_LINGER Linger delay for batch building
KAFKA_TRANSACTIONAL_ID Transactional identifier for EOS
KAFKA_TRANSACTION_TIMEOUT Max transaction duration
KAFKA_ENABLED true Enable consumer loop
KAFKA_TOPICS Comma-separated topics to consume
KAFKA_GROUP Consumer group ID
KAFKA_MAX_POLL_RECORDS 100 Max records per poll
KAFKA_POLL_INTERVAL 1s Interval between polls
KAFKA_SKIP_FATAL_ERRORS true Continue on non-retryable fetch errors
KAFKA_SUSPEND_PROCESSING_TIMEOUT 30s Backoff delay after handler error
KAFKA_SUSPEND_COMMITTING_TIMEOUT 10s Backoff delay after commit/ack error
KAFKA_INSTANCE_ID Static group membership ID
KAFKA_CONSUME_REGEX false Treat topics as regular expressions
KAFKA_DISABLE_FETCH_SESSIONS false Disable fetch sessions
KAFKA_RACK Rack ID for rack-aware fetching
KAFKA_MAX_CONCURRENT_FETCHES Max concurrent fetches buffered by client
KAFKA_SESSION_TIMEOUT Rebalance session timeout
KAFKA_REBALANCE_TIMEOUT Max time for members to rejoin
KAFKA_HEARTBEAT_INTERVAL Heartbeat interval
KAFKA_FETCH_MAX_WAIT Max broker wait time for fetches
KAFKA_FETCH_MIN_BYTES Min bytes broker accumulates before response
KAFKA_FETCH_MAX_BYTES Max bytes per fetch response
KAFKA_FETCH_MAX_PARTITION_BYTES Max bytes per partition fetch
KAFKA_SHARE_GROUP Share group ID
KAFKA_SHARE_MAX_RECORDS Max records 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

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 is a high-level Kafka client for producing records and consuming records through handlers.

func NewClient added in v0.3.1 

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

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

func (*Client) HandleFetches added in v0.3.1 

func (c *Client) HandleFetches(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 `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 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 and verifies broker connectivity.

func (*GroupTransactSession) HandleFetches added in v0.3.1 

func (g *GroupTransactSession) HandleFetches(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 *slog.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
pkg/kprom
pkg/kslog

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.