kafka

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 9, 2026 License: MIT Imports: 37 Imported by: 0

README

kafka

CI CodeQL Coverage Mutation Documentation Go Reference Release Go License

kafka provides bounded, explicit Apache Kafka producer, consumer, replay, inspection, and transactional building blocks for Go services. It wraps franz-go without hiding delivery outcomes, lifecycle ownership, topic policy, or security configuration.

The package provides at-least-once building blocks. It does not make database writes and Kafka publication atomic, make consumer side effects exactly once, or own topic and broker configuration.

Installation

go get github.com/faustbrian/go-kafka

Quick start

Configuration can be validated during bootstrap without allocating a client or dialing brokers:

config := kafka.ProducerConfig{
	Brokers:       []string{"kafka.internal:9093"},
	ClientID:      "track-outbox",
	AllowedTopics: []string{"track.tracking-event.v1"},
}
if err := config.Validate(); err != nil {
	return err
}

producer, err := kafka.NewProducer(config)
if err != nil {
	return err
}
defer producer.Close()

return producer.Publish(ctx, kafka.Message{
	Topic: "track.tracking-event.v1",
	Key:   []byte(trackedItemID),
	Value: payload,
})

Guarantees and limits

  • Producers retain franz-go idempotence, require all in-sync replica acknowledgements, and bound retries, buffering, admission, and delivery.
  • Consumers expose explicit acknowledgement, retry, dead-letter, rebalance, and shutdown behavior.
  • Topic access is restricted through constructor-copied allowlists.
  • TLS 1.2 or later is the default; authentication and broker compatibility are explicit deployment decisions.
  • Ambiguous publish and commit outcomes remain distinguishable and require application reconciliation.
  • Service, OpenTelemetry, and Amazon MSK IAM integrations remain optional modules with caller-owned lifecycle.

Documentation

Start with the documentation index, compatibility matrix, specification decision register, delivery guarantees, and operations guide. The detailed reference preserves the complete producer, consumer, replay, inspection, and lifecycle contracts.

Browse the versioned Golib ecosystem index and its Integration and data movement family to compare Kafka with companion libraries and optional adapters.

Development

Run make check for the repository contract. Backend and managed-service changes must also pass the applicable integration and interoperability gates.

License

MIT. See LICENSE.

Documentation

Overview

Package kafka provides bounded Apache Kafka producer and consumer composition for Go services.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidSecurityConfig = errors.New(
		"kafka: client security configuration is invalid",
	)
	ErrCredentialProviderFailed = errors.New(
		"kafka: credential provider failed",
	)
	ErrCredentialProviderPanic = errors.New(
		"kafka: credential provider panicked",
	)
	ErrInvalidCredentials = errors.New(
		"kafka: credential provider returned invalid credentials",
	)
	ErrExpiredOAuthBearerToken = errors.New(
		"kafka: credential provider returned an expired OAuth bearer token",
	)
	ErrInvalidClientCertificateRequest = errors.New(
		"kafka: TLS client certificate request exceeds policy limits",
	)
	ErrInvalidTrustAnchors = errors.New(
		"kafka: trust-anchor provider returned invalid certificates",
	)
)
View Source 
var (
	ErrGroupIDRequired               = errors.New("kafka: consumer group ID is required")
	ErrGroupIDTooLarge               = errors.New("kafka: consumer group ID exceeds configured limit")
	ErrInvalidGroupID                = errors.New("kafka: consumer group ID is invalid")
	ErrInvalidInstanceID             = errors.New("kafka: consumer instance ID is invalid")
	ErrInvalidRack                   = errors.New("kafka: consumer rack is invalid")
	ErrInvalidBalancePolicy          = errors.New("kafka: consumer balance policy is invalid")
	ErrInvalidRebalanceHandlerPolicy = errors.New(
		"kafka: consumer rebalance handler policy is invalid",
	)
	ErrTopicsRequired          = errors.New("kafka: at least one topic is required")
	ErrTooManyTopics           = errors.New("kafka: topic count exceeds configured limit")
	ErrDuplicateTopic          = errors.New("kafka: topic is duplicated")
	ErrInvalidOffsetPolicy     = errors.New("kafka: consumer offset policy is invalid")
	ErrHandlerRequired         = errors.New("kafka: consumer handler is required")
	ErrBatchHandlerRequired    = errors.New("kafka: consumer batch handler is required")
	ErrHandlerPanic            = errors.New("kafka: consumer handler panicked")
	ErrConsumerBusy            = errors.New("kafka: consumer runner is already active")
	ErrConsumerDraining        = errors.New("kafka: consumer is draining")
	ErrConsumerDrainActive     = errors.New("kafka: consumer drain is already active")
	ErrConsumerDrainIncomplete = errors.New(
		"kafka: consumer drain is incomplete",
	)
	ErrConsumerClosing        = errors.New("kafka: consumer is shutting down")
	ErrConsumerClosed         = errors.New("kafka: consumer is closed")
	ErrConsumerFatal          = errors.New("kafka: consumer entered a fatal state")
	ErrConsumerInstanceFenced = errors.New(
		"kafka: consumer static instance was fenced",
	)
	ErrConsumerShutdownActive = errors.New(
		"kafka: consumer shutdown is already active",
	)
	ErrConsumerShutdownIncomplete = errors.New(
		"kafka: consumer shutdown is incomplete",
	)
	ErrPausePartitionsRequired = errors.New(
		"kafka: at least one pause partition is required",
	)
	ErrTooManyPausedPartitions = errors.New(
		"kafka: paused partition count exceeds configured limit",
	)
	ErrInvalidPausePartition = errors.New(
		"kafka: pause partition is invalid",
	)
	ErrDuplicatePausePartition = errors.New(
		"kafka: pause partition is duplicated",
	)
	ErrTooManyAssignedPartitions = errors.New(
		"kafka: assigned partition count exceeds configured limit",
	)
	ErrTooManyFetchedRecords = errors.New(
		"kafka: fetched record count exceeds configured limit",
	)
	ErrInvalidAssignment = errors.New(
		"kafka: consumer assignment is invalid",
	)
	ErrConsumerOwnershipLost = errors.New(
		"kafka: consumer partition ownership was lost",
	)
	ErrConsumerRebalance = errors.New(
		"kafka: consumer handler canceled for a pending rebalance",
	)
	ErrPauseTopicNotSubscribed = errors.New(
		"kafka: pause topic is not subscribed",
	)
	ErrInvalidConsumerConfig = errors.New(
		"kafka: consumer configuration is outside bounded limits",
	)
)
View Source 
var (
	// ErrInvalidFailurePolicy identifies an incompatible or unbounded consumer
	// failure-handling configuration.
	ErrInvalidFailurePolicy = errors.New(
		"kafka: consumer failure policy is invalid",
	)
	// ErrInvalidFailureTarget identifies an invalid, unversioned, or
	// source-equal retry or dead-letter topic.
	ErrInvalidFailureTarget = errors.New(
		"kafka: consumer failure target is invalid",
	)
	// ErrFailurePublisherRequired identifies a publish strategy without its
	// narrow record publisher.
	ErrFailurePublisherRequired = errors.New(
		"kafka: consumer failure publisher is required",
	)
	// ErrFailureDelegateRequired identifies a delegated strategy without its
	// application callback.
	ErrFailureDelegateRequired = errors.New(
		"kafka: consumer failure delegate is required",
	)
	// ErrConsumerFailureStopped identifies a handler failure deliberately left
	// unsettled for Kafka redelivery.
	ErrConsumerFailureStopped = errors.New(
		"kafka: consumer failure stopped without settlement",
	)
	// ErrFailureAttemptsExhausted identifies a bounded in-process retry policy
	// that reached its final handler attempt.
	ErrFailureAttemptsExhausted = errors.New(
		"kafka: consumer failure attempts exhausted",
	)
	// ErrFailureBackoff identifies cancellation or failure while waiting for a
	// bounded in-process retry.
	ErrFailureBackoff = errors.New(
		"kafka: consumer failure retry backoff interrupted",
	)
	// ErrFailurePublish identifies a retry-topic or dead-letter publication
	// that did not receive a definite successful producer result.
	ErrFailurePublish = errors.New(
		"kafka: consumer failure publication failed",
	)
	// ErrFailureDelegate identifies an application failure delegate that did
	// not resolve the source record.
	ErrFailureDelegate = errors.New(
		"kafka: consumer failure delegate failed",
	)
	// ErrFailureCallbackPanic identifies a contained classifier, publisher, or
	// delegate panic.
	ErrFailureCallbackPanic = errors.New(
		"kafka: consumer failure callback panicked",
	)
	// ErrInvalidFailureClassification identifies a classifier result outside
	// the stable ErrorCategory set.
	ErrInvalidFailureClassification = errors.New(
		"kafka: consumer failure classification is invalid",
	)
	// ErrFailureRecordInvalid identifies a source, retry, or dead-letter record
	// whose Kafka metadata or material violates the configured bounded policy.
	ErrFailureRecordInvalid = errors.New(
		"kafka: consumer failure record is invalid",
	)
)
View Source 
var (
	ErrInspectionTargetsRequired  = errors.New("kafka: inspection targets are required")
	ErrTooManyInspectionTargets   = errors.New("kafka: inspection target count exceeds configured limit")
	ErrInvalidInspectionTarget    = errors.New("kafka: inspection target is invalid")
	ErrDuplicateInspectionTarget  = errors.New("kafka: inspection target is duplicated")
	ErrInvalidInspectorConfig     = errors.New("kafka: inspector configuration is outside bounded limits")
	ErrInvalidReadinessPolicy     = errors.New("kafka: inspector readiness policy is outside bounded limits")
	ErrInspectorClosed            = errors.New("kafka: inspector is closed")
	ErrInvalidInspectionResponse  = errors.New("kafka: broker inspection response is invalid")
	ErrInspectionResponseTooLarge = errors.New(
		"kafka: broker inspection response exceeds configured limits",
	)
	ErrInspectionTargetsFailed = errors.New(
		"kafka: one or more inspection targets failed",
	)
)
View Source 
var (
	// ErrInvalidObserverPolicy identifies an observer policy outside the
	// package's callback-count or deadline bounds.
	ErrInvalidObserverPolicy = errors.New(
		"kafka: observer policy is outside bounded limits",
	)
	// ErrObserverFailureHandlerRequired requires observer failures to have an
	// explicit reporting destination.
	ErrObserverFailureHandlerRequired = errors.New(
		"kafka: observer failure handler is required",
	)
	// ErrObserverPanic identifies a contained observer panic without retaining
	// or rendering its potentially sensitive panic value.
	ErrObserverPanic = errors.New("kafka: observer panicked")
	// ErrObserverReentry identifies an operation attempted with the context
	// supplied to an observer callback.
	ErrObserverReentry = errors.New("kafka: observer callback cannot re-enter client")
	// ErrInvalidObservation identifies metadata outside the stable public
	// observation contract.
	ErrInvalidObservation = errors.New("kafka: observation is invalid")
)
View Source 
var (
	ErrBrokersRequired           = errors.New("kafka: at least one broker is required")
	ErrTooManyBrokers            = errors.New("kafka: broker count exceeds configured limit")
	ErrInvalidBroker             = errors.New("kafka: broker address is invalid")
	ErrDuplicateBroker           = errors.New("kafka: broker address is duplicated")
	ErrClientIDRequired          = errors.New("kafka: client ID is required")
	ErrClientIDTooLarge          = errors.New("kafka: client ID exceeds configured limit")
	ErrInvalidClientID           = errors.New("kafka: client ID is invalid")
	ErrTopicRequired             = errors.New("kafka: topic is required")
	ErrTopicTooLarge             = errors.New("kafka: topic exceeds configured limit")
	ErrInvalidTopic              = errors.New("kafka: topic name is invalid")
	ErrTopicNotAllowed           = errors.New("kafka: topic is outside producer allowlist")
	ErrInvalidPartitionSelection = errors.New(
		"kafka: producer partition selection is invalid",
	)
	ErrKeyRequired          = errors.New("kafka: record key is required by producer policy")
	ErrKeyTooLarge          = errors.New("kafka: key exceeds configured limit")
	ErrValueTooLarge        = errors.New("kafka: value exceeds configured limit")
	ErrTooManyHeaders       = errors.New("kafka: header count exceeds configured limit")
	ErrHeaderKeyRequired    = errors.New("kafka: header key is required")
	ErrHeaderKeyTooLarge    = errors.New("kafka: header key exceeds configured limit")
	ErrHeaderValueTooLarge  = errors.New("kafka: header value exceeds configured limit")
	ErrHeadersTooLarge      = errors.New("kafka: headers exceed aggregate configured limit")
	ErrInvalidMessageLimits = errors.New(
		"kafka: all message limits must be positive",
	)
	ErrInvalidProducerConfig = errors.New(
		"kafka: producer configuration is outside bounded limits",
	)
	ErrInvalidCompressionPreference = errors.New(
		"kafka: producer compression preference is invalid",
	)
	ErrTransactionsDisabled      = errors.New("kafka: producer transactions are disabled")
	ErrTransactionRequired       = errors.New("kafka: transaction callback is required")
	ErrTransactionPanic          = errors.New("kafka: transaction callback panicked")
	ErrTransactionClosed         = errors.New("kafka: transaction is closed")
	ErrTransactionOutcomeUnknown = errors.New(
		"kafka: transaction commit outcome is unknown",
	)
	ErrDeliveryResultMissing = errors.New("kafka: producer omitted a delivery result")
	ErrDeliveryResultInvalid = errors.New("kafka: producer returned inconsistent delivery results")
	ErrRecordsRequired       = errors.New("kafka: at least one producer record is required")
	ErrTooManyBatchRecords   = errors.New("kafka: producer batch record count exceeds configured limit")
	ErrBatchTooLarge         = errors.New("kafka: producer batch exceeds configured byte limit")
	ErrBatchDeliveryFailed   = errors.New("kafka: one or more batch records failed delivery")
	ErrContextRequired       = errors.New("kafka: context is required")
	ErrProducerClosed        = errors.New("kafka: producer is closed")
	ErrProducerFatal         = errors.New("kafka: producer entered a fatal state")
	ErrProducerBusy          = errors.New("kafka: producer has in-flight operations")
	ErrTransactionInProgress = errors.New("kafka: producer transaction is in progress")
	ErrDrainIncomplete       = errors.New("kafka: producer drain is incomplete")
)
View Source 
var (
	ErrReplayRangesRequired    = errors.New("kafka: at least one replay range is required")
	ErrTooManyReplayRanges     = errors.New("kafka: replay range count exceeds configured limit")
	ErrInvalidReplayRange      = errors.New("kafka: replay range is invalid")
	ErrDuplicateReplayRange    = errors.New("kafka: replay range is duplicated")
	ErrInvalidReplayCheckpoint = errors.New(
		"kafka: replay checkpoint is invalid",
	)
	ErrDuplicateReplayCheckpoint = errors.New(
		"kafka: replay checkpoint position is duplicated",
	)
	ErrInvalidReplayConfig     = errors.New("kafka: replay configuration is outside bounded limits")
	ErrReplaySideEffectsDenied = errors.New(
		"kafka: replay side effects require explicit opt-in",
	)
	ErrReplayBusy               = errors.New("kafka: replay reader is already running")
	ErrReplayAlreadyRun         = errors.New("kafka: replay reader has already run")
	ErrReplayClosing            = errors.New("kafka: replay reader is shutting down")
	ErrReplayClosed             = errors.New("kafka: replay reader is closed")
	ErrReplayShutdownActive     = errors.New("kafka: replay shutdown is already active")
	ErrReplayShutdownIncomplete = errors.New("kafka: replay shutdown is incomplete")
	ErrUnexpectedReplayRecord   = errors.New(
		"kafka: replay returned a record outside the requested ranges",
	)
	ErrReplayOffsetGap        = errors.New("kafka: replay range contains an offset gap")
	ErrReplayOffsetOutOfRange = errors.New(
		"kafka: replay offset is outside broker retention bounds",
	)
	ErrReplayBoundsUnavailable = errors.New(
		"kafka: replay broker offset bounds are unavailable",
	)
	ErrReplayStalled = errors.New(
		"kafka: replay made no progress before its bounded deadline",
	)
)
View Source 
var (
	ErrInvalidReplayTimestampWindow = errors.New(
		"kafka: replay timestamp window is invalid",
	)
	ErrReplayTimestampRangeIncomplete = errors.New(
		"kafka: replay timestamp window may precede retained records",
	)
)
View Source 
var (
	ErrInvalidTransactionProcessorConfig = errors.New(
		"kafka: transaction processor configuration is invalid",
	)
	ErrTransactionHandlerRequired = errors.New(
		"kafka: transaction processor handler is required",
	)
	ErrTransactionNotCommitted = errors.New(
		"kafka: consume-transform-produce transaction was not committed",
	)
	ErrTooManyTransactionOutputRecords = errors.New(
		"kafka: transaction output record count exceeds configured limit",
	)
	ErrTransactionOutputTooLarge = errors.New(
		"kafka: transaction output bytes exceed configured limit",
	)
	ErrTransactionProcessorBusy = errors.New(
		"kafka: transaction processor runner is already active",
	)
	ErrTransactionProcessorClosing = errors.New(
		"kafka: transaction processor is shutting down",
	)
	ErrTransactionProcessorClosed = errors.New(
		"kafka: transaction processor is closed",
	)
	ErrTransactionProcessorShutdownActive = errors.New(
		"kafka: transaction processor shutdown is already active",
	)
	ErrTransactionProcessorShutdownIncomplete = errors.New(
		"kafka: transaction processor shutdown is incomplete",
	)
	ErrTransactionProcessorFatal = errors.New(
		"kafka: transaction processor entered a fatal state",
	)
)
View Source 
var ErrFetchBatchMalformed = errors.New(
	"kafka: fetched record batch compression is malformed",
)

ErrFetchBatchMalformed identifies compressed broker data that cannot be decoded as the Kafka compression codec declared by the record batch.

View Source 
var ErrFetchBatchTooLarge = errors.New(
	"kafka: fetched record batch exceeds configured decoded byte limit",
)

ErrFetchBatchTooLarge identifies a Kafka record batch whose decoded bytes exceed the configured consumer-side safety limit.

View Source 
var ErrFetchDecompressedBufferFull = errors.New(
	"kafka: fetched record batches exceed configured decoded buffer limit",
)

ErrFetchDecompressedBufferFull identifies a fetch whose decoded bytes would exceed the configured client-wide active decompression budget.

View Source 
var (
	// ErrInvalidFailureBatch identifies a batch that cannot be safely retained
	// or whose records do not match its ordered topic-partition coordinates.
	ErrInvalidFailureBatch = errors.New(
		"kafka: consumer failure batch is invalid",
	)
)
View Source 
var ErrInvalidProtocolPolicy = errors.New("kafka: protocol policy is invalid")

Functions

This section is empty.

Types

type Authentication 

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

Authentication is an immutable, redacted authentication policy constructed by one of the New*Authentication functions.

func NewOAuthBearerAuthentication 

func NewOAuthBearerAuthentication(
	provider OAuthBearerProvider,
) Authentication

NewOAuthBearerAuthentication selects a rotating bounded OAUTHBEARER token provider.

func NewPlainAuthentication 

func NewPlainAuthentication(
	provider UsernamePasswordProvider,
) Authentication

NewPlainAuthentication selects rotating SASL/PLAIN credentials. Validation rejects a nil provider and any plaintext transport configuration.

func NewSCRAMSHA256Authentication 

func NewSCRAMSHA256Authentication(
	provider UsernamePasswordProvider,
) Authentication

NewSCRAMSHA256Authentication selects rotating SCRAM-SHA-256 credentials.

func NewSCRAMSHA512Authentication 

func NewSCRAMSHA512Authentication(
	provider UsernamePasswordProvider,
) Authentication

NewSCRAMSHA512Authentication selects rotating SCRAM-SHA-512 credentials.

func (Authentication) GoString 

func (authentication Authentication) GoString() string

GoString returns a stable redacted representation for %#v formatting.

func (Authentication) Method 

func (authentication Authentication) Method() AuthenticationMethod

Method returns the stable authentication method.

func (Authentication) String 

func (authentication Authentication) String() string

String returns a stable redacted representation.

type AuthenticationMethod 

type AuthenticationMethod uint8

AuthenticationMethod identifies one supported Kafka SASL mechanism. 

const (
	// AuthenticationNone disables SASL authentication.
	AuthenticationNone AuthenticationMethod = iota
	// AuthenticationPlain selects SASL/PLAIN over verified TLS.
	AuthenticationPlain
	// AuthenticationSCRAMSHA256 selects SCRAM-SHA-256 over verified TLS.
	AuthenticationSCRAMSHA256
	// AuthenticationSCRAMSHA512 selects SCRAM-SHA-512 over verified TLS.
	AuthenticationSCRAMSHA512
	// AuthenticationOAuthBearer selects OAUTHBEARER over verified TLS.
	AuthenticationOAuthBearer
)

func (AuthenticationMethod) String 

func (method AuthenticationMethod) String() string

String returns the stable authentication method name.

type BatchFailure 

type BatchFailure struct {
	Batch    ConsumedBatch
	Attempt  int
	Category ErrorCategory
	// contains filtered or unexported fields
}

BatchFailure is the synchronous whole-partition-batch failure-policy input. Batch bytes remain borrowed for the callback unless Retain is called. Cause is deliberately omitted from formatting and telemetry by the package.

func (BatchFailure) Cause 

func (failure BatchFailure) Cause() error

Cause returns the original batch-handler error for programmatic application decisions. Callers must not render it without applying their own redaction.

func (BatchFailure) Retain 

func (failure BatchFailure) Retain() BatchFailure

Retain returns a failure whose complete batch is deeply copied. Error identity is immutable by convention and is retained without wrapping.

type BatchFailureDelegate 

type BatchFailureDelegate interface {
	HandleBatchFailure(context.Context, BatchFailure) error
}

BatchFailureDelegate owns one terminal application-specific decision for a complete partition batch. A nil result resolves every record and permits the consumer to settle the batch. An error leaves every record unsettled. The callback must be synchronous, bounded, cancellation-aware, and concurrency- safe when the consumer permits concurrent partition handlers.

type BatchFailureDelegateFunc 

type BatchFailureDelegateFunc func(context.Context, BatchFailure) error

BatchFailureDelegateFunc adapts a function to BatchFailureDelegate.

func (BatchFailureDelegateFunc) HandleBatchFailure 

func (delegate BatchFailureDelegateFunc) HandleBatchFailure(
	ctx context.Context,
	failure BatchFailure,
) error

HandleBatchFailure invokes delegate.

type BatchFailureHandlerConfig 

type BatchFailureHandlerConfig struct {
	Handler         BatchHandler
	Classifier      FailureClassifier
	Retry           FailureRetryPolicy
	Mode            FailureMode
	Target          FailureTarget
	Publisher       BatchFailurePublisher
	Delegate        BatchFailureDelegate
	Limits          MessageLimits
	MaxBatchRecords int
	MaxBatchBytes   int64
	PublishTimeout  time.Duration
}

BatchFailureHandlerConfig defines a bounded whole-partition-batch failure decorator. Retry re-invokes the complete batch. Retry-topic and dead-letter modes publish the complete batch and resolve it only after every target delivery has a definite successful result. Configuration values are copied during construction; callbacks retain caller-owned lifetime and concurrency responsibilities.

func (BatchFailureHandlerConfig) Validate 

func (config BatchFailureHandlerConfig) Validate() error

Validate reports whether the batch failure policy is explicit, compatible, and bounded without constructing a handler.

type BatchFailurePublisher 

type BatchFailurePublisher interface {
	PublishBatch(context.Context, []ProducerRecord) ([]DeliveryResult, error)
}

BatchFailurePublisher is the narrow publication seam used to reroute a complete failed partition batch. Producer satisfies this interface. Results must remain input ordered and contain exactly one result per record. The publisher owns the supplied slice and record bytes and may retain them. It must be synchronous, bounded, cancellation-aware, and concurrency-safe when the consumer permits concurrent partition handlers.

type BatchFailurePublisherFunc 

type BatchFailurePublisherFunc func(
	context.Context,
	[]ProducerRecord,
) ([]DeliveryResult, error)

BatchFailurePublisherFunc adapts a function to BatchFailurePublisher.

func (BatchFailurePublisherFunc) PublishBatch 

func (publisher BatchFailurePublisherFunc) PublishBatch(
	ctx context.Context,
	records []ProducerRecord,
) ([]DeliveryResult, error)

PublishBatch invokes publisher.

type BatchHandler 

type BatchHandler interface {
	HandleBatch(context.Context, ConsumedBatch) error
}

BatchHandler durably processes one partition batch. A nil result settles the entire batch; any error settles none of it. Implementations must be concurrency-safe when the consumer permits more than one concurrent handler.

func NewBatchFailureHandler 

func NewBatchFailureHandler(
	config BatchFailureHandlerConfig,
) (BatchHandler, error)

NewBatchFailureHandler constructs a reusable whole-partition-batch failure decorator. Each invocation validates and retains the complete source batch before the wrapped handler runs, and each retry receives an isolated copy. Construction allocates no durable resources.

Example 
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/faustbrian/go-kafka"
)

func main() {
	attempts := 0
	handler, err := kafka.NewBatchFailureHandler(kafka.BatchFailureHandlerConfig{
		Handler: kafka.BatchHandlerFunc(func(
			context.Context,
			kafka.ConsumedBatch,
		) error {
			attempts++
			if attempts == 1 {
				return errors.New("dependency unavailable")
			}

			return nil
		}),
		Classifier: kafka.FailureClassifierFunc(func(error) kafka.ErrorCategory {
			return kafka.ErrorRetryable
		}),
		Retry: kafka.FailureRetryPolicy{
			MaxAttempts:    2,
			InitialBackoff: time.Millisecond,
			MaxBackoff:     time.Millisecond,
		},
	})
	if err != nil {
		panic(err)
	}

	err = handler.HandleBatch(context.Background(), kafka.ConsumedBatch{
		Topic: "events", Partition: 0,
		Records: []kafka.ConsumedRecord{
			{Topic: "events", Partition: 0, Offset: 7},
			{Topic: "events", Partition: 0, Offset: 8},
		},
	})
	fmt.Println(err == nil, attempts)

}

Output:
true 2

type BatchHandlerFunc 

type BatchHandlerFunc func(context.Context, ConsumedBatch) error

BatchHandlerFunc adapts a function to BatchHandler.

func (BatchHandlerFunc) HandleBatch 

func (handler BatchHandlerFunc) HandleBatch(
	ctx context.Context,
	batch ConsumedBatch,
) error

HandleBatch invokes handler.

type BrokerState 

type BrokerState struct {
	NodeID int32
	Host   string
	Port   int32
	Rack   string
}

BrokerState is bounded, copied metadata for one Kafka broker.

type ClientCertificateProvider 

type ClientCertificateProvider interface {
	ClientCertificate(context.Context, ClientCertificateRequest) (tls.Certificate, error)
}

ClientCertificateProvider returns a fresh mTLS certificate for one handshake. Implementations must be concurrency-safe and honor ctx.

type ClientCertificateProviderFunc 

type ClientCertificateProviderFunc func(
	context.Context,
	ClientCertificateRequest,
) (tls.Certificate, error)

ClientCertificateProviderFunc adapts a function to ClientCertificateProvider.

func (ClientCertificateProviderFunc) ClientCertificate 

func (provider ClientCertificateProviderFunc) ClientCertificate(
	ctx context.Context,
	request ClientCertificateRequest,
) (tls.Certificate, error)

ClientCertificate invokes provider.

type ClientCertificateRequest 

type ClientCertificateRequest struct {
	AcceptableCAs    [][]byte
	SignatureSchemes []tls.SignatureScheme
	Version          uint16
}

ClientCertificateRequest is an owned, bounded view of the TLS server's client-certificate request.

type ClientSecurity 

type ClientSecurity struct {
	Transport                 TransportSecurity
	TLS                       *tls.Config
	Authentication            Authentication
	ClientCertificateProvider ClientCertificateProvider
	TrustAnchorProvider       TrustAnchorProvider
	CredentialTimeout         time.Duration
}

ClientSecurity configures transport and optional authentication. The zero value uses verified TLS with system roots and a TLS 1.2 minimum. Construction clones mutable TLS slices, certificate bytes, and root pools. Interface values, private keys, TLS callbacks, and session caches remain caller-owned and must be immutable or concurrency-safe for the client's lifetime.

func DevelopmentPlaintextSecurity 

func DevelopmentPlaintextSecurity() ClientSecurity

DevelopmentPlaintextSecurity returns the explicit development-only unencrypted transport policy.

func (ClientSecurity) GoString 

func (security ClientSecurity) GoString() string

GoString returns a stable redacted representation for %#v formatting.

func (ClientSecurity) String 

func (security ClientSecurity) String() string

String returns a stable representation that cannot include credentials, certificates, roots, or callback internals.

func (ClientSecurity) Validate 

func (security ClientSecurity) Validate() error

Validate reports whether the transport and authentication policy is safe and internally consistent without allocating a Kafka client or invoking a credential provider.

type ClusterState 

type ClusterState struct {
	ID                string
	IDVisible         bool
	ControllerID      int32
	ControllerVisible bool
	Brokers           []BrokerState
}

ClusterState is bounded, copied Kafka cluster identity and broker metadata.

type CompressionCodec 

type CompressionCodec uint8

CompressionCodec identifies one Kafka record-batch compression algorithm. 

