xkafka

package module

Go to main page

Versions in this module

v0

v0.7.0

Sep 19, 2026

v0.6.0

Sep 13, 2026

Changes in this version

type Client

+ func (c *Client) Name() string

type GroupTransactSession

+ func (g *GroupTransactSession) Name() string

+ type Hook any

+ type HookClientClosed interface

+ OnClientClosed func(*Client)

+ type HookFetchError interface

+ OnFetchError func(ctx context.Context, topic string, partition int32, recoverable bool, ...)

+ type HookGroupTransactSessionClosed interface

+ OnGroupTransactSessionClosed func(*GroupTransactSession)

+ type HookHandleEnd interface

+ OnHandleEnd func(ctx context.Context, records []*kgo.Record, duration time.Duration, err error)

+ type HookHandleStart interface

+ OnHandleStart func(ctx context.Context, records []*kgo.Record) context.Context

+ type HookNewClient interface

+ OnNewClient func(*Client)

+ type HookNewGroupTransactSession interface

+ OnNewGroupTransactSession func(*GroupTransactSession)

+ type HookOffsetCommit interface

+ OnOffsetCommit func(ctx context.Context, duration time.Duration, err error)

+ type HookProduceEnd interface

+ OnProduceEnd func(ctx context.Context, records []*kgo.Record, duration time.Duration, err error)

+ type HookProduceError interface

+ OnProduceError func(record *kgo.Record, err error)

+ type HookProduceRecord interface

+ OnProduceRecord func(ctx context.Context, record *kgo.Record)

+ type HookProduceStart interface

+ OnProduceStart func(ctx context.Context, records []*kgo.Record) context.Context

+ type HookShareAck interface

+ OnShareAck func(ctx context.Context, outcome ShareAckOutcome, recordCount int)

+ type HookShareAckFlush interface

+ OnShareAckFlush func(ctx context.Context, duration time.Duration, err error)

+ type HookTransactionEnd interface

+ OnTransactionEnd func(ctx context.Context, transactionType TransactionType, ...)

+ type HookTransactionStart interface

+ OnTransactionStart func(ctx context.Context, transactionType TransactionType) context.Context

type Opt

+ func WithBatchHandler(handler BatchHandlerFunc) Opt

+ func WithHooks(hooks ...Hook) Opt

+ func WithKafkaOptions(opts ...kgo.Opt) Opt

+ func WithMaxPollRecords(maxPollRecords int) Opt

+ func WithMaxRetries(maxRetries int) Opt

+ func WithName(name string) Opt

+ func WithPollInterval(interval time.Duration) Opt

+ func WithShareRejectAfterDeliveries(deliveries int32) Opt

+ func WithShareReleaseTimeout(timeout time.Duration) Opt

+ func WithSuspendCommittingTimeout(timeout time.Duration) Opt

+ func WithSuspendProcessingTimeout(timeout time.Duration) Opt

+ type ShareAckOutcome string

+ const ShareAckAccept

+ const ShareAckReject

+ const ShareAckRelease

+ type TransactionOutcome string

+ const TransactionOutcomeAbort

+ const TransactionOutcomeCommit

+ const TransactionOutcomeError

+ type TransactionType string

+ const TransactionTypeGroup

+ const TransactionTypeProducer

v0.5.1

Jul 4, 2026

v0.5.0

Jul 3, 2026

v0.4.4

Jul 2, 2026

Changes in this version

type Client

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

type GroupTransactSession

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

v0.4.3

Jul 1, 2026

v0.4.2

Jul 1, 2026

Changes in this version

type ConsumerOpt

+ func WithShareGroupBatchHandler(handler BatchHandlerFunc) ConsumerOpt

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

v0.4.0

Jun 30, 2026

Changes in this version

+ type Tx struct

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

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

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

v0.3.1

Jun 29, 2026

Changes in this version

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

+ type Client struct

+ 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 struct

+ AlwaysRetryEOF bool

+ BrokerMaxReadBytes *int32

+ BrokerMaxWriteBytes *int32

+ Brokers string

+ ConnIdleTimeout time.Duration

+ ConsumeRegex bool

+ DefaultProduceTopic string

+ DialTimeout time.Duration

+ DisableFetchSessions bool

+ Enabled bool

+ FetchMaxBytes int32

+ FetchMaxPartitionBytes int32

+ FetchMaxWait time.Duration

+ FetchMinBytes int32

+ Group string

+ HeartbeatInterval time.Duration

+ InstanceID string

+ MaxBufferedBytes int

+ MaxBufferedRecords int

+ MaxConcurrentFetches *int

+ MaxPollRecords int

+ MetadataMaxAge time.Duration

+ MetadataMinAge time.Duration

+ Password string

+ PollInterval time.Duration

+ ProduceRequestTimeout time.Duration

+ ProducerBatchMaxBytes int32

+ ProducerLinger *time.Duration

+ Rack string

+ RebalanceTimeout time.Duration

+ RecordDeliveryTimeout time.Duration

+ RecordRetries int

+ RequestRetries *int

+ RequestTimeoutOverhead time.Duration

+ RetryTimeout time.Duration

+ SASLMechanism string

+ SessionTimeout time.Duration

+ ShareGroup string

+ ShareMaxRecords int32

+ ShareMaxRecordsStrict bool

+ ShareRejectAfterDeliveries int32

+ ShareReleaseTimeout time.Duration

+ SkipFatalErrors bool

+ SuspendCommittingTimeout time.Duration

+ SuspendProcessingTimeout time.Duration

