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 your handler returns an error, it is retried twice (three attempts in total). If it still fails, or if it panics, the affected worker stops and Consume shuts the whole consumer down, returning that error. A panic is recovered and reported like any other handler failure, so it never crashes the process or deadlocks the remaining partition workers.
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 (because its handler exhausted its retries or panicked), in which case it shuts the workers down and returns an error.
Source Files