const (
	// CompressionNone disables record-batch compression.
	CompressionNone CompressionCodec = iota + 1
	// CompressionGzip selects gzip compression.
	CompressionGzip
	// CompressionSnappy selects snappy compression.
	CompressionSnappy
	// CompressionLz4 selects LZ4 compression.
	CompressionLz4
	// CompressionZstd selects Zstandard compression.
	CompressionZstd
)

func (CompressionCodec) String 

func (codec CompressionCodec) String() string

String returns the stable compression policy name.

type ConsumedBatch 

type ConsumedBatch struct {
	Topic     string
	Partition int32
	Records   []ConsumedRecord
}

ConsumedBatch contains one non-empty bounded set of records from a single topic partition. Records are ordered by offset. The slice is owned for the handler call, but record bytes remain borrowed unless Retain is used.

func (ConsumedBatch) Retain 

func (batch ConsumedBatch) Retain() ConsumedBatch

Retain returns a deep copy whose slice and record bytes the caller owns.

type ConsumedMessage 

type ConsumedMessage = ConsumedRecord

ConsumedMessage is retained as the pre-v1 name for ConsumedRecord.

type ConsumedRecord 

type ConsumedRecord struct {
	Topic         string
	Key           []byte
	Value         []byte
	Headers       []Header
	Timestamp     time.Time
	TimestampType TimestampType
	Partition     int32
	Offset        int64
	LeaderEpoch   int32
}

ConsumedRecord is one borrowed Kafka record. Key, value, header values, and the header slice remain valid only for the synchronous handler call unless Retain is used.

func (ConsumedRecord) Retain 

func (record ConsumedRecord) Retain() ConsumedRecord

Retain returns a deep copy whose bytes remain owned by the caller.

type Consumer 

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

Consumer processes records with explicit post-handler offset commits. One Run, RunOnce, or RunBatchOnce call may be active at a time. Handler callbacks can overlap only across independent partitions when MaxConcurrentHandlers is greater than one. Duplicate static-instance fencing permanently rejects new runners with ErrConsumerFatal and ErrConsumerInstanceFenced. Its methods are safe for concurrent lifecycle coordination.

func NewConsumer 

func NewConsumer(config ConsumerConfig) (*Consumer, error)

NewConsumer constructs a group consumer with automatic commits disabled and cooperative rebalancing blocked while each bounded poll is processed.

func (*Consumer) Assignment 

func (consumer *Consumer) Assignment() (ConsumerAssignment, error)

Assignment returns a sorted, copied snapshot of current assignment state. Its package-local epoch changes at every assign, revoke, or loss callback. Invalid or oversized broker-controlled callback metadata fails closed and is returned until the member loses its assignment and rejoins.

func (*Consumer) Close 

func (consumer *Consumer) Close() error

Close performs a bounded graceful shutdown using the configured timeout.

func (*Consumer) Drain 

func (consumer *Consumer) Drain(ctx context.Context) error

Drain stops an idle poll, lets an admitted poll finish processing and settlement, and waits for the active runner without leaving the group or closing the client. New runs are fenced until a successful drain completes. A context failure returns ErrConsumerDrainIncomplete and leaves the drain retriable. Concurrent drains return ErrConsumerDrainActive.

func (*Consumer) PausePartitions 

func (consumer *Consumer) PausePartitions(partitions ...TopicPartition) error

PausePartitions stops future fetches for explicit subscribed partitions. Records already buffered or returned by the current poll can still be processed. Pauses persist across rebalances until explicitly resumed.

func (*Consumer) PausedPartitions 

func (consumer *Consumer) PausedPartitions() []TopicPartition

PausedPartitions returns a sorted snapshot of explicitly paused partitions.

func (*Consumer) ResumePartitions 

func (consumer *Consumer) ResumePartitions(partitions ...TopicPartition) error

ResumePartitions resumes future fetches for explicit subscribed partitions. Partitions that are not paused are unchanged.

func (*Consumer) Run 

func (consumer *Consumer) Run(ctx context.Context, handler Handler) error

Run continuously executes bounded poll cycles until cancellation or the first processing failure. Context cancellation is a clean runner stop. It returns ErrContextRequired for a nil context, ErrConsumerBusy when another runner owns the consumer, ErrConsumerFatal and ErrConsumerInstanceFenced after static-membership fencing, and a lifecycle error once shutdown begins.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"os"
	"os/signal"
	"syscall"

	"github.com/faustbrian/go-kafka"
)

func main() {
	ctx, stop := signal.NotifyContext(
		context.Background(),
		os.Interrupt,
		syscall.SIGTERM,
	)
	defer stop()

	consumer, err := kafka.NewConsumer(kafka.ConsumerConfig{
		Brokers:       []string{"kafka.internal:9093"},
		ClientID:      "billing-projection",
		GroupID:       "billing-projection-v1",
		Topics:        []string{"orders.created.v1"},
		ResetOffset:   kafka.OffsetEarliest,
		BalancePolicy: kafka.BalanceCooperativeSticky,
	})
	if err != nil {
		log.Fatal(err)
	}

	runErr := consumer.Run(ctx, kafka.HandlerFunc(func(
		ctx context.Context,
		record kafka.ConsumedRecord,
	) error {
		return persistProjection(ctx, record)
	}))
	if err := errors.Join(runErr, consumer.Close()); err != nil {
		log.Fatal(err)
	}
}

func persistProjection(context.Context, kafka.ConsumedRecord) error {

	return nil
}

func (*Consumer) RunBatchOnce 

func (consumer *Consumer) RunBatchOnce(
	ctx context.Context,
	handler BatchHandler,
) (PollResult, error)

RunBatchOnce polls at most the configured record limit and invokes handler once per non-empty partition batch, with bounded concurrency across partitions. A successful batch commits its last record; a failed batch commits no record from that partition. Independent successful partition batches remain committable. A nil context returns ErrContextRequired. Static-membership fencing returns ErrConsumerFatal and ErrConsumerInstanceFenced, then permanently rejects later runner calls.

func (*Consumer) RunOnce 

func (consumer *Consumer) RunOnce(ctx context.Context, handler Handler) (PollResult, error)

RunOnce polls at most the configured record limit and processes each partition in fetch order, with bounded concurrency across partitions. Each partition stops at its first handler failure; successful contiguous prefixes from that partition and independent partitions are committed before the first handler error in stable poll-partition order is returned. It returns ErrContextRequired for a nil context, ErrConsumerBusy when another runner owns the consumer, ErrConsumerFatal and ErrConsumerInstanceFenced after static-membership fencing, and a lifecycle error once shutdown begins.

func (*Consumer) Shutdown 

func (consumer *Consumer) Shutdown(ctx context.Context) (err error)

Shutdown fences new runs, interrupts an idle poll without canceling admitted handlers, waits for their settlement, and closes the Kafka client. Dynamic members leave the group before close; static members preserve their membership window. A context or leave failure is joined with ErrConsumerShutdownIncomplete and leaves the consumer fenced so Shutdown can be retried. A nil context returns ErrContextRequired without fencing the consumer. Concurrent shutdown calls return ErrConsumerShutdownActive; shutdown during Drain returns ErrConsumerDrainActive.

type ConsumerAssignment 

type ConsumerAssignment struct {
	Epoch      uint64
	Partitions []TopicPartition
	Lost       bool
}

ConsumerAssignment is a copied snapshot of the member's current partition ownership. Epoch is a package-local lifecycle fence, not Kafka's broker generation ID. Lost reports that the latest lifecycle transition was a fatal ownership loss.

type ConsumerConfig 

type ConsumerConfig struct {
	Brokers               []string
	ClientID              string
	Protocol              ProtocolPolicy
	GroupID               string
	InstanceID            string
	Rack                  string
	Topics                []string
	ResetOffset           OffsetPolicy
	BalancePolicy         GroupBalancePolicy
	RebalanceHandler      RebalanceHandlerPolicy
	Limits                MessageLimits
	MaxPollRecords        int
	MaxPausedPartitions   int
	MaxAssignedPartitions int
	MaxConcurrentFetches  int
	// MaxConcurrentHandlers bounds simultaneous callbacks across independent
	// topic partitions. One partition always remains sequential. The zero
	// value defaults to one.
	MaxConcurrentHandlers int
	// FetchMinBytes is the minimum encoded record bytes a broker tries to
	// collect before answering a fetch. Zero defaults to one byte, values must
	// not exceed FetchMaxBytes, and FetchMaxWait bounds the wait.
	FetchMinBytes          int32
	FetchMaxBytes          int32
	FetchMaxPartitionBytes int32
	// BrokerMaxReadBytes is the hard maximum encoded Kafka response accepted
	// from one broker connection. It must be at least FetchMaxBytes.
	BrokerMaxReadBytes int32
	// MaxDecompressedBatchBytes is the hard decoded-byte limit for one Kafka
	// record batch before records are admitted to package handlers.
	MaxDecompressedBatchBytes int64
	// MaxBufferedDecompressedBytes bounds decoded compressed-batch memory held
	// across active and prefetched Kafka responses.
	MaxBufferedDecompressedBytes int64
	FetchMaxWait                 time.Duration

	SessionTimeout    time.Duration
	RebalanceTimeout  time.Duration
	HeartbeatInterval time.Duration
	HandlerTimeout    time.Duration
	CommitTimeout     time.Duration
	ShutdownTimeout   time.Duration
	DialTimeout       time.Duration
	Security          ClientSecurity
	Observers         ObserverPolicy
}

ConsumerConfig defines one bounded consumer-group member.

func (ConsumerConfig) Validate 

func (config ConsumerConfig) Validate() error

Validate reports whether the consumer configuration satisfies the bounded group policy without constructing a client or dialing brokers.

type ConsumerError 

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

ConsumerError classifies a consumer-group infrastructure failure without rendering its potentially sensitive cause. Handler errors are application failures and are not converted to ConsumerError values.

func (*ConsumerError) Category 

func (err *ConsumerError) Category() ErrorCategory

Category returns the stable operational category.

func (*ConsumerError) Error 

func (err *ConsumerError) Error() string

Error implements error with a stable redacted diagnostic.

func (*ConsumerError) Operation 

func (err *ConsumerError) Operation() ConsumerOperation

Operation returns the consumer-group phase that failed.

func (*ConsumerError) Retryable 

func (err *ConsumerError) Retryable() bool

Retryable reports whether a later bounded poll, commit, or shutdown attempt may succeed without changing application input or consumer configuration.

func (*ConsumerError) Unwrap 

func (err *ConsumerError) Unwrap() error

Unwrap returns the original failure for errors.Is and errors.As.

type ConsumerGroupInspectionResult 

type ConsumerGroupInspectionResult struct {
	Group    string
	State    ConsumerGroupState
	Category ErrorCategory
	Err      error
}

ConsumerGroupInspectionResult is one input-ordered consumer-group inspection outcome. State is populated only when Err is nil. Err retains the target-specific broker or policy failure, and Category provides its stable package classification.

type ConsumerGroupMemberState 

type ConsumerGroupMemberState struct {
	MemberID          string
	InstanceID        string
	InstanceIDVisible bool
	ClientID          string
	ClientHost        string
	Assignments       []TopicPartition
}

ConsumerGroupMemberState is bounded, copied classic consumer-group member identity and current partition assignment.

type ConsumerGroupPartitionLag 

type ConsumerGroupPartitionLag struct {
	Topic           string
	Partition       int32
	CommittedOffset int64
	StartOffset     int64
	EndOffset       int64
	Lag             int64
}

ConsumerGroupPartitionLag is one committed offset and broker end-offset comparison.

type ConsumerGroupState 

type ConsumerGroupState struct {
	Group         string
	CoordinatorID int32
	State         string
	ProtocolType  string
	Protocol      string
	Members       []ConsumerGroupMemberState
	Partitions    []ConsumerGroupPartitionLag
}

ConsumerGroupState is the current state and lag for one requested group.

type ConsumerOperation 

type ConsumerOperation uint8

ConsumerOperation identifies the consumer-group phase that failed. 

const (
	// ConsumerOperationPoll identifies a group poll or join-session failure.
	ConsumerOperationPoll ConsumerOperation = iota + 1
	// ConsumerOperationCommit identifies a source-offset commit failure.
	ConsumerOperationCommit
	// ConsumerOperationLeave identifies a graceful group-leave failure.
	ConsumerOperationLeave
)

func (ConsumerOperation) String 

func (operation ConsumerOperation) String() string

String returns a stable low-cardinality consumer operation name.

type ConsumerProtocolGroupInspectionResult 

type ConsumerProtocolGroupInspectionResult struct {
	Group string
	// State is populated only when Err is nil.
	State ConsumerProtocolGroupState
	// Category is zero on success and classifies Err on failure.
	Category ErrorCategory
	Err      error
}

ConsumerProtocolGroupInspectionResult is one input-ordered KIP-848 group inspection outcome. State is populated only when Err is nil. Err retains the target-specific broker or policy failure, and Category provides its stable package classification.

type ConsumerProtocolGroupMemberState 

type ConsumerProtocolGroupMemberState struct {
	MemberID string
	// InstanceID is the static member identity when InstanceIDVisible is true.
	InstanceID        string
	InstanceIDVisible bool
	// RackID is the member rack identity when RackIDVisible is true.
	RackID        string
	RackIDVisible bool
	MemberEpoch   int32
	MemberType    ConsumerProtocolMemberType
	ClientID      string
	ClientHost    string
	// SubscribedTopics is an owned, sorted explicit topic subscription.
	SubscribedTopics []string
	// SubscribedTopicRegex is the broker-side subscription expression when
	// SubscribedTopicRegexVisible is true. Kafka can expose an empty expression
	// for an explicit topic subscription, so visibility must be checked first.
	SubscribedTopicRegex        string
	SubscribedTopicRegexVisible bool
	// Assignments and TargetAssignments are owned and sorted by topic and
	// partition. They remain distinct while a member reconciles.
	Assignments       []TopicPartition
	TargetAssignments []TopicPartition
}

ConsumerProtocolGroupMemberState is bounded, copied KIP-848 consumer-group member state. Assignments are the member's current ownership; target assignments are the broker-selected state toward which it is reconciling.

type ConsumerProtocolGroupState 

type ConsumerProtocolGroupState struct {
	Group         string
	CoordinatorID int32
	// State retains Kafka's title-cased wire value.
	State string
	Epoch int32
	// AssignmentEpoch identifies the target assignment computed for Epoch.
	AssignmentEpoch int32
	Assignor        string
	// Members are owned and sorted by member ID.
	Members []ConsumerProtocolGroupMemberState
	// Partitions are owned and sorted by topic and partition.
	Partitions []ConsumerGroupPartitionLag
}

ConsumerProtocolGroupState is current KIP-848 group state and lag. Epoch is the group metadata epoch; AssignmentEpoch identifies the target assignment. Members may temporarily have lower epochs and current assignments while independently reconciling toward their target assignments.

type ConsumerProtocolMemberType 

type ConsumerProtocolMemberType int8

ConsumerProtocolMemberType identifies the KIP-848 member representation reported by Kafka. Unknown is retained because version-zero describe responses do not expose the member type. 

const (
	// ConsumerProtocolMemberTypeUnknown means the response version omitted the
	// member type.
	ConsumerProtocolMemberTypeUnknown ConsumerProtocolMemberType = -1
	// ConsumerProtocolMemberTypeClassic identifies a classic member migrating
	// within a KIP-848 consumer group.
	ConsumerProtocolMemberTypeClassic ConsumerProtocolMemberType = 0
	// ConsumerProtocolMemberTypeConsumer identifies a KIP-848 consumer member.
	ConsumerProtocolMemberTypeConsumer ConsumerProtocolMemberType = 1
)

type DeliveryError 

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

DeliveryError classifies one producer delivery failure. Error deliberately omits the underlying diagnostic so endpoints, credentials, record bytes, and headers cannot be rendered accidentally. Unwrap preserves programmatic error identity for callers that intentionally inspect the cause.

func (*DeliveryError) Category 

func (err *DeliveryError) Category() ErrorCategory

Category returns the stable operational category.

func (*DeliveryError) Error 

func (err *DeliveryError) Error() string

Error implements error with a stable redacted diagnostic.

func (*DeliveryError) Retryable 

func (err *DeliveryError) Retryable() bool

Retryable reports whether the package classified the failure as transient.

func (*DeliveryError) Unwrap 

func (err *DeliveryError) Unwrap() error

Unwrap returns the original failure for errors.Is and errors.As.

type DeliveryResult 

type DeliveryResult struct {
	Topic     string
	Partition int32
	Offset    int64
	Timestamp time.Time
	Err       error
}

DeliveryResult reports the broker outcome for one produced record. A nil Err means Kafka acknowledged the record under the configured acknowledgement policy; it does not mean an application side effect consumed the record.

type ErrorCategory 

type ErrorCategory uint8

ErrorCategory is a stable operational classification independent of franz-go and Kafka protocol error types. 

const (
	// ErrorPermanent identifies a definite failure that policy must not retry
	// without changing input or configuration.
	ErrorPermanent ErrorCategory = iota + 1
	// ErrorRetryable identifies a transient broker or transport failure.
	ErrorRetryable
	// ErrorAuthorization identifies authentication or authorization denial.
	ErrorAuthorization
	// ErrorFenced identifies loss of producer, transaction, or group ownership.
	ErrorFenced
	// ErrorOversized identifies a broker-side record or batch size rejection.
	ErrorOversized
	// ErrorTimeout identifies expiry of a bounded operation.
	ErrorTimeout
	// ErrorCanceled identifies caller cancellation.
	ErrorCanceled
	// ErrorShutdown identifies an operation rejected or failed by client close.
	ErrorShutdown
	// ErrorAmbiguous identifies an operation whose durable outcome is unknown.
	ErrorAmbiguous
	// ErrorFatal identifies producer state that cannot safely continue.
	ErrorFatal
)
const ErrorUnknown ErrorCategory = 0

ErrorUnknown is the zero value used when no failure category applies or a caller supplies an unrecognized category.

func (ErrorCategory) String 

func (category ErrorCategory) String() string

String returns a stable low-cardinality category name.

type FailureClassifier 

type FailureClassifier interface {
	ClassifyFailure(error) ErrorCategory
}

FailureClassifier maps an application handler error to one stable, low-cardinality operational category. Implementations must be synchronous, bounded, concurrency-safe, and must not render record data or credentials.

type FailureClassifierFunc 

type FailureClassifierFunc func(error) ErrorCategory

FailureClassifierFunc adapts a function to FailureClassifier.

func (FailureClassifierFunc) ClassifyFailure 

func (classifier FailureClassifierFunc) ClassifyFailure(err error) ErrorCategory

ClassifyFailure invokes classifier.

type FailureDelegate 

type FailureDelegate interface {
	HandleFailure(context.Context, HandlerFailure) error
}

FailureDelegate owns a terminal application-specific failure decision. A nil result declares the source record resolved and permits normal consumer settlement. An error leaves the source record unsettled.

type FailureDelegateFunc 

type FailureDelegateFunc func(context.Context, HandlerFailure) error

FailureDelegateFunc adapts a function to FailureDelegate.

func (FailureDelegateFunc) HandleFailure 

func (delegate FailureDelegateFunc) HandleFailure(
	ctx context.Context,
	failure HandlerFailure,
) error

HandleFailure invokes delegate.

type FailureHandlerConfig 

type FailureHandlerConfig struct {
	Handler        Handler
	Classifier     FailureClassifier
	Retry          FailureRetryPolicy
	Mode           FailureMode
	Target         FailureTarget
	Publisher      FailurePublisher
	Delegate       FailureDelegate
	Limits         MessageLimits
	PublishTimeout time.Duration
}

FailureHandlerConfig defines a bounded failure-policy handler decorator. Configuration is validated and copied before construction. Callback implementations retain their own lifetime and concurrency ownership.

func (FailureHandlerConfig) Validate 

func (config FailureHandlerConfig) Validate() error

Validate reports whether the failure policy is explicit, compatible, and bounded without constructing a handler.

type FailureHandlingError 

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

FailureHandlingError reports a redacted failure-policy outcome while preserving sentinel, handler, and operation error identity through Unwrap.

func (*FailureHandlingError) Attempt 

func (err *FailureHandlingError) Attempt() int

Attempt returns the last handler attempt involved in this outcome.

func (*FailureHandlingError) Category 

func (err *FailureHandlingError) Category() ErrorCategory

Category returns the original handler failure classification.

func (*FailureHandlingError) DeliveryResults 

func (err *FailureHandlingError) DeliveryResults() []DeliveryResult

DeliveryResults returns owned input-ordered retry-topic or dead-letter delivery outcomes when a whole-batch publication failed. Other failure outcomes return nil.

func (*FailureHandlingError) Error 

func (err *FailureHandlingError) Error() string

Error returns a stable diagnostic that excludes topics, keys, payloads, headers, credentials, and callback error text.

func (*FailureHandlingError) Stage 

func (err *FailureHandlingError) Stage() FailureStage

Stage returns the bounded phase that failed.

func (*FailureHandlingError) Unwrap 

func (err *FailureHandlingError) Unwrap() []error

Unwrap preserves programmatic error identity without rendering causes.

type FailureMode 

type FailureMode uint8

FailureMode selects the terminal action after the original handler and any bounded in-process attempts fail. 

const (
	// FailureModeStop returns a redacted error and leaves the source offset
	// unsettled. This is the zero value and preserves at-least-once redelivery.
	FailureModeStop FailureMode = iota
	// FailureModeRetryTopic publishes an owned copy to one explicit versioned retry
	// topic. A successful publish resolves the handler so the caller may settle
	// the source offset as a separate Kafka effect.
	FailureModeRetryTopic
	// FailureModeDeadLetter publishes an owned copy to one explicit versioned
	// dead-letter topic. Source settlement remains a separate Kafka effect.
	FailureModeDeadLetter
	// FailureModeDelegate transfers the terminal decision to one synchronous
	// application callback. A nil delegate result explicitly resolves the
	// source handler; an error leaves it unsettled.
	FailureModeDelegate
)

type FailurePublisher 

type FailurePublisher interface {
	PublishRecord(context.Context, ProducerRecord) DeliveryResult
}

FailurePublisher is the narrow publication seam used for non-transactional retry and dead-letter topics. Producer satisfies this interface. A custom implementation owns the supplied record and must return one definite delivery result before returning.

type FailureRetryPolicy 

type FailureRetryPolicy struct {
	MaxAttempts    int
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	Categories     []ErrorCategory
}

FailureRetryPolicy bounds optional in-process handler retries. MaxAttempts includes the initial handler call. The zero value performs one attempt.

type FailureStage 

type FailureStage uint8

FailureStage identifies the bounded phase that failed. It is stable and low-cardinality. 

const (
	FailureStageStop FailureStage = iota + 1
	FailureStageClassify
	FailureStageBackoff
	FailureStagePublish
	FailureStageDelegate
)

func (FailureStage) String 

func (stage FailureStage) String() string

String returns the stable failure-stage name.

type FailureTarget 

type FailureTarget struct {
	Topic   string
	Version uint16
}

FailureTarget identifies one explicit versioned retry or dead-letter topic. Version describes the application's topic/envelope contract version and is propagated in package-owned failure metadata.

type GroupBalancePolicy 

type GroupBalancePolicy uint8

GroupBalancePolicy selects the consumer-group partition assignment and rebalance protocol. 

const (
	// BalanceCooperativeSticky is the safe default for new groups and avoids
	// revoking every assignment during a rebalance.
	BalanceCooperativeSticky GroupBalancePolicy = iota
	// BalanceEagerSticky revokes all assignments during each rebalance for
	// compatibility with eager group members.
	BalanceEagerSticky
	// BalanceEagerToCooperative advertises eager sticky first and cooperative
	// sticky second for the first rolling deployment of a migration. A second
	// deployment must select BalanceCooperativeSticky.
	BalanceEagerToCooperative
)

type Handler 

type Handler interface {
	Handle(context.Context, ConsumedMessage) error
}

Handler durably processes one consumed message before its offset may be committed. Implementations must be concurrency-safe when the consumer permits more than one concurrent handler.

func NewFailureHandler 

func NewFailureHandler(config FailureHandlerConfig) (Handler, error)

NewFailureHandler constructs a reusable handler decorator implementing explicit stop, bounded retry, retry-topic, dead-letter, or delegated failure policy. Each invocation validates the source metadata and record limits before retaining bytes or calling the wrapped handler, then gives every attempt an isolated copy so handler mutation cannot alter later attempts or failure publication. Construction allocates no durable resources.

type HandlerFailure 

type HandlerFailure struct {
	Record   ConsumedRecord
	Attempt  int
	Category ErrorCategory
	// contains filtered or unexported fields
}

HandlerFailure is the synchronous failure-policy input. Record bytes remain borrowed for the callback unless Retain is called. Cause is deliberately omitted from formatting and telemetry by the package.

func (HandlerFailure) Cause 

func (failure HandlerFailure) Cause() error

Cause returns the original handler error for programmatic application decisions. Callers must not render it without applying their own redaction.

func (HandlerFailure) Retain 

func (failure HandlerFailure) Retain() HandlerFailure

Retain returns a failure whose record bytes are deeply copied. Error identity is immutable by convention and is retained without wrapping.

type HandlerFunc 

type HandlerFunc func(context.Context, ConsumedMessage) error

HandlerFunc adapts a function to Handler.

func (HandlerFunc) Handle 

func (handler HandlerFunc) Handle(ctx context.Context, message ConsumedMessage) error

Handle invokes handler. 

type Header struct {
	Key   string
	Value []byte
}

Header is one ordered Kafka record header.

type Inspector 

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

Inspector provides bounded, read-only protocol administration used by readiness checks, dashboards, and replay planning.

func NewInspector 

func NewInspector(config InspectorConfig) (*Inspector, error)

NewInspector constructs a read-only Kafka inspector.

func (*Inspector) Close 

func (inspector *Inspector) Close() error

Close idempotently closes the underlying Kafka client. Calls made from an inspector observer fail with ErrObserverReentry and leave the client open.

func (*Inspector) Cluster 

func (inspector *Inspector) Cluster(
	ctx context.Context,
) (result ClusterState, resultErr error)

Cluster returns bounded, sorted cluster identity and broker metadata. A controller ID is visible only when it identifies a returned broker.

func (*Inspector) ConsumerGroupLag 

func (inspector *Inspector) ConsumerGroupLag(
	ctx context.Context,
	groups ...string,
) (result []ConsumerGroupState, resultErr error)

ConsumerGroupLag requests sorted stable committed and end offsets for an explicit bounded consumer-group set. On KIP-447 brokers, pending transactional offset commits resolve within the configured request deadline before offsets are returned.

func (*Inspector) ConsumerProtocolGroupLag 

func (inspector *Inspector) ConsumerProtocolGroupLag(
	ctx context.Context,
	groups ...string,
) (result []ConsumerProtocolGroupState, resultErr error)

ConsumerProtocolGroupLag requests sorted, bounded KIP-848 consumer-group state and stable committed-offset lag for an explicit group set. It is separate from ConsumerGroupLag because Kafka's classic and consumer group protocols expose materially different assignment and fencing state.

func (*Inspector) DependencyHealth 

func (inspector *Inspector) DependencyHealth(
	ctx context.Context,
) (resultErr error)

DependencyHealth verifies current Kafka connectivity within the configured request deadline. It is diagnostic input, not liveness or readiness.

func (*Inspector) Health 

func (inspector *Inspector) Health(ctx context.Context) error

Health is the compatibility alias for DependencyHealth.

func (*Inspector) InspectConsumerGroups 

func (inspector *Inspector) InspectConsumerGroups(
	ctx context.Context,
	groups ...string,
) ([]ConsumerGroupInspectionResult, error)

InspectConsumerGroups returns one input-ordered result for every explicit consumer group. Target failures do not discard independent successes: the returned slice is complete and ErrInspectionTargetsFailed signals that callers must inspect each result. Requests share the inspector request deadline and never exceed MaxConcurrentInspections. ConsumerGroupLag remains the fail-closed batch API.

func (*Inspector) InspectConsumerProtocolGroups 

func (inspector *Inspector) InspectConsumerProtocolGroups(
	ctx context.Context,
	groups ...string,
) ([]ConsumerProtocolGroupInspectionResult, error)

InspectConsumerProtocolGroups returns one input-ordered result for every explicit KIP-848 consumer group. Target failures do not discard independent successes. The returned error is ErrInspectionTargetsFailed when any result failed. Requests share one deadline and never exceed MaxConcurrentInspections.

func (*Inspector) InspectTopics 

func (inspector *Inspector) InspectTopics(
	ctx context.Context,
	topics ...string,
) ([]TopicInspectionResult, error)

InspectTopics returns one input-ordered result for every explicit topic. Target failures do not discard independent successes: the returned slice is complete and ErrInspectionTargetsFailed signals that callers must inspect each result. Requests share the inspector request deadline and never exceed MaxConcurrentInspections. Topics remains the fail-closed batch API.

func (*Inspector) Liveness 

func (inspector *Inspector) Liveness() LivenessState

Liveness reports only whether this inspector remains locally open. Broker connectivity and readiness hysteresis do not affect it.

func (*Inspector) PlanReplayByTimestamp 

func (inspector *Inspector) PlanReplayByTimestamp(
	ctx context.Context,
	request ReplayTimestampRequest,
) (ReplayTimestampPlan, error)

PlanReplayByTimestamp resolves one explicit timestamp window to exact Kafka offsets without polling records, joining a group, changing group offsets, or invoking replay handlers. Any error returns a zero plan.

func (*Inspector) Readiness 

func (inspector *Inspector) Readiness(
	ctx context.Context,
) (state ReadinessState, resultErr error)

Readiness observes current dependency health and applies the configured consecutive-failure and recovery thresholds. A temporary dependency failure does not immediately make a previously ready inspector unready.

func (*Inspector) Topics 

func (inspector *Inspector) Topics(
	ctx context.Context,
	topics ...string,
) (result []TopicState, resultErr error)

Topics returns sorted metadata for an explicit bounded topic set.

type InspectorConfig 

type InspectorConfig struct {
	Brokers        []string
	ClientID       string
	Protocol       ProtocolPolicy
	Security       ClientSecurity
	DialTimeout    time.Duration
	RequestTimeout time.Duration

	MaxMetadataBrokers    int
	MaxMetadataPartitions int
	MaxGroupMembers       int
	// MaxConcurrentInspections bounds independently isolated per-target
	// requests made by InspectTopics, InspectConsumerGroups, and
	// InspectConsumerProtocolGroups.
	MaxConcurrentInspections int
	Readiness                ReadinessPolicy
	// Observers receive payload-free inspection, health, readiness, shutdown,
	// and broker events under one bounded synchronous callback policy.
	Observers ObserverPolicy
}

InspectorConfig defines a read-only Kafka metadata and lag client.

type KeyPolicy 

type KeyPolicy uint8

KeyPolicy controls whether a producer accepts records without keys. 

const (
	// KeyRequired is the safe default and preserves a stable partition-ordering
	// identity for every record.
	KeyRequired KeyPolicy = iota
	// UnkeyedAllowed explicitly permits records for which the configured
	// partitioner chooses a partition without a key.
	UnkeyedAllowed
)

type LivenessState 

type LivenessState struct {
	Live bool
}

LivenessState reports whether the inspector remains locally usable. Kafka connectivity does not affect this signal.

type Message 

type Message = ProducerRecord

Message is retained as the pre-v1 name for ProducerRecord.

type MessageLimits 

type MessageLimits struct {
	MaxTopicBytes       int
	MaxKeyBytes         int
	MaxValueBytes       int
	MaxHeaders          int
	MaxHeaderKeyBytes   int
	MaxHeaderValueBytes int
	MaxHeaderBytes      int
}

MessageLimits bounds caller-controlled record fields before they reach the client buffer.

func DefaultMessageLimits 

func DefaultMessageLimits() MessageLimits

DefaultMessageLimits returns conservative limits below Kafka's default one-megabyte broker record limit.

func (MessageLimits) Validate 

func (limits MessageLimits) Validate() error

Validate reports whether every message limit is positive.

type OAuthBearerProvider 

type OAuthBearerProvider interface {
	Token(context.Context) (OAuthBearerToken, error)
}

OAuthBearerProvider returns a fresh token for one authentication session. Implementations must be concurrency-safe and honor ctx.

type OAuthBearerProviderFunc 

type OAuthBearerProviderFunc func(context.Context) (OAuthBearerToken, error)

OAuthBearerProviderFunc adapts a function to OAuthBearerProvider.

func (OAuthBearerProviderFunc) Token 

func (provider OAuthBearerProviderFunc) Token(
	ctx context.Context,
) (OAuthBearerToken, error)

Token invokes provider.

type OAuthBearerToken 

type OAuthBearerToken struct {
	Token           []byte
	ExpiresAt       time.Time
	AuthorizationID string
	Extensions      map[string]string
}

OAuthBearerToken contains one owned token result. ExpiresAt is required so expired credentials fail before any authentication bytes are constructed. String formatting is always redacted.

func (OAuthBearerToken) GoString 

func (token OAuthBearerToken) GoString() string

GoString returns a stable redacted representation for %#v formatting.

func (OAuthBearerToken) String 

func (token OAuthBearerToken) String() string

String returns a stable redacted representation.

type Observation 

type Observation struct {
	// Kind identifies the completed package operation.
	Kind ObservationKind
	// StartedAt is the local operation start time.
	StartedAt time.Time
	// Duration is the elapsed local operation time through final delivery.
	Duration time.Duration
	// ClientID is the copied configured Kafka client identity.
	ClientID string
	// GroupID is the copied configured consumer-group identity when applicable.
	GroupID string
	// BrokerID is the Kafka node ID when BrokerKnown is true. Broker endpoints
	// are never copied into observations.
	BrokerID int32
	// BrokerKnown reports whether BrokerID is authoritative.
	BrokerKnown bool
	// AuthenticationMethod is the configured SASL method for a broker-connect
	// initialization. AuthenticationNone means no SASL flow was configured.
	AuthenticationMethod AuthenticationMethod
	// APIKey is the Kafka protocol request key when APIKeyKnown is true.
	APIKey int16
	// APIKeyKnown reports whether APIKey is authoritative.
	APIKeyKnown bool
	// RequestBytes is the request size written below TLS framing.
	RequestBytes int64
	// ResponseBytes is the response size read below TLS framing.
	ResponseBytes int64
	// QueueDuration is time the request waited inside franz-go before its
	// network write, including client-side throttle waiting.
	QueueDuration time.Duration
	// ThrottleDuration is the broker-imposed throttle interval.
	ThrottleDuration time.Duration
	// ThrottledAfterResponse reports that franz-go applies the throttle after
	// the broker response rather than the broker delaying its response.
	ThrottledAfterResponse bool
	// Topic is present only when validated metadata has one common topic.
	Topic string
	// Partition is the delivered partition when PartitionKnown is true.
	Partition int32
	// PartitionKnown reports whether Partition is authoritative.
	PartitionKnown bool
	// Offset is the delivered Kafka offset when OffsetKnown is true.
	Offset int64
	// OffsetKnown reports whether Offset is authoritative.
	OffsetKnown bool
	// Timestamp is Kafka's delivered record timestamp when available.
	Timestamp time.Time
	// RecordCount is the bounded operation input count.
	RecordCount int
	// PartitionCount is the bounded number of Kafka partitions represented.
	PartitionCount int
	// BrokerCount is the bounded number of Kafka brokers represented by a
	// cluster inspection. It is zero for other observations.
	BrokerCount int
	// TopicCount is the bounded number of requested Kafka topics represented
	// by a topic inspection. It is zero for other observations.
	TopicCount int
	// GroupCount is the bounded number of requested Kafka consumer groups
	// represented by a group inspection. It is zero for other observations.
	GroupCount int
	// GroupMemberCount is the bounded number of consumer-group members
	// represented by a group inspection. It is zero for other observations.
	GroupMemberCount int
	// ProcessedCount is the number of records whose handler completed.
	ProcessedCount int
	// CommittedCount is the number of source records durably settled.
	CommittedCount int
	// RecordBytes is a conservative payload and framing size, not a broker
	// encoded-byte measurement.
	RecordBytes int64
	// ReplayProcessed is the exact number of records processed by a replay
	// operation. It is zero for non-replay observations.
	ReplayProcessed int64
	// ReplaySkipped is the exact number of records skipped by a replay
	// operation. It is zero for non-replay observations.
	ReplaySkipped int64
	// ReplayFailed is the exact number of records failed by a replay operation.
	// It is zero for non-replay observations.
	ReplayFailed int64
	// ReplayRemaining is the exact number of requested offsets not yet
	// processed. It is zero for non-replay observations.
	ReplayRemaining int64
	// DependencyHealthy reports the result of a dependency probe or the
	// dependency state used by a readiness decision.
	DependencyHealthy bool
	// Ready reports the stateful readiness decision after a conclusive probe.
	Ready bool
	// ConsecutiveFailures is the bounded readiness failure count after a
	// conclusive probe.
	ConsecutiveFailures int
	// ConsecutiveSuccesses is the bounded readiness success count after a
	// conclusive probe.
	ConsecutiveSuccesses int
	// Succeeded reports whether the package operation returned success.
	Succeeded bool
	// Truncated reports that bounded diagnostic counts or metadata were clipped.
	Truncated bool
	// Category classifies failure and is ErrorUnknown after success.
	Category ErrorCategory
}

Observation is copied, payload-free metadata for one completed Kafka policy operation. Topic is populated only for a single validated topic. Category is ErrorUnknown when Succeeded is true.

func (Observation) Validate 

func (observation Observation) Validate() error

Validate reports whether the observation satisfies the public bounded metadata, settlement-count, and event-cardinality invariants.

type ObservationFailure 

type ObservationFailure struct {
	// ObserverIndex is the failed callback's index in ObserverPolicy.Observers.
	ObserverIndex int
	// Kind identifies the event being observed.
	Kind ObservationKind
	// TimedOut reports that the shared callback deadline expired before this
	// observer returned.
	TimedOut bool
	// Panicked reports that the observer panic was contained.
	Panicked bool
	// contains filtered or unexported fields
}

ObservationFailure reports which observer failed without formatting its potentially sensitive returned error. The application owns any error returned by Cause and must redact it before external reporting.

func (ObservationFailure) Cause 

func (failure ObservationFailure) Cause() error

Cause returns the observer error for explicit application handling.

func (ObservationFailure) Error 

func (failure ObservationFailure) Error() string

Error returns a stable message that does not render the observer error or panic value.

type ObservationFailureFunc 

type ObservationFailureFunc func(context.Context, ObservationFailure)

ObservationFailureFunc synchronously receives a contained observer failure. It must follow the same deadline and concurrency rules as ObserverFunc.

type ObservationKind 

type ObservationKind uint8

ObservationKind identifies one stable package-policy event. 

const (
	// ObservationProduceRecord reports completion of synchronous single-record
	// production.
	ObservationProduceRecord ObservationKind = iota + 1
	// ObservationProduceBatch reports completion of synchronous batch
	// production.
	ObservationProduceBatch
	// ObservationProduceAsync reports final asynchronous record delivery.
	ObservationProduceAsync
	// ObservationConsumeRecord reports one completed record-handler call.
	ObservationConsumeRecord
	// ObservationConsumeBatch reports one completed partition-batch handler call.
	ObservationConsumeBatch
	// ObservationConsumeCommit reports one completed source-offset commit attempt.
	ObservationConsumeCommit
	// ObservationConsumePoll reports one completed bounded consumer poll cycle.
	ObservationConsumePoll
	// ObservationBrokerConnect reports one completed broker connection
	// initialization, including protocol negotiation and configured SASL.
	ObservationBrokerConnect
	// ObservationBrokerRequest reports one completed Kafka protocol request.
	ObservationBrokerRequest
	// ObservationBrokerThrottle reports broker-imposed request throttling.
	ObservationBrokerThrottle
	// ObservationBrokerDisconnect reports a broker connection closing.
	ObservationBrokerDisconnect
	// ObservationConsumeAssigned reports a completed consumer-group partition
	// assignment callback.
	ObservationConsumeAssigned
	// ObservationConsumeRevoked reports a completed consumer-group partition
	// revocation callback.
	ObservationConsumeRevoked
	// ObservationConsumeLost reports fatal consumer-group partition ownership
	// loss.
	ObservationConsumeLost
	// ObservationConsumeBlocked reports that a rebalance callback is waiting
	// for the current bounded poll to release its rebalance gate.
	ObservationConsumeBlocked
	// ObservationConsumeGroupError reports an error that ended a consumer-group
	// management session.
	ObservationConsumeGroupError
	// ObservationTransactionBegin reports a completed Kafka transaction begin
	// attempt.
	ObservationTransactionBegin
	// ObservationTransactionCommit reports a completed Kafka transaction commit
	// attempt.
	ObservationTransactionCommit
	// ObservationTransactionAbort reports a completed Kafka transaction abort
	// attempt.
	ObservationTransactionAbort
	// ObservationReplayPlan reports broker validation of one bounded replay
	// plan without executing its handlers.
	ObservationReplayPlan
	// ObservationReplayRecord reports one replay record outcome, including
	// processed, skipped, and failed records.
	ObservationReplayRecord
	// ObservationReplayRun reports the exact aggregate outcome of one replay
	// execution.
	ObservationReplayRun
	// ObservationReplayShutdown reports one bounded replay-reader shutdown.
	ObservationReplayShutdown
	// ObservationInspectorCluster reports one bounded cluster metadata query.
	ObservationInspectorCluster
	// ObservationInspectorTopics reports one bounded topic metadata and
	// durability query.
	ObservationInspectorTopics
	// ObservationInspectorConsumerGroups reports one bounded consumer-group
	// lag query.
	ObservationInspectorConsumerGroups
	// ObservationDependencyHealth reports one bounded Kafka connectivity probe.
	ObservationDependencyHealth
	// ObservationReadiness reports one conclusive readiness-hysteresis update.
	ObservationReadiness
	// ObservationInspectorShutdown reports the inspector client closing.
	ObservationInspectorShutdown
	// ObservationProducerShutdown reports one bounded producer shutdown
	// attempt that acquired lifecycle ownership.
	ObservationProducerShutdown
	// ObservationConsumerShutdown reports one bounded consumer shutdown
	// attempt that acquired lifecycle ownership.
	ObservationConsumerShutdown
	// ObservationTransactionProcessorShutdown reports one bounded
	// consume-transform-produce processor shutdown attempt that acquired
	// lifecycle ownership.
	ObservationTransactionProcessorShutdown
	// ObservationConsumeRetryScheduled reports one failed handler attempt that
	// selected a bounded in-process retry before its backoff wait begins.
	ObservationConsumeRetryScheduled
	// ObservationConsumeRebalanceWait reports the bounded local wait from
	// franz-go blocked-callback entry until poll-gate release, callback
	// cancellation, or timeout.
	ObservationConsumeRebalanceWait
)

func (ObservationKind) String 

func (kind ObservationKind) String() string

String returns the stable low-cardinality observation name.

type ObserverFunc 

type ObserverFunc func(context.Context, Observation) error

ObserverFunc synchronously observes one copied event. Implementations must return when ctx is done, must be concurrency-safe, and must not retain the callback context for later work.

type ObserverPolicy 

type ObserverPolicy struct {
	// Observers run synchronously in slice order and are copied during
	// construction.
	Observers []ObserverFunc
	// FailureHandler receives every contained observer error, panic, or
	// cooperative timeout before the next observer runs.
	FailureHandler ObservationFailureFunc
	// Timeout is one shared cooperative budget for an event's observers and
	// failure callbacks.
	Timeout time.Duration
}

ObserverPolicy bounds ordered synchronous observation. Observers and their failure handler share one timeout budget per event and are copied during client construction.

func (ObserverPolicy) Validate 

func (policy ObserverPolicy) Validate() error

Validate reports whether the observer policy is internally compatible and bounded.

type OffsetPolicy 

type OffsetPolicy uint8

OffsetPolicy controls the first offset used when no committed group offset exists. 

const (
	OffsetEarliest OffsetPolicy = iota + 1
	OffsetLatest
)

type PartitionSelection 

type PartitionSelection struct {
	Mode      PartitionSelectionMode
	Partition int32
}

PartitionSelection is one immutable-by-value producer partition decision. Automatic selection requires Partition to remain zero. Explicit selections require a non-negative Partition and are validated before admission.

func ExplicitPartition 

func ExplicitPartition(partition int32) PartitionSelection

ExplicitPartition selects one exact non-negative Kafka partition. A negative value is retained so normal record validation can return a classifiable error without panicking during record construction.

type PartitionSelectionMode 

type PartitionSelectionMode uint8

PartitionSelectionMode identifies automatic or explicit producer partition selection. The zero value preserves Kafka key-based or unkeyed partitioning. 

const (
	// PartitionAutomatic delegates partition selection to the producer's
	// automatic keyed or unkeyed partitioner.
	PartitionAutomatic PartitionSelectionMode = iota
	// PartitionExplicit sends the record to one exact Kafka partition.
	PartitionExplicit
)

type PollResult 

type PollResult struct {
	Polled    int
	Processed int
	Committed int
}

PollResult summarizes one bounded fetch, processing, and commit cycle.

type Producer 

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

Producer publishes records with Kafka's idempotent producer and all in-sync replica acknowledgements.

func NewProducer 

func NewProducer(config ProducerConfig) (*Producer, error)

NewProducer constructs a producer without dialing brokers. Connectivity is established lazily by franz-go.

func (*Producer) Abort 

func (producer *Producer) Abort(ctx context.Context) error

Abort drops records still buffered by franz-go and waits for their delivery callbacks to run. It is an explicit data-loss operation intended for recovery after the caller has accepted the undelivered-record outcome.

func (*Producer) Close 

func (producer *Producer) Close() error

Close performs a configured bounded graceful shutdown. It returns ErrDrainIncomplete without closing the client when admitted records cannot resolve before ShutdownTimeout; callers may retry Shutdown or explicitly Abort.

func (*Producer) Diagnostic 

func (producer *Producer) Diagnostic() ProducerDiagnostic

Diagnostic returns a bounded, payload-free snapshot of local producer state without performing Kafka I/O, exposing the retained fatal error, or invoking methods on it.

func (*Producer) Drain 

func (producer *Producer) Drain(ctx context.Context) error

Drain waits for every admitted asynchronous record to resolve without closing the producer. New operations are rejected while the drain is in progress. A timeout or cancellation reports both ErrDrainIncomplete and the context failure; admitted records remain owned by the producer.

func (*Producer) Health 

func (producer *Producer) Health(ctx context.Context) error

Health verifies that a broker is reachable and responds to a metadata request.

func (*Producer) Publish 

func (producer *Producer) Publish(ctx context.Context, message Message) error

Publish waits for Kafka to accept the message or returns the first delivery error. A nil result does not provide end-to-end exactly-once delivery.

func (*Producer) PublishAsync 

func (producer *Producer) PublishAsync(
	ctx context.Context,
	record ProducerRecord,
) (<-chan DeliveryResult, error)

PublishAsync admits one owned record to the bounded franz-go producer and returns a one-result buffered channel. The caller may stop waiting without cancelling a record after this method returns; the package delivery deadline continues resolving that record and publishes the eventual result. Caller cancellation while admission is still blocked remains authoritative.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/faustbrian/go-kafka"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	producer, err := kafka.NewProducer(kafka.ProducerConfig{
		Brokers:       []string{"kafka.internal:9093"},
		ClientID:      "orders-api",
		AllowedTopics: []string{"orders.created.v1"},
	})
	if err != nil {
		log.Fatal(err)
	}

	delivery, err := producer.PublishAsync(ctx, kafka.ProducerRecord{
		Topic: "orders.created.v1",
		Key:   []byte("order-123"),
		Value: []byte(`{"order_id":"order-123"}`),
	})
	if err != nil {
		log.Fatal(errors.Join(err, producer.Close()))
	}

	var result kafka.DeliveryResult
	select {
	case result = <-delivery:
	case <-ctx.Done():
		// PublishAsync continues resolving an admitted record. Shutdown below
		// drains it rather than silently dropping it. A successful close means
		// the buffered result is now available for reconciliation.
		log.Print(ctx.Err())
		if err := producer.Close(); err != nil {
			log.Fatal(err)
		}
		result = <-delivery
		if result.Err != nil {
			log.Fatal(result.Err)
		}

		return
	}

	if err := errors.Join(result.Err, producer.Close()); err != nil {
		log.Fatal(err)
	}
}

func (*Producer) PublishBatch 

func (producer *Producer) PublishBatch(
	ctx context.Context,
	records []ProducerRecord,
) (results []DeliveryResult, resultErr error)

PublishBatch validates and owns an entire bounded batch before producing any record. Results remain in input order and expose every partial delivery failure.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/faustbrian/go-kafka"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	producer, err := kafka.NewProducer(kafka.ProducerConfig{
		Brokers:       []string{"kafka.internal:9093"},
		ClientID:      "orders-importer",
		AllowedTopics: []string{"orders.created.v1"},
	})
	if err != nil {
		log.Fatal(err)
	}

	results, publishErr := producer.PublishBatch(ctx, []kafka.ProducerRecord{
		{
			Topic: "orders.created.v1",
			Key:   []byte("order-123"),
			Value: []byte(`{"order_id":"order-123"}`),
		},
		{
			Topic: "orders.created.v1",
			Key:   []byte("order-124"),
			Value: []byte(`{"order_id":"order-124"}`),
		},
	})
	if publishErr != nil {
		// Results remain input-ordered and identify definite successes beside
		// failures. Reconcile them instead of retrying the complete batch.
		for index, result := range results {
			log.Printf("record %d: topic=%s partition=%d offset=%d error=%v",
				index,
				result.Topic,
				result.Partition,
				result.Offset,
				result.Err,
			)
		}
	}

	if err := errors.Join(publishErr, producer.Close()); err != nil {
		log.Fatal(err)
	}
}

func (*Producer) PublishRecord 

func (producer *Producer) PublishRecord(
	ctx context.Context,
	record ProducerRecord,
) DeliveryResult

PublishRecord synchronously publishes one record and returns its individual broker delivery metadata. The producer owns copies of all input bytes before passing the record to franz-go. Caller cancellation can stop an admitted non-transactional record and is reported as an ambiguous delivery.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/faustbrian/go-kafka"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	producer, err := kafka.NewProducer(kafka.ProducerConfig{
		Brokers:       []string{"kafka.internal:9093"},
		ClientID:      "orders-api",
		AllowedTopics: []string{"orders.created.v1"},
	})
	if err != nil {
		log.Fatal(err)
	}

	result := producer.PublishRecord(ctx, kafka.ProducerRecord{
		Topic: "orders.created.v1",
		Key:   []byte("order-123"),
		Value: []byte(`{"order_id":"order-123"}`),
	})
	if err := errors.Join(result.Err, producer.Close()); err != nil {
		// A timeout can be ambiguous. Reconcile it before deciding to retry.
		log.Fatal(err)
	}
	log.Printf("delivered to %s[%d] at offset %d",
		result.Topic,
		result.Partition,
		result.Offset,
	)
}

func (*Producer) RunTransaction 

func (producer *Producer) RunTransaction(
	ctx context.Context,
	callback func(Transaction) error,
) error

RunTransaction serializes one producer transaction. The transaction is committed only when callback returns nil; callback failure or panic triggers a bounded abort whose completion ignores caller cancellation. An ambiguous transactional publish deadline closes and permanently fences the producer, so RunTransaction returns without attempting commit or abort on that client.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/faustbrian/go-kafka"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	producer, err := kafka.NewProducer(kafka.ProducerConfig{
		Brokers:         []string{"kafka.internal:9093"},
		ClientID:        "billing-writer",
		AllowedTopics:   []string{"billing.balance.v1"},
		TransactionalID: "billing-writer-instance-7",
	})
	if err != nil {
		log.Fatal(err)
	}

	transactionErr := producer.RunTransaction(ctx, func(
		transaction kafka.Transaction,
	) error {
		return transaction.Publish(ctx, kafka.ProducerRecord{
			Topic: "billing.balance.v1",
			Key:   []byte("account-123"),
			Value: []byte(`{"balance":4200}`),
		})
	})
	if err := errors.Join(transactionErr, producer.Close()); err != nil {
		// An unknown commit outcome must be reconciled, not blindly retried.
		log.Fatal(err)
	}
}

func (*Producer) Shutdown 

func (producer *Producer) Shutdown(ctx context.Context) (resultErr error)

Shutdown fences new operations, drains every admitted record within ctx, and closes the underlying client. If draining is incomplete the producer remains fenced but open so the caller can retry Shutdown or explicitly Abort. Successful shutdown is idempotent.

type ProducerConfig 

type ProducerConfig struct {
	Brokers                []string
	ClientID               string
	Protocol               ProtocolPolicy
	AllowedTopics          []string
	KeyPolicy              KeyPolicy
	Limits                 MessageLimits
	MaxBufferedRecords     int
	MaxBufferedBytes       int
	MaxBatchRecords        int
	MaxBatchBytes          int32
	RecordRetries          int
	RetryBackoffMin        time.Duration
	RetryBackoffMax        time.Duration
	DeliveryTimeout        time.Duration
	ShutdownTimeout        time.Duration
	RequestTimeout         time.Duration
	DialTimeout            time.Duration
	Linger                 time.Duration
	CompressionPreferences []CompressionCodec
	TransactionalID        string
	TransactionTimeout     time.Duration

	TransactionEndTimeout time.Duration
	Security              ClientSecurity
	Observers             ObserverPolicy
}

ProducerConfig defines bounded Kafka producer identity, routing, delivery, lifecycle, and security policy. AllowedTopics is required and copied during construction.

func (ProducerConfig) Validate 

func (config ProducerConfig) Validate() error

Validate reports whether the producer configuration satisfies the bounded producer policy without constructing a client or dialing brokers.

type ProducerDiagnostic 

type ProducerDiagnostic struct {
	// Accepting reports whether a new non-transactional producer operation can
	// currently enter package admission.
	Accepting bool
	// TransactionsEnabled reports whether the producer owns a configured
	// transactional ID and accepts RunTransaction calls.
	TransactionsEnabled bool
	// TransactionActive reports package-local ownership of one RunTransaction
	// call; it is not coordinator state or proof of an open broker transaction.
	TransactionActive bool
	// MaintenanceActive reports that drain, abort, or shutdown currently owns
	// the producer lifecycle gate.
	MaintenanceActive bool
	// Closed reports that new operations are permanently fenced.
	Closed bool
	// ShutdownComplete reports that the underlying client close completed.
	ShutdownComplete bool
	// Fatal reports that the producer retained an unrecoverable lifecycle error.
	Fatal bool
	// FatalCategory is the redacted stable category of the retained fatal error,
	// or ErrorUnknown when Fatal is false.
	FatalCategory ErrorCategory
	// InFlightOperations is the number of admitted operations whose delivery or
	// health work has not completed.
	InFlightOperations int
	// AdmissionsInProgress is the number of operations still entering the
	// bounded franz-go client.
	AdmissionsInProgress int
	// BufferedRecords is franz-go's current buffered-produce record count.
	BufferedRecords int64
	// BufferedBytes is franz-go's current sum of buffered key, value, and header
	// bytes, excluding Kafka framing.
	BufferedBytes int64
}

ProducerDiagnostic is a payload-free local snapshot of producer lifecycle and buffered-record state. It does not probe Kafka, describe a transaction coordinator, or prove that any record was delivered. Lifecycle fields are captured together under the producer state lock; buffered counts are a separate concurrent sample from franz-go and may change immediately.

type ProducerRecord 

type ProducerRecord struct {
	Topic     string
	Partition PartitionSelection
	Key       []byte
	Value     []byte
	Headers   []Header
	Timestamp time.Time
}

ProducerRecord is one Kafka record submitted for production. The producer copies all byte slices before retaining or passing the record to franz-go.

func (ProducerRecord) Validate 

func (record ProducerRecord) Validate(limits MessageLimits) error

Validate reports whether the record can be owned under limits without allocating or transferring its byte slices. It validates limits before inspecting record fields.

type ProtocolPolicy 

type ProtocolPolicy struct {
	MinimumVersion string
}

ProtocolPolicy controls Kafka request-version negotiation without exposing franz-go version types. An empty MinimumVersion uses franz-go's negotiated request versions without a package-imposed downgrade floor.

func (ProtocolPolicy) Validate 

func (policy ProtocolPolicy) Validate() error

Validate reports whether the configured minimum is a Kafka release known to the pinned franz-go version table.

type ReadinessPolicy 

type ReadinessPolicy struct {
	FailureThreshold  int
	RecoveryThreshold int
}

ReadinessPolicy controls stateful Kafka dependency-probe hysteresis.

func (ReadinessPolicy) Validate 

func (policy ReadinessPolicy) Validate() error

Validate checks readiness policy using the documented zero-value defaults.

type ReadinessState 

type ReadinessState struct {
	Ready                bool
	DependencyHealthy    bool
	ConsecutiveFailures  int
	ConsecutiveSuccesses int
}

ReadinessState is the current stateful readiness decision and the latest dependency observation. Ready is the service-composition signal; the method error retains the latest dependency failure for diagnostics.

type RebalanceHandlerPolicy 

type RebalanceHandlerPolicy uint8

RebalanceHandlerPolicy controls active handlers when franz-go reports that a group rebalance callback is waiting for the current poll to finish. 

const (
	// RebalanceCancelHandler requests cancellation through every active handler
	// context. It is the safe zero-value policy because it releases rebalances
	// promptly.
	RebalanceCancelHandler RebalanceHandlerPolicy = iota
	// RebalanceDrainHandler lets active handlers finish within their configured
	// deadlines, then settles successful results before releasing the rebalance.
	RebalanceDrainHandler
)

type ReplayCheckpoint 

type ReplayCheckpoint struct {
	Positions []ReplayPosition
}

ReplayCheckpoint is an externally persisted set of next offsets. Positions may omit configured ranges, which then start at their inclusive start.

func (ReplayCheckpoint) Retain 

func (checkpoint ReplayCheckpoint) Retain() ReplayCheckpoint

Retain returns a checkpoint with an independently owned position slice.

type ReplayConfig 

type ReplayConfig struct {
	Brokers     []string
	ClientID    string
	Protocol    ProtocolPolicy
	Ranges      []ReplayRange
	Checkpoint  ReplayCheckpoint
	SideEffects ReplaySideEffectPolicy
	Limits      MessageLimits
	Security    ClientSecurity
	// Observers receive payload-free plan, record, run, shutdown, and broker
	// events. Replay partition workers and broker goroutines can invoke the
	// copied callbacks concurrently.
	Observers ObserverPolicy

	MaxPollRecords int
	// MaxConcurrentFetches bounds broker fetch requests.
	MaxConcurrentFetches int
	// MaxConcurrentHandlers bounds simultaneous callbacks across independent
	// partitions. Records within one partition always remain sequential.
	MaxConcurrentHandlers int
	// FetchMinBytes is the minimum encoded record bytes a broker tries to
	// collect before answering a fetch. Zero defaults to one byte, values must
	// not exceed FetchMaxBytes, and FetchMaxWait bounds the wait.
	FetchMinBytes          int32
	FetchMaxBytes          int32
	FetchMaxPartitionBytes int32
	// BrokerMaxReadBytes is the hard maximum encoded Kafka response accepted
	// from one broker connection. It must be at least FetchMaxBytes.
	BrokerMaxReadBytes int32
	// MaxDecompressedBatchBytes is the hard decoded-byte limit for one Kafka
	// record batch.
	MaxDecompressedBatchBytes int64
	// MaxBufferedDecompressedBytes bounds decoded compressed-batch memory held
	// across active and prefetched replay responses.
	MaxBufferedDecompressedBytes int64
	FetchMaxWait                 time.Duration
	PlanningTimeout              time.Duration
	ProgressTimeout              time.Duration
	HandlerTimeout               time.Duration
	ShutdownTimeout              time.Duration
	DialTimeout                  time.Duration
}

ReplayConfig defines a bounded direct-partition reader. Replay readers do not join consumer groups or mutate group offsets.

func (ReplayConfig) Validate 

func (config ReplayConfig) Validate() error

Validate reports whether the replay policy is explicit, compatible, and bounded without constructing a Kafka client or retaining caller-owned data.

type ReplayHandler 

type ReplayHandler interface {
	HandleReplay(context.Context, ReplayRecord) error
}

ReplayHandler processes one replay record. Implementations must be concurrency-safe when replay permits more than one concurrent handler.

type ReplayHandlerFunc 

type ReplayHandlerFunc func(context.Context, ReplayRecord) error

ReplayHandlerFunc adapts a function to ReplayHandler.

func (ReplayHandlerFunc) HandleReplay 

func (handler ReplayHandlerFunc) HandleReplay(
	ctx context.Context,
	record ReplayRecord,
) error

HandleReplay invokes handler.

type ReplayMetadata 

type ReplayMetadata struct {
	Range                ReplayRange
	EffectiveStartOffset int64
}

ReplayMetadata identifies the complete requested range and the effective inclusive start selected by the external checkpoint for one replay run.

type ReplayPlan 

type ReplayPlan struct {
	Ranges         []ReplayPlannedRange
	TotalRemaining int64
}

ReplayPlan is an owned dry-run plan. Plan produces it without broker validation; PlanAgainstBroker returns it only after validating broker bounds.

type ReplayPlannedRange 

type ReplayPlannedRange struct {
	ReplayRange
	NextOffset int64
	Remaining  int64
}

ReplayPlannedRange is one immutable dry-run range after applying the external checkpoint.

type ReplayPosition 

type ReplayPosition struct {
	Topic      string
	Partition  int32
	NextOffset int64
}

ReplayPosition is the next offset an external replay checkpoint requests for one configured topic partition.

type ReplayRange 

type ReplayRange struct {
	Topic       string
	Partition   int32
	StartOffset int64
	EndOffset   int64
}

ReplayRange is one inclusive start and exclusive end partition range.

type ReplayRangeResult 

type ReplayRangeResult struct {
	ReplayRange
	NextOffset int64
	Processed  int64
	Skipped    int64
	Failed     int64
	Complete   bool
}

ReplayRangeResult reports exact progress for one configured range.

type ReplayReader 

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

ReplayReader processes exact offset ranges without changing consumer-group state.

func NewReplayReader 

func NewReplayReader(config ReplayConfig) (*ReplayReader, error)

NewReplayReader constructs a direct-partition replay reader.

func (*ReplayReader) Close 

func (reader *ReplayReader) Close() error

Close performs bounded shutdown using the configured shutdown timeout.

func (*ReplayReader) Plan 

func (reader *ReplayReader) Plan() ReplayPlan

Plan returns an owned dry-run plan after applying the external checkpoint. It performs no broker request and cannot prove current retention bounds.

func (*ReplayReader) PlanAgainstBroker 

func (reader *ReplayReader) PlanAgainstBroker(
	ctx context.Context,
) (plan ReplayPlan, resultErr error)

PlanAgainstBroker returns an owned dry-run plan after confirming that every effective start remains retained and every exclusive end is at or before the current broker high watermark. It does not poll records, invoke handlers, mutate group offsets, or consume the reader's single execution. Any error returns a zero plan so an unvalidated local plan cannot be mistaken for a broker-validated result.

func (*ReplayReader) Replay 

func (reader *ReplayReader) Replay(
	ctx context.Context,
	handler ReplayHandler,
) (result ReplayResult, resultErr error)

Replay performs one execution of every requested retained offset in partition order. A reader is single-use even after failure. Missing offsets fail closed; the caller must explicitly approve side effects and persist the returned checkpoint outside this package before resuming.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/faustbrian/go-kafka"
)

func persistProjection(context.Context, kafka.ConsumedRecord) error {

	return nil
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	reader, err := kafka.NewReplayReader(kafka.ReplayConfig{
		Brokers:  []string{"kafka.internal:9093"},
		ClientID: "billing-replay-2026-08-10",
		Ranges: []kafka.ReplayRange{{
			Topic:       "orders.created.v1",
			Partition:   0,
			StartOffset: 100,
			EndOffset:   200,
		}},
		SideEffects: kafka.ReplaySideEffectsAllowed,
	})
	if err != nil {
		log.Fatal(err)
	}

	plan, err := reader.PlanAgainstBroker(ctx)
	if err != nil {
		log.Fatal(errors.Join(err, reader.Close()))
	}
	log.Printf("reviewed %d retained records", plan.TotalRemaining)

	result, replayErr := reader.Replay(ctx, kafka.ReplayHandlerFunc(func(
		ctx context.Context,
		record kafka.ReplayRecord,
	) error {
		// The application must make replay side effects idempotent.
		return persistProjection(ctx, record.ConsumedRecord)
	}))
	if err := errors.Join(replayErr, reader.Close()); err != nil {
		// Persist the checkpoint externally before constructing a new reader.
		checkpoint := result.Checkpoint()
		log.Printf("replay incomplete at %+v", checkpoint.Positions)
		log.Fatal(err)
	}
}

func (*ReplayReader) Shutdown 

func (reader *ReplayReader) Shutdown(
	ctx context.Context,
) (resultErr error)

Shutdown fences new replay work, waits for the active replay, and closes the direct Kafka client. An incomplete shutdown remains fenced and can be retried. Concurrent shutdown calls fail with ErrReplayShutdownActive.

type ReplayRecord 

type ReplayRecord struct {
	ConsumedRecord
	Metadata ReplayMetadata
}

ReplayRecord is one borrowed Kafka record plus immutable replay provenance. Record bytes remain valid only for the synchronous handler call unless Retain is used.

func (ReplayRecord) Retain 

func (record ReplayRecord) Retain() ReplayRecord

Retain returns a replay record with independently owned Kafka record bytes.

type ReplayResult 

type ReplayResult struct {
	Polled           int64
	Processed        int64
	Skipped          int64
	Failed           int64
	CompletedRanges  int
	IncompleteRanges int
	Ranges           []ReplayRangeResult
}

ReplayResult summarizes completed and resumable replay progress.

func (ReplayResult) Checkpoint 

func (result ReplayResult) Checkpoint() ReplayCheckpoint

Checkpoint returns an independently owned checkpoint for resuming every configured range after the last successfully processed offset.

Example 
package main

import (
	"fmt"

	"github.com/faustbrian/go-kafka"
)

func main() {
	result := kafka.ReplayResult{
		Ranges: []kafka.ReplayRangeResult{{
			ReplayRange: kafka.ReplayRange{
				Topic:       "events",
				Partition:   2,
				StartOffset: 100,
				EndOffset:   200,
			},
			NextOffset: 137,
			Processed:  37,
		}},
	}

	checkpoint := result.Checkpoint()
	position := checkpoint.Positions[0]
	fmt.Printf("%s[%d] resumes at %d\n",
		position.Topic,
		position.Partition,
		position.NextOffset,
	)

}

Output:
events[2] resumes at 137

type ReplaySideEffectPolicy 

type ReplaySideEffectPolicy uint8

ReplaySideEffectPolicy controls whether Replay may invoke an application handler. The zero value fails closed so dry-run planning cannot accidentally execute side effects. 

const (
	// ReplaySideEffectsDenied permits planning but rejects Replay.
	ReplaySideEffectsDenied ReplaySideEffectPolicy = iota
	// ReplaySideEffectsAllowed explicitly permits handler invocation.
	ReplaySideEffectsAllowed
)

type ReplayTimestampPartition 

type ReplayTimestampPartition struct {
	Topic       string
	Partition   int32
	StartOffset int64
	EndOffset   int64
	Remaining   int64
}

ReplayTimestampPartition is one resolved partition range. A zero Remaining count is a valid empty time window and is omitted by ReplayRanges.

type ReplayTimestampPlan 

type ReplayTimestampPlan struct {
	StartInclusive time.Time
	EndExclusive   time.Time
	Partitions     []ReplayTimestampPartition
	TotalRemaining int64
}

ReplayTimestampPlan is an owned broker-resolved timestamp plan. Boundaries are canonical UTC millisecond values and partitions are sorted.

func (ReplayTimestampPlan) ReplayRanges 

func (plan ReplayTimestampPlan) ReplayRanges() []ReplayRange

ReplayRanges returns independently owned non-empty exact offset ranges that can be supplied to ReplayConfig. An empty result means the timestamp window currently contains no records in the selected partitions.

type ReplayTimestampRequest 

type ReplayTimestampRequest struct {
	StartInclusive time.Time
	EndExclusive   time.Time
	Partitions     []TopicPartition
}

ReplayTimestampRequest selects explicit partitions within one inclusive start and exclusive end timestamp window. Kafka resolves timestamps at millisecond precision, so both boundaries must be exact milliseconds at or after the Unix epoch.

func (ReplayTimestampRequest) Validate 

func (request ReplayTimestampRequest) Validate() error

Validate reports whether the timestamp window and explicit partition set can be resolved through bounded exact-partition Kafka requests.

type TimestampType 

type TimestampType int8

TimestampType identifies how Kafka assigned a record timestamp. 

const (
	// TimestampUnknown identifies records from message formats without a
	// timestamp.
	TimestampUnknown TimestampType = -1
	// TimestampCreateTime identifies a timestamp assigned by the producer.
	TimestampCreateTime TimestampType = 0
	// TimestampLogAppendTime identifies a timestamp assigned by the broker.
	TimestampLogAppendTime TimestampType = 1
)

type TopicCleanupPolicy 

type TopicCleanupPolicy uint8

TopicCleanupPolicy is the effective Kafka log cleanup policy. Zero means that no cleanup policy is active. 

const (
	// TopicCleanupDelete removes old segments under the effective retention
	// time or per-partition byte limit.
	TopicCleanupDelete TopicCleanupPolicy = 1
	// TopicCleanupCompact retains the latest record for each key, subject to
	// Kafka's compaction and tombstone-retention policy.
	TopicCleanupCompact TopicCleanupPolicy = 2
)

type TopicInspectionResult 

type TopicInspectionResult struct {
	Topic    string
	State    TopicState
	Category ErrorCategory
	Err      error
}

TopicInspectionResult is one input-ordered topic inspection outcome. State is populated only when Err is nil. Err retains the target-specific broker or policy failure, and Category provides its stable package classification.

type TopicPartition 

type TopicPartition struct {
	Topic     string
	Partition int32
}

TopicPartition identifies one exact Kafka partition. It does not include a consumer generation or imply current assignment ownership.

type TopicPartitionState 

type TopicPartitionState struct {
	Partition         int32
	Leader            int32
	LeaderEpoch       int32
	Replicas          []int32
	InSyncReplicaIDs  []int32
	OfflineReplicaIDs []int32
	ReplicationFactor int
	InSyncReplicas    int
	OfflineReplicas   int
	BeginningOffset   int64
	EndOffset         int64
}

TopicPartitionState is the current broker metadata for one partition.

type TopicState 

type TopicState struct {
	// Name is the requested Kafka topic.
	Name string
	// Internal reports whether the broker marks the topic as internal.
	Internal bool
	// MinInSyncReplicas is the effective min.insync.replicas value.
	MinInSyncReplicas int
	// CleanupPolicy is the effective cleanup.policy value.
	CleanupPolicy TopicCleanupPolicy
	// RetentionMilliseconds is the effective retention.ms value. Minus one
	// means unlimited time retention.
	RetentionMilliseconds int64
	// RetentionBytesPerPartition is the effective retention.bytes value.
	// Minus one means no size limit.
	RetentionBytesPerPartition int64
	// LocalRetentionMilliseconds is the effective local.retention.ms value.
	// Minus two inherits RetentionMilliseconds and minus one means unlimited.
	// Kafka applies this only while remote storage is enabled and remote copying
	// is not disabled.
	LocalRetentionMilliseconds int64
	// LocalRetentionBytesPerPartition is the effective local.retention.bytes
	// value. Minus two inherits RetentionBytesPerPartition and minus one means
	// no local size limit.
	LocalRetentionBytesPerPartition int64
	// LocalRetentionVisible reports whether Kafka returned both local-retention
	// fields. Older supported brokers may omit them.
	LocalRetentionVisible bool
	// RemoteStorageEnabled is the effective remote.storage.enable value.
	RemoteStorageEnabled bool
	// RemoteStorageEnabledVisible reports whether Kafka returned
	// remote.storage.enable.
	RemoteStorageEnabledVisible bool
	// RemoteLogCopyDisabled is the effective remote.log.copy.disable value.
	// When true, Kafka ignores the local-retention limits.
	RemoteLogCopyDisabled bool
	// RemoteLogCopyDisabledVisible reports whether Kafka returned
	// remote.log.copy.disable.
	RemoteLogCopyDisabledVisible bool
	// DeleteRetentionMilliseconds is the effective delete.retention.ms value.
	DeleteRetentionMilliseconds int64
	// MinimumCompactionLagMilliseconds is the effective
	// min.compaction.lag.ms value.
	MinimumCompactionLagMilliseconds int64
	// MaximumCompactionLagMilliseconds is the effective
	// max.compaction.lag.ms value.
	MaximumCompactionLagMilliseconds int64
	// MinimumCleanableDirtyRatio is the effective
	// min.cleanable.dirty.ratio value in the inclusive range zero to one.
	MinimumCleanableDirtyRatio float64
	// SegmentBytes is the effective segment.bytes value.
	SegmentBytes int64
	// SegmentMilliseconds is the effective segment.ms value.
	SegmentMilliseconds int64
	// UncleanLeaderElectionEnabled is the effective
	// unclean.leader.election.enable value.
	UncleanLeaderElectionEnabled bool
	// Partitions contains copied state sorted by partition number.
	Partitions []TopicPartitionState
}