+ Topics string

+ TransactionTimeout time.Duration

+ TransactionalID string

+ User string

+ type ConsumerOpt interface

+ func WithConsumerShareBatchHandler(handler BatchHandlerFunc) ConsumerOpt

+ func WithGroupTransactSessionBatchHandler(handler BatchTxHandlerFunc) ConsumerOpt

+ type GroupOpt interface

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

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

+ type GroupTransactSession struct

+ 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 interface

+ 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 func(record *kgo.Record, err error)

+ type ProducerOpt interface

+ func WithAllowIdempotentProduceCancellation() ProducerOpt

+ func WithProducePromise(promiseFunc ProducePromiseFunc) ProducerOpt

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

+ func WithRecordPartitioner(partitioner kgo.Partitioner) ProducerOpt

+ func WithRequiredAcks(acks kgo.Acks) ProducerOpt

+ type TxFunc func(ctx context.Context) error

v0.1.14

Jun 21, 2026

Changes in this version

+ var JSONMarshal = MarshalJSON()

+ var JSONUnmarshal = UnmarshalJSON()

+ var ProtoMarshal = MarshalProto()

+ var ProtoUnmarshal = UnmarshalProto()

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

+ func GenerateUUID() string

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

+ type Consumer struct

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

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

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

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

+ type ConsumerConfig struct

+ BrokerMaxReadBytes *int32

+ BrokerMaxWriteBytes *int32

+ Brokers string

+ ConnIdleTimeout time.Duration

+ ConsumeRegex bool

+ DialTimeout time.Duration

+ DisableFetchSessions bool

+ Enabled bool

+ FetchMaxBytes int32

+ FetchMaxPartitionBytes int32

+ FetchMaxWait time.Duration

+ FetchMinBytes int32

+ Group string

+ HeartbeatInterval time.Duration

+ InstanceID string

+ MaxPollRecords int

+ MetadataMaxAge time.Duration

+ MetadataMinAge time.Duration

+ Password string

+ PollInterval time.Duration

+ RebalanceTimeout time.Duration

+ RequestRetries *int

+ RequestTimeoutOverhead time.Duration

+ RequireStableFetchOffsets bool

+ RetryTimeout time.Duration

+ SASLMechanism string

+ SessionTimeout time.Duration

+ SkipFatalErrors bool

+ SuspendCommitingTimeout time.Duration

+ SuspendProcessingTimeout time.Duration

+ Topics string

+ User string

+ type ConsumerOption interface

+ func WithConsumeResetOffset(offset kgo.Offset) ConsumerOption

+ func WithConsumerBatchHandler(handlerFunc BatchHandlerFunc) ConsumerOption

+ func WithConsumerClientID(v string) ConsumerOption

+ func WithConsumerConfig(cfg *ConsumerConfig) ConsumerOption

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

+ func WithConsumerHandler(handlerFunc HandlerFunc) ConsumerOption

+ func WithConsumerLogger(v *slog.Logger) ConsumerOption

+ func WithConsumerMeterProvider(provider metric.MeterProvider) ConsumerOption

+ func WithConsumerMetricsNamespace(ns string) ConsumerOption

+ func WithConsumerTLS(tls *tls.Config) ConsumerOption

+ func WithConsumerTracerPropagator(propagator propagation.TextMapPropagator) ConsumerOption

+ func WithConsumerTracerProvider(provider trace.TracerProvider) ConsumerOption

+ func WithFetchIsolationLevel(level kgo.IsolationLevel) ConsumerOption

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

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

+ func MarshalJSON() MarshalFunc

+ func MarshalProto() MarshalFunc

+ type NumberOrString interface

+ type Producer struct

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

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

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

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

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

+ type ProducerConfig struct

+ BrokerMaxReadBytes *int32

+ BrokerMaxWriteBytes *int32

+ Brokers string

+ ConnIdleTimeout time.Duration

+ DefaultProduceTopic string

+ DialTimeout time.Duration

+ MaxBufferedBytes int

+ MaxBufferedRecords int

+ MetadataMaxAge time.Duration

+ MetadataMinAge time.Duration

+ Password string

+ ProduceRequestTimeout time.Duration

+ ProducerBatchMaxBytes int32

+ ProducerLinger time.Duration

+ RecordDeliveryTimeout time.Duration

+ RecordRetries int

+ RequestRetries *int

+ RequestTimeoutOverhead time.Duration

+ RetryTimeout time.Duration

+ SASLMechanism string

+ TransactionTimeout time.Duration

+ TransactionalID string

+ User string

+ type ProducerOption interface

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

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

+ func WithProducerClientID(v string) ProducerOption

+ func WithProducerConfig(cfg *ProducerConfig) ProducerOption

+ func WithProducerLogger(v *slog.Logger) ProducerOption

+ func WithProducerMeterProvider(provider metric.MeterProvider) ProducerOption

+ func WithProducerMetricsNamespace(ns string) ProducerOption

+ func WithProducerPartitioner(partitioner kgo.Partitioner) ProducerOption

+ func WithProducerRequiredAcks(acks kgo.Acks) ProducerOption

+ func WithProducerTLS(tls *tls.Config) ProducerOption

+ func WithProducerTracerPropagator(propagator propagation.TextMapPropagator) ProducerOption

+ func WithProducerTracerProvider(provider trace.TracerProvider) ProducerOption

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

+ func UnmarshalJSON() UnmarshalFunc

+ func UnmarshalProto() UnmarshalFunc

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL