xkafka

README

Kafka toolkit for Go

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

xkafka preserves the native franz-go record model and configuration while adding a compact runtime layer for handler-driven consumption, retries, Kafka Share Groups, producer transactions, and Kafka-to-Kafka exactly-once processing (EOS).

Kafka behavior remains fully configurable through franz-go. Rather than introducing a parallel configuration layer, xkafka focuses on the processing lifecycle around it.

Features

• Unified Client: Produce and consume through a single client.
• Native Configuration: Use native franz-go options without introducing a parallel Kafka configuration layer.
• Flexible Producing: Synchronous, asynchronous, non-blocking, and transactional produce operations.
• Batch Consumption: A unified handler model for regular consumer groups and Kafka Share Groups.
• Retries and Recovery: Configurable handler retries and backoff with panic recovery through the same processing path.
• Share Groups (KIP-932): Record acknowledgement, release, redelivery, delivery-count based rejection, and acknowledgement flushing.
• Producer Transactions: Transactional producing with automatic commit and abort handling.
• Exactly-Once Semantics (EOS): Kafka-to-Kafka consume-process-produce transactions with atomic produced records and consumed offsets.
• Runtime Hooks: Extensible hooks for producing, fetching, processing, offset commits, Share Groups, and transactions.
• OpenTelemetry: Optional runtime metrics, distributed tracing, and context propagation.

Installation

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

go get github.com/mkbeh/xkafka

Optional integrations are released independently under extra:

go get github.com/mkbeh/xkafka/extra/otelxkafka

Getting started

Here's a basic overview of producing and consuming:

seeds := []string{"localhost:9092"}

// One client can both produce and consume!
client, err := xkafka.NewClient(
    xkafka.WithKafkaOptions(
        kgo.SeedBrokers(seeds...),
        kgo.ConsumeTopics("foo"),
        kgo.ConsumerGroup("my-group-identifier"),
    ),
    xkafka.WithBatchHandler(func(ctx context.Context, records []*kgo.Record) error {
        for _, record := range records {
            fmt.Printf("received: %s\n", record.Value)
        }

        return nil
    }),
)
if err != nil {
    panic(err)
}
defer client.Shutdown(context.Background())

// 1.) Producing a message.
record := &kgo.Record{Topic: "foo", Value: []byte("value")}
if err := client.ProduceSync(context.Background(), record); err != nil {
    panic(err)
}

// 2.) Consuming messages through the configured batch handler.
if err := client.HandleFetches(context.Background()); err != nil {
    panic(err)
}

This only shows producing and consuming in the most basic sense. The sections below cover transactions, Kafka Share Groups, exactly-once processing, and telemetry in more detail. Check out the examples directory for more!

Transactions

Configure a TransactionalID and execute transactional operations using the Tx passed to RunInTx:

seeds := []string{"localhost:9092"}

// Initialize a client with a transactional ID.
client, err := xkafka.NewClient(
    xkafka.WithKafkaOptions(
        kgo.SeedBrokers(seeds...),
        kgo.TransactionalID("my-tx-identifier"),
    ),
)
if err != nil {
    panic(err)
}
defer client.Shutdown(context.Background())

// RunInTx manages the transaction lifecycle.
err = client.RunInTx(context.Background(), func(ctx context.Context, tx *xkafka.Tx) error {
    record := &kgo.Record{
        Topic: "foo",
        Value: []byte("value"),
    }

    // Produce transactional records through the provided Tx.
    return tx.ProduceSync(ctx, record)
})
if err != nil {
    panic(err)
}

The transaction function controls how xkafka completes the transaction:

Function Result	Behavior
nil	Flushes buffered records and attempts to commit the transaction.
error	Attempts to abort the transaction and returns the error.
panic	Recovers the panic, attempts to abort the transaction, and returns an error.

Share Groups

Kafka Share Groups use the same batch handler API as standard consumer groups, but processing results are translated into record-level acknowledgements rather than consumer-group offset semantics.

seeds := []string{"localhost:9092"}