TopicState is the current metadata for one requested topic.

type Transaction 

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

Transaction is the producer surface available inside RunTransaction.

func (Transaction) Publish 

func (transaction Transaction) Publish(ctx context.Context, message Message) error

Publish synchronously publishes one message inside the active transaction. If its context or the configured delivery bound expires after admission, the outcome is ambiguous and the owning producer enters ErrProducerFatal; callers must close and replace it rather than retrying on the same client.

type TransactionConnectionConfig 

type TransactionConnectionConfig struct {
	Brokers     []string
	ClientID    string
	Protocol    ProtocolPolicy
	DialTimeout time.Duration
	Security    ClientSecurity
}

TransactionConnectionConfig defines shared broker, identity, protocol, and security policy for one consume-transform-produce client.

type TransactionError 

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

TransactionError classifies a Kafka transaction lifecycle failure without rendering its potentially sensitive cause. Abortable reports that Kafka definitively rejected the commit and the producer may continue only after a successful abort. OutcomeKnown reports whether the attempted transaction is known not to have committed; false requires reconciliation before reuse.

func (*TransactionError) Abortable 

func (err *TransactionError) Abortable() bool

Abortable reports whether Kafka requires a bounded abort before the transactional ID can be reused.

func (*TransactionError) Category 

func (err *TransactionError) Category() ErrorCategory

Category returns the stable operational category.

func (*TransactionError) Error 

func (err *TransactionError) Error() string

Error implements error with a stable redacted diagnostic.

func (*TransactionError) Operation 

func (err *TransactionError) Operation() TransactionOperation

Operation returns the transaction phase that failed.

func (*TransactionError) OutcomeKnown 

func (err *TransactionError) OutcomeKnown() bool

OutcomeKnown reports whether the transaction is known not to have committed.

func (*TransactionError) Unwrap 

func (err *TransactionError) Unwrap() error

Unwrap returns the original failure for errors.Is and errors.As.

type TransactionGroupConfig 

type TransactionGroupConfig struct {
	GroupID              string
	InstanceID           string
	Rack                 string
	Topics               []string
	ResetOffset          OffsetPolicy
	BalancePolicy        GroupBalancePolicy
	MaxPollRecords       int
	MaxConcurrentFetches int
	// FetchMinBytes is the minimum encoded record bytes a broker tries to
	// collect before answering a fetch. Zero defaults to one byte, values must
	// not exceed FetchMaxBytes, and FetchMaxWait bounds the wait.
	FetchMinBytes          int32
	FetchMaxBytes          int32
	FetchMaxPartitionBytes int32
	// BrokerMaxReadBytes is the hard maximum encoded Kafka response accepted
	// from one broker connection. It must be at least FetchMaxBytes.
	BrokerMaxReadBytes int32
	// MaxDecompressedBatchBytes is the hard decoded-byte limit for one Kafka
	// source record batch.
	MaxDecompressedBatchBytes int64
	// MaxBufferedDecompressedBytes bounds decoded compressed-batch memory held
	// across active and prefetched source responses.
	MaxBufferedDecompressedBytes int64
	FetchMaxWait                 time.Duration
	SessionTimeout               time.Duration
	RebalanceTimeout             time.Duration
	HeartbeatInterval            time.Duration
	ProcessingTimeout            time.Duration
}

TransactionGroupConfig defines the bounded read-committed consumer-group side of one consume-transform-produce client.

type TransactionHandler 

type TransactionHandler interface {
	Handle(context.Context, ConsumedRecord, Transaction) error
}

TransactionHandler processes one borrowed source record and may publish records through the transaction capability. Returning an error aborts the complete poll and leaves every source offset unsettled.

type TransactionHandlerFunc 

type TransactionHandlerFunc func(context.Context, ConsumedRecord, Transaction) error

TransactionHandlerFunc adapts a function to TransactionHandler.

func (TransactionHandlerFunc) Handle 

func (handler TransactionHandlerFunc) Handle(
	ctx context.Context,
	record ConsumedRecord,
	transaction Transaction,
) error

Handle invokes handler.

type TransactionOperation 

type TransactionOperation uint8

TransactionOperation identifies the Kafka transaction phase that failed. 

const (
	// TransactionOperationBegin identifies failure to begin a transaction.
	TransactionOperationBegin TransactionOperation = iota + 1
	// TransactionOperationCommit identifies failure while ending a transaction
	// with a commit attempt.
	TransactionOperationCommit
	// TransactionOperationAbort identifies failure while discarding buffered
	// records or ending a transaction with an abort attempt.
	TransactionOperationAbort
)

func (TransactionOperation) String 

func (operation TransactionOperation) String() string

String returns a stable low-cardinality transaction operation name.

type TransactionOutputConfig 

type TransactionOutputConfig struct {
	AllowedTopics          []string
	KeyPolicy              KeyPolicy
	MaxBufferedRecords     int
	MaxBufferedBytes       int
	MaxBatchBytes          int32
	MaxOutputRecords       int
	MaxOutputBytes         int64
	RecordRetries          int
	RetryBackoffMin        time.Duration
	RetryBackoffMax        time.Duration
	DeliveryTimeout        time.Duration
	RequestTimeout         time.Duration
	Linger                 time.Duration
	CompressionPreferences []CompressionCodec
	TransactionalID        string
	TransactionTimeout     time.Duration
	TransactionEndTimeout  time.Duration
}

TransactionOutputConfig defines bounded transactional production policy. TransactionalID must be unique to one live processor instance.

type TransactionPollResult 

type TransactionPollResult struct {
	Polled    int
	Processed int
	Published int
	Committed bool
}

TransactionPollResult summarizes one all-or-nothing source poll. Published counts broker-acknowledged output records inside the transaction; those records are visible to read-committed consumers only when Committed is true.

type TransactionProcessor 

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

TransactionProcessor owns one read-committed group member and transactional producer. One Run or RunOnce call may be active at a time. An ambiguous transactional output deadline closes and permanently fences the processor.

func NewTransactionProcessor 

func NewTransactionProcessor(
	config TransactionProcessorConfig,
) (*TransactionProcessor, error)

NewTransactionProcessor constructs a Kafka-only consume-transform-produce runner. The franz-go client establishes broker connections lazily.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"os"
	"os/signal"
	"syscall"

	"github.com/faustbrian/go-kafka"
)

func main() {
	ctx, stop := signal.NotifyContext(
		context.Background(),
		os.Interrupt,
		syscall.SIGTERM,
	)
	defer stop()

	processor, err := kafka.NewTransactionProcessor(
		kafka.TransactionProcessorConfig{
			Connection: kafka.TransactionConnectionConfig{
				Brokers:  []string{"kafka.internal:9093"},
				ClientID: "billing-projection",
			},
			Group: kafka.TransactionGroupConfig{
				GroupID:     "billing-projection-v1",
				Topics:      []string{"orders.created.v1"},
				ResetOffset: kafka.OffsetEarliest,
			},
			Output: kafka.TransactionOutputConfig{
				AllowedTopics:   []string{"billing.balance.v1"},
				TransactionalID: "billing-projection-instance-7",
			},
		},
	)
	if err != nil {
		log.Fatal(err)
	}

	runErr := processor.Run(ctx, kafka.TransactionHandlerFunc(func(
		ctx context.Context,
		source kafka.ConsumedRecord,
		transaction kafka.Transaction,
	) error {
		return transaction.Publish(ctx, kafka.ProducerRecord{
			Topic: "billing.balance.v1",
			Key:   source.Key,
			Value: source.Value,
		})
	}))
	if err := errors.Join(runErr, processor.Close()); err != nil {
		log.Fatal(err)
	}
}

func (*TransactionProcessor) Close 

func (processor *TransactionProcessor) Close() error

Close performs a bounded graceful shutdown using the configured timeout.

func (*TransactionProcessor) Diagnostic 

func (processor *TransactionProcessor) Diagnostic() TransactionProcessorDiagnostic

Diagnostic returns a bounded, payload-free snapshot of local processor state without performing Kafka I/O, exposing the retained fatal error, or invoking methods on it.

func (*TransactionProcessor) Run 

func (processor *TransactionProcessor) Run(
	ctx context.Context,
	handler TransactionHandler,
) error

Run executes bounded all-or-nothing polls until cancellation or the first source-poll, processing, or transaction failure. Caller cancellation is a clean stop only after the active transaction aborts successfully; cleanup failure is returned and fences the processor.

func (*TransactionProcessor) RunOnce 

func (processor *TransactionProcessor) RunOnce(
	ctx context.Context,
	handler TransactionHandler,
) (TransactionPollResult, error)

RunOnce polls at most the configured source-record limit. Every fetched record must complete successfully before output records and all source offsets are committed in one Kafka transaction. Source poll and group-join failures return a ConsumerError with ConsumerOperationPoll.

func (*TransactionProcessor) Shutdown 

func (processor *TransactionProcessor) Shutdown(ctx context.Context) (err error)

Shutdown fences new runs, waits for the active runner, leaves a dynamic group, and closes the client. An incomplete shutdown remains retryable.

type TransactionProcessorConfig 

type TransactionProcessorConfig struct {
	Connection      TransactionConnectionConfig
	Group           TransactionGroupConfig
	Output          TransactionOutputConfig
	Limits          MessageLimits
	Observers       ObserverPolicy
	ShutdownTimeout time.Duration
}

TransactionProcessorConfig composes shared connection, source group, output, record-limit, and lifecycle policy for Kafka-only consume-transform-produce.

func (TransactionProcessorConfig) Validate 

func (config TransactionProcessorConfig) Validate() error

Validate reports whether the complete processor policy is bounded and internally consistent without constructing a client or dialing brokers.

type TransactionProcessorDiagnostic 

type TransactionProcessorDiagnostic struct {
	// Accepting reports whether a new Run or RunOnce call can currently start.
	Accepting bool
	// Running reports package-local ownership of one Run or RunOnce call.
	Running bool
	// TransactionActive reports a locally begun transaction attempt that has
	// not reached the package's commit, abort, or client-termination boundary.
	TransactionActive bool
	// Closing reports that shutdown has fenced new runs but has not completed.
	Closing bool
	// ShutdownActive reports that one Shutdown call currently owns lifecycle
	// completion.
	ShutdownActive bool
	// Closed reports that graceful shutdown completed.
	Closed bool
	// Fatal reports that the processor retained an unrecoverable lifecycle error.
	Fatal bool
	// ClientTerminated reports that the underlying client was forcefully closed
	// to bound an unsafe or ambiguous transactional output.
	ClientTerminated bool
	// FatalCategory is the redacted stable category of the retained fatal error,
	// or ErrorUnknown when Fatal is false.
	FatalCategory ErrorCategory
	// BufferedRecords is franz-go's current transactional-output record count.
	BufferedRecords int64
	// BufferedBytes is franz-go's current sum of buffered output key, value, and
	// header bytes, excluding Kafka framing.
	BufferedBytes int64
}

TransactionProcessorDiagnostic is a payload-free local snapshot of one consume-transform-produce processor. It does not probe Kafka, describe the transaction coordinator, or prove a transaction outcome. Lifecycle fields are captured together under the processor lifecycle lock; buffered counts are a separate concurrent sample from franz-go and may change immediately.

type TransportSecurity 

type TransportSecurity uint8

TransportSecurity selects verified TLS or an explicitly development-only plaintext connection. The zero value is verified TLS. 

const (
	// TransportTLS requires verified TLS with TLS 1.2 or newer.
	TransportTLS TransportSecurity = iota
	// TransportDevelopmentPlaintext permits an unencrypted connection for
	// isolated development fixtures. Authentication is forbidden in this mode.
	TransportDevelopmentPlaintext
)

func (TransportSecurity) String 

func (transport TransportSecurity) String() string

String returns the stable transport policy name.

type TrustAnchorProvider 

type TrustAnchorProvider interface {
	TrustAnchors(context.Context) (TrustAnchors, error)
}

TrustAnchorProvider returns the complete root set for one new TLS connection. Implementations must be concurrency-safe and honor ctx.

type TrustAnchorProviderFunc 

type TrustAnchorProviderFunc func(context.Context) (TrustAnchors, error)

TrustAnchorProviderFunc adapts a function to TrustAnchorProvider.

func (TrustAnchorProviderFunc) TrustAnchors 

func (provider TrustAnchorProviderFunc) TrustAnchors(
	ctx context.Context,
) (TrustAnchors, error)

TrustAnchors invokes provider.

type TrustAnchors 

type TrustAnchors struct {
	Certificates [][]byte
}

TrustAnchors contains one complete result from a TrustAnchorProvider. Each certificate must be one DER-encoded X.509 trust anchor. Ownership of the returned slice structure and certificate bytes transfers to the package; a provider must not mutate or reuse them after returning. The package retains only owned copies. String formatting is always redacted.

func (TrustAnchors) GoString 

func (anchors TrustAnchors) GoString() string

GoString returns a stable redacted representation for %#v formatting.

func (TrustAnchors) String 

func (anchors TrustAnchors) String() string

String returns a stable redacted representation.

type UsernamePassword 

type UsernamePassword struct {
	Username        string
	Password        []byte
	AuthorizationID string
}

UsernamePassword contains one owned credential result. String formatting is always redacted; callers remain responsible for not logging the fields.

func (UsernamePassword) GoString 

func (credentials UsernamePassword) GoString() string

GoString returns a stable redacted representation for %#v formatting.

func (UsernamePassword) String 

func (credentials UsernamePassword) String() string

String returns a stable redacted representation.

type UsernamePasswordProvider 

type UsernamePasswordProvider interface {
	Credentials(context.Context) (UsernamePassword, error)
}

UsernamePasswordProvider returns fresh credentials for one authentication session. Implementations must be concurrency-safe and honor ctx.

type UsernamePasswordProviderFunc 

type UsernamePasswordProviderFunc func(context.Context) (UsernamePassword, error)

UsernamePasswordProviderFunc adapts a function to UsernamePasswordProvider.

func (UsernamePasswordProviderFunc) Credentials 

func (provider UsernamePasswordProviderFunc) Credentials(
	ctx context.Context,
) (UsernamePassword, error)

Credentials invokes provider.

Source Files

View all Source files

Directories

Path Synopsis
adapters
golog
Package golog preserves the original Kafka slog adapter import path.
 Package golog preserves the original Kafka slog adapter import path.
slog
Package kafkaslog translates bounded Kafka observations into fixed structured log/slog records without adding payloads, credentials, endpoints, raw headers, or application error text.
 Package kafkaslog translates bounded Kafka observations into fixed structured log/slog records without adding payloads, credentials, endpoints, raw headers, or application error text.
gotelemetry module
mskiam module
otel module
service module
kafkaservice module
Package kafkatest provides public conformance suites for Kafka policy implementations and compatible broker fixtures.
 Package kafkatest provides public conformance suites for Kafka policy implementations and compatible broker fixtures.

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.