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

kafka

package module
v0.1.14 Latest Latest
Warning

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

 Go to latest Published: Jun 21, 2026 License: MIT Imports: 21 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, and exposing Kafka observability with OpenTelemetry and Prometheus.

Features

  • Producer: Synchronous, asynchronous, and transactional message publishing.
  • Consumer: Per-record and batch handlers for flexible message processing.
  • Commits: Automatic offset commits only after successful processing.
  • Retries: Built-in retry handling with configurable backoff strategies.
  • Observability: OpenTelemetry instrumentation and Prometheus metrics hooks out of the box.
  • Security: TLS and SASL support, including PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512.
  • Configuration: Configure via Go structs or environment variables.

Installation

go get github.com/mkbeh/xkafka

Quick start

The examples below show the basic sending and receiving flow. For production workloads, tune retries, offset commits, observability, security, and graceful shutdown behavior for your needs.

Sending messages
package main

import (
	"context"
	"log"

	kafka "github.com/mkbeh/xkafka"
	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	producer, err := kafka.NewProducer(
		kafka.WithProducerConfig(&kafka.ProducerConfig{
			Brokers: "localhost:9092",
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer producer.Close(context.Background())

	ctx := context.Background()

	// Async (fire-and-forget with logging on error)
	producer.ProduceAsync(ctx, &kgo.Record{
		Topic: "my-topic",
		Value: []byte("hello async"),
	})

	// Sync produce waits until the broker acknowledges the record.
	if err := producer.ProduceSync(ctx, &kgo.Record{
		Topic: "my-topic",
		Value: []byte("hello sync"),
	}); err != nil {
		log.Fatal(err)
	}
}
Receiving messages
package main

import (
	"context"
	"fmt"
	"log"

	kafka "github.com/mkbeh/xkafka"
	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	consumer, err := kafka.NewConsumer(
		kafka.WithConsumerConfig(&kafka.ConsumerConfig{
			Enabled: true,
			Brokers: "localhost:9092",
			Topics:  "my-topic",
			Group:   "my-group",
		}),
		kafka.WithConsumerHandler(func(ctx context.Context, record *kgo.Record) error {
			fmt.Printf("key=%s value=%s\n", record.Key, record.Value)
			return nil
		}),
	)
	if err != nil {
		log.Fatal(err)
	}

	ctx := context.Background()

	if err := consumer.PreRun(ctx); err != nil {
		log.Fatal(err)
	}
	defer consumer.Shutdown(ctx)

	if err := consumer.Run(ctx); err != nil {
		log.Fatal(err)
	}
}
Receiving messages in batches
consumer, err := kafka.NewConsumer(
    kafka.WithConsumerConfig(&kafka.ConsumerConfig{
        Enabled:        true,
        Brokers:        "localhost:9092",
        Topics:         "orders.created",
        Group:          "orders-batch-worker-group",
        MaxPollRecords: 500,
    }),
    kafka.WithConsumerBatchHandler(func(ctx context.Context, records []*kgo.Record) error {
        for _, record := range records {
            // process record
        }

        return nil
    }),
)

Note: Offsets are committed only after successful handler execution. If the handler returns an error, processing is retried after KAFKA_SUSPEND_PROCESSING_TIMEOUT (30s by default).

These examples show the basic sending and receiving flow. For production use, configure retries, commits, observability, security, and shutdown behavior for your workload.

Transactions

To enable transactional producing, configure TransactionalID and use RunInTx.

producer, err := kafka.NewProducer(
    kafka.WithProducerConfig(&kafka.ProducerConfig{
        Brokers:         "localhost:9092",
        TransactionalID: "orders-tx-producer",
    }),
)
if err != nil {
    log.Fatal(err)
}

records := []*kgo.Record{
    {
        Topic: "orders.created",
        Key:   []byte("order-1"),
        Value: []byte("created"),
    },
    {
        Topic: "audit.events",
        Key:   []byte("order-1"),
        Value: []byte("order created"),
    },
}

if err := producer.RunInTx(context.Background(), records...); err != nil {
    log.Fatal(err)
}

RunInTx sends all records in a single Kafka transaction. If producing fails, the transaction is aborted and the records are not committed.

Observability

xkafka uses franz-go hooks for Kafka instrumentation and supports custom OpenTelemetry meter and tracer providers.

Producer
producer, err := kafka.NewProducer(
    kafka.WithProducerConfig(&kafka.ProducerConfig{
        Brokers: "localhost:9092",
    }),
    kafka.WithProducerMeterProvider(meterProvider),
    kafka.WithProducerTracerProvider(tracerProvider),
    kafka.WithProducerMetricsNamespace("orders"),
)
Consumer
consumer, err := kafka.NewConsumer(
    kafka.WithConsumerConfig(&kafka.ConsumerConfig{
        Enabled: true,
        Brokers: "localhost:9092",
        Topics:  "orders.created",
        Group:   "orders-worker-group",
    }),
    kafka.WithConsumerMeterProvider(meterProvider),
    kafka.WithConsumerTracerProvider(tracerProvider),
    kafka.WithConsumerMetricsNamespace("orders"),
    kafka.WithConsumerHandler(handler),
)

The following metric labels are added automatically:

Label Description
client_id Generated client identifier, or configured client ID with a unique suffix.
client_kind producer or consumer.
consumer_group Consumer group ID. Added only for consumers with a configured group.

Configuration

ProducerConfig and ConsumerConfig can be configured directly as Go structs. Their fields also include envconfig tags, so you can populate them from environment variables using your preferred configuration layer.

Common configuration
ENV Description
KAFKA_BROKERS Comma-separated seed brokers, for example kafka-1:9092,kafka-2:9092.
KAFKA_SASL_MECHANISM SASL mechanism: PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.
KAFKA_USER SASL username.
KAFKA_PASSWORD SASL password.
KAFKA_REQUEST_RETRIES Number of retries for retryable requests.
KAFKA_RETRY_TIMEOUT Total retry time limit.
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 to a broker connection in one write.
KAFKA_MAX_READ_BYTES Maximum bytes read from a broker response.
Producer configuration
ENV Description
KAFKA_DEFAULT_PRODUCE_TOPIC Default topic used when kgo.Record.Topic is empty.
KAFKA_PRODUCER_BATCH_MAX_BYTES Maximum size of a producer batch.
KAFKA_MAX_BUFFERED_RECORDS Maximum number of buffered records before producing blocks.
KAFKA_MAX_BUFFERED_BYTES Maximum number of buffered bytes before producing blocks.
KAFKA_PRODUCE_REQUEST_TIMEOUT Maximum time a broker has to respond to a produce request.
KAFKA_RECORD_RETRIES Number of record-level produce retries.
KAFKA_PRODUCER_LINGER Time to wait for more records before building a batch.
KAFKA_RECORD_DELIVERY_TIMEOUT Maximum time a record can remain buffered before timing out.
KAFKA_TRANSACTIONAL_ID Enables transactional producing.
KAFKA_TRANSACTION_TIMEOUT Maximum allowed transaction duration.
Consumer configuration
ENV Default Description
KAFKA_ENABLED false Enables the consumer. If false, PreRun and Run are no-ops.
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 returned per poll.
KAFKA_POLL_INTERVAL 300ms Interval between polls.
KAFKA_SKIP_FATAL_ERRORS true Continue after non-retryable fetch errors.
KAFKA_SUSPEND_PROCESSING_TIMEOUT 30s Backoff after handler error.
KAFKA_SUSPEND_COMMITTING_TIMEOUT 10s Backoff after offset commit error.
KAFKA_INSTANCE_ID — Static group member ID.
KAFKA_CONSUME_REGEX false Treat configured topics as regular expressions.
KAFKA_DISABLE_FETCH_SESSIONS false Disable Kafka fetch sessions.
KAFKA_SESSION_TIMEOUT 45s Maximum time between heartbeats before rebalance.
KAFKA_REBALANCE_TIMEOUT — Maximum time for members to rejoin during rebalance.
KAFKA_HEARTBEAT_INTERVAL — Heartbeat interval.
KAFKA_REQUIRE_STABLE_FETCH_OFFS false Require stable offsets before consuming.
KAFKA_FETCH_MAX_WAIT — Maximum broker wait time before returning a fetch response.
KAFKA_FETCH_MAX_BYTES — Maximum bytes per fetch response.
KAFKA_FETCH_MIN_BYTES — Minimum bytes the broker should accumulate before responding.
KAFKA_FETCH_MAX_PARTITION_BYTES — Maximum bytes per partition in a fetch response.

License

This project is licensed under the MIT License.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ProtoMarshal   = MarshalProto()
	ProtoUnmarshal = UnmarshalProto()
	JSONMarshal    = MarshalJSON()
	JSONUnmarshal  = UnmarshalJSON()
)

Functions

func ConvertAnyToBytes 

func ConvertAnyToBytes[T NumberOrString](v T) []byte

func GenerateUUID 

func GenerateUUID() string

Types

type BatchHandlerFunc 

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

type Consumer 

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

func NewConsumer 

func NewConsumer(opts ...ConsumerOption) (*Consumer, error)

func (*Consumer) PreRun 

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

func (*Consumer) Run 

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

func (*Consumer) Shutdown 

func (c *Consumer) Shutdown(_ context.Context) error

type ConsumerConfig 

type ConsumerConfig struct {
	// SeedBrokers sets the seed brokers for the client to use, overriding the
	// default 127.0.0.1:9092.
	//
	// Any seeds that are missing a port use the default Kafka port 9092.
	Brokers string `envconfig:"KAFKA_BROKERS"`

	// SaslMechanism SASL mechanism to use for authentication.
	// Supported: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512.
	// NOTE: Despite the name only one mechanism must be configured.
	SASLMechanism string `envconfig:"KAFKA_SASL_MECHANISM"`
	// User sasl username for use with the PLAIN and SASL-SCRAM-.. mechanisms.
	User string `envconfig:"KAFKA_USER"`
	// Password sasl password for use with the PLAIN and SASL-SCRAM-.. mechanism.
	Password string `envconfig:"KAFKA_PASSWORD"`

	// Enabled custom option: detecting enabling.
	Enabled bool `envconfig:"KAFKA_ENABLED"`
	// SkipFatalErrors custom option: skip fatal errors while fetching records,
	// otherwise leave the process.
	SkipFatalErrors bool `envconfig:"KAFKA_SKIP_FATAL_ERRORS"`
	// ConsumeRegex sets the client to parse all topics passed to Topics as
	// regular expressions.
	ConsumeRegex bool `envconfig:"KAFKA_CONSUME_REGEX"`
	// Topics adds topics to use for consuming.
	Topics string `envconfig:"KAFKA_TOPICS"`
	// Group sets the consumer group for the client to join and consume in.
	// This option is required if using any other group options.
	Group string `envconfig:"KAFKA_GROUP"`
	// MaxPollRecords maximum of maxPollRecords total across all fetches.
	MaxPollRecords int `envconfig:"KAFKA_MAX_POLL_RECORDS"`
	// InstanceID sets the group consumer's instance ID, switching the group member
	// from "dynamic" to "static".
	InstanceID string `envconfig:"KAFKA_INSTANCE_ID"`

	// PollInterval interval between handle batches.
	PollInterval time.Duration `envconfig:"KAFKA_POLL_INTERVAL"`
	// SuspendProcessingTimeout waiting timeout after batch processing failed (custom property).
	SuspendProcessingTimeout time.Duration `envconfig:"KAFKA_SUSPEND_PROCESSING_TIMEOUT"`
	// SuspendCommitingTimeout waiting timeout after committing failed (custom property).
	SuspendCommitingTimeout time.Duration `envconfig:"KAFKA_SUSPEND_COMMITTING_TIMEOUT"`

	// FetchMaxWait sets the maximum amount of time a broker will wait for a
	// fetch response to hit the minimum number of required bytes before returning.
	FetchMaxWait time.Duration `envconfig:"KAFKA_FETCH_MAX_WAIT"`
	// FetchMaxBytes sets the maximum amount of bytes a broker will try to send
	// during a fetch.
	FetchMaxBytes int32 `envconfig:"KAFKA_FETCH_MAX_BYTES"`
	// FetchMinBytes sets the minimum amount of bytes a broker will try to send
	// during a fetch.
	FetchMinBytes int32 `envconfig:"KAFKA_FETCH_MIN_BYTES"`
	// FetchMaxPartitionBytes sets the maximum amount of bytes that will be
	// consumed for a single partition in a fetch request.
	FetchMaxPartitionBytes int32 `envconfig:"KAFKA_FETCH_MAX_PARTITION_BYTES"`
	// DisableFetchSessions sets the client to not use fetch sessions (Kafka 1.0+).
	DisableFetchSessions bool `envconfig:"KAFKA_DISABLE_FETCH_SESSIONS"`
	// SessionTimeout sets how long a member in the group can go between
	// heartbeats, overriding the default 45,000ms. If a member does not heartbeat
	// in this timeout, the broker will remove the member from the group and
	// initiate a rebalance.
	SessionTimeout time.Duration `envconfig:"KAFKA_SESSION_TIMEOUT"`
	// RebalanceTimeout sets how long group members are allowed to take when a
	// rebalance has begun.
	RebalanceTimeout time.Duration `envconfig:"KAFKA_REBALANCE_TIMEOUT"`
	// HeartbeatInterval sets how long a group member goes between heartbeats to
	// Kafka.
	HeartbeatInterval time.Duration `envconfig:"KAFKA_HEARTBEAT_INTERVAL"`
	// RequireStableFetchOffsets sets the group consumer to require "stable" fetch
	// offsets before consuming from the group.
	RequireStableFetchOffsets bool `envconfig:"KAFKA_REQUIRE_STABLE_FETCH_OFFS"`

	// RequestTimeoutOverhead uses the given time as overhead while deadlining
	// requests.
	RequestTimeoutOverhead time.Duration `envconfig:"KAFKA_REQUEST_TIMEOUT_OVERHEAD"`
	// ConnIdleTimeout is a rough amount of time to allow connections to idle
	// before they are closed.
	ConnIdleTimeout time.Duration `envconfig:"KAFKA_CONN_IDLE_TIMEOUT"`
	// DialTimeout sets the dial timeout.
	DialTimeout time.Duration `envconfig:"KAFKA_DIAL_TIMEOUT"`
	// RequestRetries sets the number of tries that retryable requests are allowed.
	RequestRetries *int `envconfig:"KAFKA_REQUEST_RETRIES"`
	// RetryTimeout sets the upper limit on how long we allow a request to be
	// issued and then reissued on failure. That is, this control the total
	// end-to-end maximum time we allow for trying a request.
	RetryTimeout time.Duration `envconfig:"KAFKA_RETRY_TIMEOUT"`
	// BrokerMaxWriteBytes upper bounds the number of bytes written to a broker
	// connection in a single write.
	BrokerMaxWriteBytes *int32 `envconfig:"KAFKA_MAX_WRITE_BYTES"`
	// BrokerMaxReadBytes sets the maximum response size that can be read from
	// Kafka.
	BrokerMaxReadBytes *int32 `envconfig:"KAFKA_MAX_READ_BYTES"`
	// MetadataMaxAge sets the maximum age for the client's cached metadata.
	MetadataMaxAge time.Duration `envconfig:"KAFKA_METADATA_MAX_AGE"`
	// MetadataMinAge sets the minimum time between metadata queries.
	MetadataMinAge time.Duration `envconfig:"KAFKA_METADATA_MIN_AGE"`
}

type ConsumerOption 

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

func WithConsumeResetOffset 

func WithConsumeResetOffset(offset kgo.Offset) ConsumerOption

WithConsumeResetOffset sets the offset to start consuming from, or if OffsetOutOfRange is seen while fetching, to restart consuming from.

func WithConsumerBatchHandler 

func WithConsumerBatchHandler(handlerFunc BatchHandlerFunc) ConsumerOption

WithConsumerBatchHandler set batch message handler

func WithConsumerClientID 

func WithConsumerClientID(v string) ConsumerOption

func WithConsumerConfig 

func WithConsumerConfig(cfg *ConsumerConfig) ConsumerOption

func WithConsumerGroupBalancers 

func WithConsumerGroupBalancers(balancers ...kgo.GroupBalancer) ConsumerOption

WithConsumerGroupBalancers sets the group balancers to use for dividing topic partitions among group members, overriding the current default [cooperative-sticky]. This option is equivalent to Kafka's partition.assignment.strategies option.

For balancing, Kafka chooses the first protocol that all group members agree to support.

Note that if you opt into cooperative-sticky rebalancing, cooperative group balancing is incompatible with eager (classical) rebalancing and requires a careful rollout strategy (see KIP-429).

func WithConsumerHandler 

func WithConsumerHandler(handlerFunc HandlerFunc) ConsumerOption

WithConsumerHandler set message handler

func WithConsumerLogger 

func WithConsumerLogger(v *slog.Logger) ConsumerOption

func WithConsumerMeterProvider 

func WithConsumerMeterProvider(provider metric.MeterProvider) ConsumerOption

func WithConsumerMetricsNamespace 

func WithConsumerMetricsNamespace(ns string) ConsumerOption

func WithConsumerTLS 

func WithConsumerTLS(tls *tls.Config) ConsumerOption

WithConsumerTLS opts into dialing brokers with the given TLS config with a 10s dial timeout.

Every dial, the input config is cloned. If the config's ServerName is not specified, this function uses net.SplitHostPort to extract the host from the broker being dialed and sets the ServerName. In short, it is not necessary to set the ServerName.

func WithConsumerTracerPropagator 

func WithConsumerTracerPropagator(propagator propagation.TextMapPropagator) ConsumerOption

func WithConsumerTracerProvider 

func WithConsumerTracerProvider(provider trace.TracerProvider) ConsumerOption

func WithFetchIsolationLevel 

func WithFetchIsolationLevel(level kgo.IsolationLevel) ConsumerOption

WithFetchIsolationLevel sets the "isolation level" used for fetching records.

type HandlerFunc 

type HandlerFunc func(ctx context.Context, msg *kgo.Record) error

type MarshalFunc 

type MarshalFunc func(m interface{}) ([]byte, error)

func MarshalJSON 

func MarshalJSON() MarshalFunc

func MarshalProto 

func MarshalProto() MarshalFunc

type NumberOrString 

type NumberOrString interface {
	int | int8 | int16 | int32 | int64 | uint | uint8 | uint16 | uint32 | uint64 | float32 | float64 | string
}

type Producer 

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

func NewProducer 

func NewProducer(opts ...ProducerOption) (*Producer, error)

func (*Producer) Close 

func (p *Producer) Close(ctx context.Context)

func (*Producer) ProduceAsync 

func (p *Producer) ProduceAsync(ctx context.Context, record *kgo.Record)

func (*Producer) ProduceSync 

func (p *Producer) ProduceSync(ctx context.Context, records ...*kgo.Record) error

func (*Producer) RunInTx 

func (p *Producer) RunInTx(ctx context.Context, record ...*kgo.Record) error

type ProducerConfig 

type ProducerConfig struct {
	// Brokers sets the seed brokers for the client to use, overriding the
	// default 127.0.0.1:9092.
	//
	// Any seeds that are missing a port use the default Kafka port 9092.
	Brokers string `envconfig:"KAFKA_BROKERS"`

	// SaslMechanism SASL mechanism to use for authentication.
	// Supported: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512.
	// NOTE: Despite the name only one mechanism must be configured.
	SASLMechanism string `envconfig:"KAFKA_SASL_MECHANISM"`
	// User sasl username for use with the PLAIN and SASL-SCRAM-.. mechanisms.
	User string `envconfig:"KAFKA_USER"`
	// Password sasl password for use with the PLAIN and SASL-SCRAM-.. mechanism.
	Password string `envconfig:"KAFKA_PASSWORD"`

	// RequestTimeoutOverhead uses the given time as overhead while deadlining
	// requests.
	RequestTimeoutOverhead time.Duration `envconfig:"KAFKA_REQUEST_TIMEOUT_OVERHEAD"`
	// ConnIdleTimeout is a rough amount of time to allow connections to idle
	// before they are closed.
	ConnIdleTimeout time.Duration `envconfig:"KAFKA_CONN_IDLE_TIMEOUT"`
	// DialTimeout sets the dial timeout.
	DialTimeout time.Duration `envconfig:"KAFKA_DIAL_TIMEOUT"`
	// RequestRetries sets the number of tries that retryable requests are allowed.
	RequestRetries *int `envconfig:"KAFKA_REQUEST_RETRIES"`
	// RetryTimeout sets the upper limit on how long we allow a request to be
	// issued and then reissued on failure. That is, this control the total
	// end-to-end maximum time we allow for trying a request.
	RetryTimeout time.Duration `envconfig:"KAFKA_RETRY_TIMEOUT"`
	// BrokerMaxWriteBytes upper bounds the number of bytes written to a broker
	// connection in a single write.
	BrokerMaxWriteBytes *int32 `envconfig:"KAFKA_MAX_WRITE_BYTES"`
	// BrokerMaxReadBytes sets the maximum response size that can be read from
	// Kafka.
	BrokerMaxReadBytes *int32 `envconfig:"KAFKA_MAX_READ_BYTES"`
	// MetadataMaxAge sets the maximum age for the client's cached metadata.
	MetadataMaxAge time.Duration `envconfig:"KAFKA_METADATA_MAX_AGE"`
	// MetadataMinAge sets the minimum time between metadata queries.
	MetadataMinAge time.Duration `envconfig:"KAFKA_METADATA_MIN_AGE"`

	// DefaultProduceTopic sets the default topic to produce to if the topic field
	// is empty in a Record.
	DefaultProduceTopic string `envconfig:"KAFKA_DEFAULT_PRODUCE_TOPIC"`
	// ProducerBatchMaxBytes upper bounds the size of a record batch.
	ProducerBatchMaxBytes int32 `envconfig:"KAFKA_PRODUCER_BATCH_MAX_BYTES"`
	// MaxBufferedRecords sets the max amount of records the client will buffer,
	// blocking produces until records are finished if this limit is reached.
	MaxBufferedRecords int `envconfig:"KAFKA_MAX_BUFFERED_RECORDS"`
	// MaxBufferedBytes sets the max amount of bytes that the client will buffer
	// while producing, blocking produces until records are finished if this limit
	// is reached.
	MaxBufferedBytes int `envconfig:"KAFKA_MAX_BUFFERED_BYTES"`
	// ProduceRequestTimeout sets how long Kafka broker's are allowed to respond to
	// produce requests, overriding the default 10s. If a broker exceeds this
	// duration, it will reply with a request timeout error.
	ProduceRequestTimeout time.Duration `envconfig:"KAFKA_PRODUCE_REQUEST_TIMEOUT"`
	// RecordRetries sets the number of tries for producing records.
	RecordRetries int `envconfig:"KAFKA_RECORD_RETRIES"`
	// ProducerLinger sets how long individual topic partitions will linger waiting
	// for more records before triggering a request to be built.
	ProducerLinger time.Duration `envconfig:"KAFKA_PRODUCER_LINGER"`
	// RecordDeliveryTimeout sets a rough time of how long a record can sit around
	// in a batch before timing out.
	RecordDeliveryTimeout time.Duration `envconfig:"KAFKA_RECORD_DELIVERY_TIMEOUT"`
	// TransactionalID sets a transactional ID for the client, ensuring that
	// records are produced transactionally under this ID (exactly once semantics).
	TransactionalID string `envconfig:"KAFKA_TRANSACTIONAL_ID"`
	// TransactionTimeout sets the allowed for a transaction.
	TransactionTimeout time.Duration `envconfig:"KAFKA_TRANSACTION_TIMEOUT"`
}

type ProducerOption 

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

func WithProducerBatchCompression 

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

func WithProducerCB 

func WithProducerCB(cb func(record *kgo.Record, err error)) ProducerOption

WithProducerCB set callback func which calls after attempt to send message in topic

func WithProducerClientID 

func WithProducerClientID(v string) ProducerOption

func WithProducerConfig 

func WithProducerConfig(cfg *ProducerConfig) ProducerOption

func WithProducerLogger 

func WithProducerLogger(v *slog.Logger) ProducerOption

func WithProducerMeterProvider 

func WithProducerMeterProvider(provider metric.MeterProvider) ProducerOption

func WithProducerMetricsNamespace 

func WithProducerMetricsNamespace(ns string) ProducerOption

func WithProducerPartitioner 

func WithProducerPartitioner(partitioner kgo.Partitioner) ProducerOption

func WithProducerRequiredAcks 

func WithProducerRequiredAcks(acks kgo.Acks) ProducerOption

func WithProducerTLS 

func WithProducerTLS(tls *tls.Config) ProducerOption

WithProducerTLS opts into dialing brokers with the given TLS config with a 10s dial timeout.

Every dial, the input config is cloned. If the config's ServerName is not specified, this function uses net.SplitHostPort to extract the host from the broker being dialed and sets the ServerName. In short, it is not necessary to set the ServerName.

func WithProducerTracerPropagator 

func WithProducerTracerPropagator(propagator propagation.TextMapPropagator) ProducerOption

func WithProducerTracerProvider 

func WithProducerTracerProvider(provider trace.TracerProvider) ProducerOption

type UnmarshalFunc 

type UnmarshalFunc func(data []byte, v interface{}) error

func UnmarshalJSON 

func UnmarshalJSON() UnmarshalFunc

func UnmarshalProto 

func UnmarshalProto() UnmarshalFunc

Source Files

View all Source files

Directories

Path Synopsis
examples
sample command
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.