// Configure a client as a Share Group consumer.
client, err := xkafka.NewClient(
    xkafka.WithKafkaOptions(
        kgo.SeedBrokers(seeds...),
        kgo.ConsumeTopics("foo"),
        kgo.ShareGroup("my-share-group"),
    ),
    xkafka.WithBatchHandler(func(ctx context.Context, records []*kgo.Record) error {
        for _, record := range records {
            // DeliveryCount reports how many times the record has been delivered.
            fmt.Printf("received: %s, delivery_count=%d\n", record.Value, record.DeliveryCount())
        }

        // Returning nil acknowledges all records in the batch with AckAccept.
        return nil
    }),
)
if err != nil {
    panic(err)
}
defer client.Shutdown(context.Background())

// Start the blocking Share Group consumption loop.
if err := client.HandleFetches(context.Background()); err != nil {
    panic(err)
}

For Share Groups, handler outcomes are mapped to record-level Kafka acknowledgements:

Handler Result	Behavior
nil	Acknowledges all records in the batch with AckAccept.
error	Uses AckRelease for broker redelivery. WithShareReleaseTimeout can delay acknowledgement flushing, while WithShareRejectAfterDeliveries changes failed records to AckReject once their delivery count reaches the configured threshold.
panic	Recovers the panic and follows the same failure path as an error, including the configured release delay and delivery-count rejection policy.

Exactly-Once Semantics (EOS)

Use GroupTransactSession for Kafka-to-Kafka consume-process-produce workflows where produced records and consumed offsets must be committed atomically.

The session coordinates the consumer group and transactional producer lifecycles, requiring a consumer group, TransactionalID, and transaction batch handler.

seeds := []string{"localhost:9092"}

// Initialize a Kafka-to-Kafka exactly-once processing session.
session, err := xkafka.NewGroupTransactSession(
    xkafka.WithKafkaOptions(
        kgo.SeedBrokers(seeds...),
        kgo.ConsumeTopics("foo-input"),
        kgo.ConsumerGroup("my-group-identifier"),
        kgo.TransactionalID("my-tx-identifier"),
    ),
    xkafka.WithGroupTransactSessionBatchHandler(
        func(ctx context.Context, records []*kgo.Record, tx *xkafka.Tx) error {
            for _, record := range records {
                // 1.) Process the input record and produce the result through Tx.
                out := &kgo.Record{
                    Topic: "foo-output",
                    Value: record.Value,
                }

                if err := tx.ProduceSync(ctx, out); err != nil {
                    return err
                }
            }

            // 2.) Returning nil allows the produced records and consumed offsets
            // to be committed atomically in the same Kafka transaction.
            return nil
        },
    ),
)
if err != nil {
    panic(err)
}
defer session.Shutdown(context.Background())

// Start the blocking transactional processing loop.
if err := session.HandleFetches(context.Background()); err != nil {
    panic(err)
}

Each fetched batch is processed inside a single group transaction:

Handler Result	Behavior
nil	Attempts to atomically commit the produced records and consumed offsets.
error	Attempts to abort the transaction, leaving the input offsets uncommitted for redelivery.
panic	Recovers the panic and follows the same abort path as an error.

[!IMPORTANT] Unrecoverable transaction errors are returned from HandleFetches and stop the session. Handler errors and panics abort the current transaction but do not stop the fetch loop by themselves.

Telemetry

The core library is telemetry-agnostic, exposing runtime hooks for processing, fetching, producing, offset commits, Share Groups, and transactions.

Optional OpenTelemetry integration is provided by extra/otelxkafka, which adds runtime metrics, distributed tracing, and context propagation. It can be used alongside franz-go/plugin/kotel to collect native franz-go client metrics.

See extra/otelxkafka for the complete setup guide and telemetry reference.

License

This project is licensed under the MIT License.

Documentation

Overview

Package xkafka provides Kafka producing, consuming, Share Group processing, and transactions on top of franz-go.

Use Client for regular producer and consumer workloads and GroupTransactSession for Kafka-to-Kafka exactly-once consume-process-produce workflows. Native franz-go configuration can be supplied with WithKafkaOptions.

Index

• type BatchHandlerFunc
• type BatchTxHandlerFunc
• type Client
  • func NewClient(opts ...Opt) (*Client, error)
  • func (c *Client) HandleFetches(ctx context.Context) error
  • func (c *Client) Name() string
  • func (c *Client) Ping(ctx context.Context) error
  • func (c *Client) Produce(ctx context.Context, record *kgo.Record, promise PromiseFunc)
  • 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 PromiseFunc)
• type GroupTransactSession
  • func NewGroupTransactSession(opts ...Opt) (*GroupTransactSession, error)
  • func (g *GroupTransactSession) HandleFetches(ctx context.Context) error
  • func (g *GroupTransactSession) Name() string
  • func (g *GroupTransactSession) Ping(ctx context.Context) error
  • func (g *GroupTransactSession) Produce(ctx context.Context, record *kgo.Record, promise PromiseFunc)
  • 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 PromiseFunc)
• type Hook
• type HookClientClosed
• type HookFetchError
• type HookGroupTransactSessionClosed
• type HookHandleEnd
• type HookHandleStart
• type HookNewClient
• type HookNewGroupTransactSession
• type HookOffsetCommit
• type HookProduceEnd
• type HookProduceError
• type HookProduceRecord
• type HookProduceStart
• type HookShareAck
• type HookShareAckFlush
• type HookTransactionEnd
• type HookTransactionStart
• type Opt
  • func WithBatchHandler(handler BatchHandlerFunc) Opt
  • func WithGroupTransactSessionBatchHandler(handler BatchTxHandlerFunc) Opt
  • func WithHooks(hooks ...Hook) Opt
  • func WithKafkaOptions(opts ...kgo.Opt) Opt
  • func WithLogger(logger kgo.Logger) Opt
  • func WithMaxPollRecords(maxPollRecords int) Opt
  • func WithMaxRetries(maxRetries int) Opt
  • func WithName(name string) Opt
  • func WithPollInterval(interval time.Duration) Opt
  • func WithProducePromise(promise PromiseFunc) 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 PromiseFunc
• type ShareAckOutcome
• type TransactionOutcome
• type TransactionType
• type Tx
  • func (tx *Tx) Produce(ctx context.Context, record *kgo.Record, promise PromiseFunc)
  • func (tx *Tx) ProduceSync(ctx context.Context, records ...*kgo.Record) error
  • func (tx *Tx) TryProduce(ctx context.Context, record *kgo.Record, promise PromiseFunc)
• type TxFunc

Types

type BatchHandlerFunc

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

BatchHandlerFunc processes a batch of consumed Kafka records.

Returning nil signals successful processing. For regular consumers, returning an error triggers the configured retry behavior. For Share Groups, returning an error applies release or rejection semantics. Panics are recovered and handled as errors.

type BatchTxHandlerFunc

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

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

Produced records should be written through tx. Returning nil allows GroupTransactSession to attempt to commit the produced records together with the consumed offsets. Returning an error causes an abort to be attempted. Panics are recovered and handled as errors.

type Client

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

Client provides Kafka produce, consume, and producer transaction operations.

Consumption is driven by Client.HandleFetches.

func NewClient

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

NewClient creates a Client configured with opts.

Configure WithBatchHandler to use Client.HandleFetches. Consumer group and Share Group configurations require a batch handler at construction time.

func (*Client) HandleFetches

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

HandleFetches polls Kafka and dispatches fetched records to the configured batch handler.

It returns ctx.Err() when ctx is canceled, nil when the client is shut down, or a terminal processing error. Only one HandleFetches call may run at a time.

func (*Client) Name

func (c *Client) Name() string

Name returns the logical client name configured with WithName.

It returns an empty string if no name was configured.

func (*Client) Ping

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

Ping verifies that at least one Kafka broker is reachable.

func (*Client) Produce

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

Produce enqueues record for asynchronous delivery.

promise is called when delivery completes. If promise is nil, the callback configured with WithProducePromise is used, if any.

Produce may wait for producer buffer space. Use Client.TryProduce when enqueueing must not wait for buffer availability.

func (*Client) ProduceSync

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

ProduceSync produces records and waits for all of them to complete.

It returns the first produce error, if any.

func (*Client) RunInTx

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

RunInTx executes fn inside a producer transaction.

The client must be configured with a transactional ID through WithKafkaOptions. If fn returns nil, RunInTx flushes buffered records and attempts to commit the transaction. If fn or the pre-commit flush returns an error, RunInTx attempts to abort the transaction. Commit failures are handled according to franz-go transaction recovery semantics.

If fn panics, RunInTx recovers the panic, converts it to an error, and attempts to abort the transaction. If the panic value is an error, the returned error wraps it.

Produce transactional records through the Tx passed to fn. For consume-process-produce exactly-once workflows, use GroupTransactSession.

func (*Client) Shutdown

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

Shutdown stops polling, flushes pending producer records and Share Group acknowledgements, and closes the underlying Kafka client.

Shutdown is idempotent and safe to call concurrently. Only the first call performs shutdown; subsequent calls return the same result.

ctx bounds flushing of producer records and Share Group acknowledgements. The underlying client is closed even if flushing returns an error.

func (*Client) TryProduce

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

TryProduce attempts to enqueue record without waiting for producer buffer space.

If the producer buffer is full, delivery fails immediately. Other delivery semantics are the same as Client.Produce.

type GroupTransactSession

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

GroupTransactSession provides transactional Kafka consume-process-produce operations.

Each batch processed by GroupTransactSession.HandleFetches runs inside a group transaction so produced records and consumed offsets can be committed atomically.

func NewGroupTransactSession

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

NewGroupTransactSession creates a GroupTransactSession configured with opts.

A batch transaction handler configured with WithGroupTransactSessionBatchHandler is required. Kafka-to-Kafka EOS also requires a consumer group and transactional ID configured through WithKafkaOptions.

func (*GroupTransactSession) HandleFetches

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

HandleFetches polls Kafka and processes each fetched batch inside a group transaction.

If the handler succeeds, the session attempts to commit produced records and consumed offsets atomically. If the handler returns an error or panics, the session attempts to abort the transaction and waits for the configured suspension delay before continuing.

Transaction begin or end errors terminate the fetch loop.

HandleFetches runs until ctx is canceled, the session is shut down, or processing returns a terminal error. Only one HandleFetches call may run at a time.

func (*GroupTransactSession) Name

func (g *GroupTransactSession) Name() string

Name returns the logical session name configured with WithName.

It returns an empty string if no name was configured.

func (*GroupTransactSession) Ping

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

Ping verifies that at least one Kafka broker is reachable.

func (*GroupTransactSession) Produce

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

Produce enqueues record for asynchronous delivery using the session producer.

Delivery and backpressure semantics are the same as Client.Produce.

func (*GroupTransactSession) ProduceSync

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

ProduceSync produces records and waits for all of them to complete.

Error semantics are the same as Client.ProduceSync.

func (*GroupTransactSession) Shutdown

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

Shutdown stops polling and closes the underlying group transaction session.

Shutdown is idempotent and safe to call concurrently. Only the first call performs shutdown.

func (*GroupTransactSession) TryProduce

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

TryProduce attempts to enqueue record without waiting for producer buffer space.

Delivery semantics are the same as Client.TryProduce.

type Hook

type Hook any

Hook may implement one or more xkafka hook interfaces.

Register hooks with WithHooks. Hooks are called in registration order. Context-returning hooks are chained, so each hook receives the context returned by the previous hook.

Hook implementations must be safe for concurrent use and should return quickly.

type HookClientClosed

type HookClientClosed interface {
	OnClientClosed(*Client)
}

HookClientClosed is called once after Client.Shutdown closes a Client.

type HookFetchError

type HookFetchError interface {
	OnFetchError(ctx context.Context, topic string, partition int32, recoverable bool, err error)
}

HookFetchError is called for each error reported while fetching records.

recoverable reports whether the individual error is classified as non-terminal.

type HookGroupTransactSessionClosed

type HookGroupTransactSessionClosed interface {
	OnGroupTransactSessionClosed(*GroupTransactSession)
}

HookGroupTransactSessionClosed is called once after GroupTransactSession.Shutdown closes a GroupTransactSession.

type HookHandleEnd

type HookHandleEnd interface {
	OnHandleEnd(ctx context.Context, records []*kgo.Record, duration time.Duration, err error)
}

HookHandleEnd is called after a handler attempt completes.

duration measures handler execution. err contains the handler error, including an error converted from a recovered handler panic.

type HookHandleStart

type HookHandleStart interface {
	OnHandleStart(ctx context.Context, records []*kgo.Record) context.Context
}

HookHandleStart is called before records are passed to the configured handler.

The returned context is passed to subsequent handler hooks and the handler.

type HookNewClient

type HookNewClient interface {
	OnNewClient(*Client)
}

HookNewClient is called after NewClient successfully creates a Client.

type HookNewGroupTransactSession

type HookNewGroupTransactSession interface {
	OnNewGroupTransactSession(*GroupTransactSession)
}

HookNewGroupTransactSession is called after NewGroupTransactSession successfully creates a GroupTransactSession.

type HookOffsetCommit

type HookOffsetCommit interface {
	OnOffsetCommit(ctx context.Context, duration time.Duration, err error)
}

HookOffsetCommit is called after each manual consumer offset commit attempt.

duration measures the commit attempt.

type HookProduceEnd

type HookProduceEnd interface {
	OnProduceEnd(ctx context.Context, records []*kgo.Record, duration time.Duration, err error)
}

HookProduceEnd is called after a synchronous produce operation completes.

duration measures the underlying synchronous produce operation. err is non-nil if any record failed to produce.

type HookProduceError

type HookProduceError interface {
	OnProduceError(record *kgo.Record, err error)
}

HookProduceError is called for each record that fails to produce.

type HookProduceRecord

type HookProduceRecord interface {
	OnProduceRecord(ctx context.Context, record *kgo.Record)
}

HookProduceRecord is called before a record is passed to franz-go for producing.

It is called for both synchronous and asynchronous produce operations. For synchronous produces, ctx is the context returned by the HookProduceStart chain. For asynchronous produces, record.Context is used when non-nil; otherwise the produce call context is used.

Implementations may modify record before it is passed to franz-go, for example to inject propagation headers.

type HookProduceStart

type HookProduceStart interface {
	OnProduceStart(ctx context.Context, records []*kgo.Record) context.Context
}

HookProduceStart is called before a synchronous produce operation starts.

The returned context is passed to subsequent produce hooks and the underlying produce operation.

type HookShareAck

type HookShareAck interface {
	OnShareAck(ctx context.Context, outcome ShareAckOutcome, recordCount int)
}

HookShareAck is called when an acknowledgement outcome is assigned to one or more Share Group records.

recordCount is the number of records assigned outcome. The hook is not called when recordCount is zero.

type HookShareAckFlush

type HookShareAckFlush interface {
	OnShareAckFlush(ctx context.Context, duration time.Duration, err error)
}

HookShareAckFlush is called after each Share Group acknowledgement flush attempt.

duration measures the flush attempt.

type HookTransactionEnd

type HookTransactionEnd interface {
	OnTransactionEnd(ctx context.Context, transactionType TransactionType, outcome TransactionOutcome, duration time.Duration, err error)
}

HookTransactionEnd is called after a transaction attempt has finished, including cleanup.

outcome describes the final transaction result. duration covers the entire transaction attempt, including cleanup. err reports the associated error, if any.

type HookTransactionStart

type HookTransactionStart interface {
	OnTransactionStart(ctx context.Context, transactionType TransactionType) context.Context
}

HookTransactionStart is called when a transaction attempt starts.

The returned context is passed to subsequent transaction hooks and transaction processing.

type Opt

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

Opt configures Client and GroupTransactSession behavior.

func WithBatchHandler

func WithBatchHandler(handler BatchHandlerFunc) Opt

WithBatchHandler sets the batch handler used by Client.HandleFetches.

For regular consumers, handler errors are retried according to WithMaxRetries and WithSuspendProcessingTimeout. Share Group handler errors use release or rejection semantics.

func WithGroupTransactSessionBatchHandler

func WithGroupTransactSessionBatchHandler(handler BatchTxHandlerFunc) Opt

WithGroupTransactSessionBatchHandler sets the batch handler used by GroupTransactSession.HandleFetches.

A handler is required when creating a GroupTransactSession.

func WithHooks

func WithHooks(hooks ...Hook) Opt

WithHooks registers hooks for xkafka runtime events.

See Hook for hook behavior and concurrency requirements.

func WithKafkaOptions

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

WithKafkaOptions adds native franz-go options to the underlying Kafka client.

func WithLogger

func WithLogger(logger kgo.Logger) Opt

WithLogger sets the logger used by xkafka and franz-go.

Logging is disabled by default.

func WithMaxPollRecords

func WithMaxPollRecords(maxPollRecords int) Opt

WithMaxPollRecords sets the maximum number of buffered records processed in one poll iteration.

The default is 100. Values less than or equal to zero process all currently buffered records.

func WithMaxRetries

func WithMaxRetries(maxRetries int) Opt

WithMaxRetries sets the maximum number of retries after the initial regular batch handler attempt.

The default is unlimited retries. A value of zero disables retries. Negative values are ignored.

This option does not apply to Share Groups or GroupTransactSession handlers.

func WithName

func WithName(name string) Opt

