consumer

Documentation

Index

• type CommitMode
• type ConsumerBuilder
  • func NewConsumer() *ConsumerBuilder
  • func (b *ConsumerBuilder) Build() (*KafkaConsumer, error)
  • func (b *ConsumerBuilder) WithBaseOpts(opts []kgo.Opt) *ConsumerBuilder
  • func (b *ConsumerBuilder) WithCommitMode(m CommitMode) *ConsumerBuilder
  • func (b *ConsumerBuilder) WithGroupID(id string) *ConsumerBuilder
  • func (b *ConsumerBuilder) WithLogger(l *zap.Logger) *ConsumerBuilder
  • func (b *ConsumerBuilder) WithStartOffset(o kgo.Offset) *ConsumerBuilder
  • func (b *ConsumerBuilder) WithSubscription(sub TopicSubscription) *ConsumerBuilder
  • func (b *ConsumerBuilder) WithTarget(t bus.Subscriber) *ConsumerBuilder
• type KafkaConsumer
  • func (c *KafkaConsumer) Start(ctx context.Context) error
  • func (c *KafkaConsumer) Stop() error
• type TopicSubscription
• type VinculumTopicFunc

Types

type CommitMode

type CommitMode int

CommitMode controls when Kafka offsets are committed.

const (
	// CommitAfterProcess commits the offset after target.OnEvent returns
	// without error. This is the default. Provides at-least-once delivery.
	CommitAfterProcess CommitMode = iota

	// CommitPeriodic delegates offset management to franz-go's auto-commit.
	CommitPeriodic

	// CommitManual reserves the mode for future caller-controlled commits.
	CommitManual
)

type ConsumerBuilder

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

ConsumerBuilder constructs a KafkaConsumer with a fluent API.

func NewConsumer

func NewConsumer() *ConsumerBuilder

NewConsumer returns a ConsumerBuilder with default settings: commit_mode=after_process, start_offset=stored.

func (*ConsumerBuilder) Build

func (b *ConsumerBuilder) Build() (*KafkaConsumer, error)

Build validates configuration, creates the kgo.Client, and returns a KafkaConsumer ready to be started.

func (*ConsumerBuilder) WithBaseOpts

func (b *ConsumerBuilder) WithBaseOpts(opts []kgo.Opt) *ConsumerBuilder

WithBaseOpts sets the connection-level kgo options (brokers, TLS, SASL, timeouts) shared with the producer client.

func (*ConsumerBuilder) WithCommitMode

func (b *ConsumerBuilder) WithCommitMode(m CommitMode) *ConsumerBuilder

WithCommitMode sets the offset commit strategy.

func (*ConsumerBuilder) WithGroupID

func (b *ConsumerBuilder) WithGroupID(id string) *ConsumerBuilder

WithGroupID sets the Kafka consumer group ID (required).

func (*ConsumerBuilder) WithLogger

func (b *ConsumerBuilder) WithLogger(l *zap.Logger) *ConsumerBuilder

WithLogger sets the logger used for poll and processing errors.

func (*ConsumerBuilder) WithStartOffset

func (b *ConsumerBuilder) WithStartOffset(o kgo.Offset) *ConsumerBuilder

WithStartOffset sets the offset to use when no committed offset exists for a partition. Use kgo.NewOffset().AtStart() for earliest, kgo.NewOffset().AtEnd() for latest, or kgo.NewOffset() for stored (default).

func (*ConsumerBuilder) WithSubscription

func (b *ConsumerBuilder) WithSubscription(sub TopicSubscription) *ConsumerBuilder

WithSubscription appends a topic subscription.

func (*ConsumerBuilder) WithTarget

func (b *ConsumerBuilder) WithTarget(t bus.Subscriber) *ConsumerBuilder

WithTarget sets the subscriber that receives deserialized messages (required).

type KafkaConsumer

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

KafkaConsumer runs a poll loop that reads records from Kafka and publishes them to a target bus.Subscriber. Create via NewConsumer().Build().

KafkaConsumer is a source, not a sink — it does NOT implement bus.Subscriber.

func (*KafkaConsumer) Start

func (c *KafkaConsumer) Start(ctx context.Context) error

Start launches the poll goroutine and returns immediately. The goroutine runs until ctx is cancelled or Stop is called.

func (*KafkaConsumer) Stop

func (c *KafkaConsumer) Stop() error

Stop signals the poll goroutine to exit, waits for it to finish, then closes the underlying kgo.Client.

type TopicSubscription

type TopicSubscription struct {
	// KafkaTopic is the exact Kafka topic name (no wildcards).
	KafkaTopic        string
	VinculumTopicFunc VinculumTopicFunc
}

TopicSubscription maps one Kafka topic to a vinculum topic resolver.

type VinculumTopicFunc

type VinculumTopicFunc func(kafkaTopic, key string, fields map[string]string, msg any) (string, error)

VinculumTopicFunc resolves the vinculum topic for an inbound Kafka record. It is called per message. kafkaTopic is the source Kafka topic; key is the record key decoded as UTF-8 (empty string if no key); fields are populated from record headers; msg is the deserialized payload.

Constructed by the config layer to avoid a circular import dependency.