Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var DefaultPollTimeout = 100 * time.Millisecond
DefaultPollTimeout is default time that the kafka consumer polls for a new message before it checks the context for cancellation.
Functions ¶
func RunConsumer ¶ added in v2.7.10
func RunConsumer(ctx context.Context, partitionConsumer *PartitionConsumer, handler HandleMessage)
RunConsumer runs partitionConsumer in a supervision loop that keeps it alive across failures. If Consume returns an error or panics, the panic is recovered, its stack is printed, and the consumer is restarted after a short delay. The loop only stops when ctx is canceled or the underlying kafka consumer is closed.
Types ¶
type HandleMessage ¶
HandleMessage is the user-provided function called for each consumed message. It receives the raw *kafka.Message so the caller has access to key, headers, and partition/offset metadata.
type PartitionConsumer ¶
PartitionConsumer subscribes to a topic and runs a dedicated goroutine per assigned partition. Offset storage is explicit: only messages that were successfully handled are stored for auto-commit.
If you return an error from your handler, you get two retries before the consumer will quit and shutdown with an error.
func (*PartitionConsumer) Consume ¶
func (c *PartitionConsumer) Consume(ctx context.Context, handle HandleMessage) error
Consume subscribes to the configured topics and dispatches every consumed message to handle. Each assigned partition is processed by its own worker goroutine, so messages within a partition are handled in order while different partitions run concurrently.
Offsets are stored only after a message was handled successfully and are flushed to the broker roughly every five seconds (and once more on shutdown). Consume blocks until ctx is canceled or a worker fails, in which case it shuts the workers down and returns an error.
Source Files