WithName sets the logical client name used for observability and as the Kafka client ID.

Leading and trailing whitespace is removed. A non-empty name overrides a client ID configured through WithKafkaOptions.

func WithPollInterval

func WithPollInterval(interval time.Duration) Opt

WithPollInterval sets the interval between consumer poll iterations.

The default is 1 second. Non-positive values are ignored.

func WithProducePromise

func WithProducePromise(promise PromiseFunc) Opt

WithProducePromise sets the default callback for asynchronous produce operations.

The callback is used when an asynchronous produce call is made with a nil promise.

func WithShareRejectAfterDeliveries

func WithShareRejectAfterDeliveries(deliveries int32) Opt

WithShareRejectAfterDeliveries rejects a failed Share Group record when its delivery count reaches deliveries.

The default is zero, which disables rejection and releases failed records for redelivery. Negative values are ignored.

func WithShareReleaseTimeout

func WithShareReleaseTimeout(timeout time.Duration) Opt

WithShareReleaseTimeout sets the delay before flushing Share Group acknowledgements when records are released for redelivery.

The default is zero. Negative values are ignored.

func WithSuspendCommittingTimeout

func WithSuspendCommittingTimeout(timeout time.Duration) Opt

WithSuspendCommittingTimeout sets the delay between manual offset commit retries.

The default is 10 seconds. A value of zero retries immediately. Negative values are ignored.

func WithSuspendProcessingTimeout

func WithSuspendProcessingTimeout(timeout time.Duration) Opt

WithSuspendProcessingTimeout sets the delay after a handler failure.

The delay applies between regular consumer retries and after GroupTransactSession handler errors. The default is 30 seconds.

A value of zero disables the delay. Negative values are ignored.

type PromiseFunc

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

PromiseFunc is called when an asynchronous produce operation completes.

err is nil when the record was produced successfully and non-nil when delivery failed.

type ShareAckOutcome

type ShareAckOutcome string

ShareAckOutcome describes the outcome assigned to a Share Group record.

const (
	// ShareAckAccept indicates that a record was successfully processed.
	ShareAckAccept ShareAckOutcome = "accept"

	// ShareAckRelease indicates that a record was released for redelivery.
	ShareAckRelease ShareAckOutcome = "release"

	// ShareAckReject indicates that a record was rejected.
	ShareAckReject ShareAckOutcome = "reject"
)

type TransactionOutcome

type TransactionOutcome string

TransactionOutcome describes how a transaction attempt ended.

const (
	// TransactionOutcomeCommit indicates that the transaction committed
	// successfully.
	TransactionOutcomeCommit TransactionOutcome = "commit"

	// TransactionOutcomeAbort indicates that the transaction was aborted
	// successfully.
	TransactionOutcomeAbort TransactionOutcome = "abort"

	// TransactionOutcomeError indicates that the transaction attempt ended with
	// an error and was not reported as a successful commit or abort.
	TransactionOutcomeError TransactionOutcome = "error"
)

type TransactionType

type TransactionType string

TransactionType identifies the kind of transaction reported to transaction hooks.

const (
	// TransactionTypeProducer identifies a producer transaction started by
	// Client.RunInTx.
	TransactionTypeProducer TransactionType = "producer"

	// TransactionTypeGroup identifies a consume-process-produce transaction
	// managed by GroupTransactSession.
	TransactionTypeGroup TransactionType = "group"
)

type Tx

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

Tx provides producing operations within an active Kafka transaction.

A Tx is provided to TxFunc and BatchTxHandlerFunc. Transaction commit and abort are managed by Client.RunInTx or GroupTransactSession.

func (*Tx) Produce

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

Produce enqueues record for asynchronous delivery within the current transaction.

Delivery and backpressure semantics are the same as Client.Produce.

func (*Tx) ProduceSync

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

ProduceSync produces records within the current transaction and waits for all of them to complete.

Error semantics are the same as Client.ProduceSync.

func (*Tx) TryProduce

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

TryProduce attempts to enqueue record within the current transaction without waiting for producer buffer space.

Delivery semantics are the same as Client.TryProduce.

type TxFunc

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

TxFunc performs work inside a producer transaction.

Returning nil allows Client.RunInTx to attempt to commit the transaction. Returning an error causes Client.RunInTx to attempt to abort the transaction. Panics are recovered and handled as errors. If the panic value is an error, the returned error wraps it.