fix: add SeekTimeout consumer option

Author: deltics

consumer.go

@@ -29,6 +29,7 @@ type consumer struct {
 	groupId     string
 	handlers    map[string]Handler
 	readTimeout time.Duration
+	seekTimeout time.Duration
 	sig         chan os.Signal
 	logsChannel chan kafka.LogEvent
 	waiting     []chan error
@@ -48,6 +49,60 @@ type consumer struct {
 	}
 }
 
+// NewConsumer creates a new Kafka consumer with the specified configuration
+// and options.  The consumer must be started before it can receive messages.
+func NewConsumer(config *Config, opts ...ConsumerOption) (Consumer, error) {
+	handle := func(err error) (Consumer, error) {
+		return nil, fmt.Errorf("kafka.NewConsumer: %w", err)
+	}
+
+	const defaultReadTimeout = 1 * time.Second
+	const defaultSeekTimeout = 100 * time.Millisecond
+
+	consumer := &consumer{
+		handlers:    map[string]Handler{},
+		readTimeout: defaultReadTimeout,
+		seekTimeout: defaultSeekTimeout,
+	}
+
+	// clone the config and apply options
+	config = config.clone()
+	if err := consumer.configure(config, opts...); err != nil {
+		return handle(err)
+	}
+
+	// create the confluent consumer
+	var err error
+	if consumer.Consumer, err = createConsumer(&config.config); err != nil {
+		return handle(fmt.Errorf("createConsumer: %w", err))
+	}
+	consumer.funcs.close = consumer.Consumer.Close
+	consumer.funcs.commitOffsets = consumer.Consumer.CommitOffsets
+	consumer.funcs.readMessage = consumer.Consumer.ReadMessage
+	consumer.funcs.seek = consumer.Consumer.Seek
+	consumer.funcs.subscribe = consumer.Consumer.SubscribeTopics
+
+	// start a go routine to read events from the logs channel to prevent
+	// it filling up;
+	consumer.logsChannel = consumer.Consumer.Logs()
+	go func() {
+		for {
+			if _, ok := <-consumer.logsChannel; !ok {
+				consumer.logsChannel = nil
+				return
+			}
+		}
+	}()
+
+	// establish a ctrl-c signal channel for running in interactive console
+	consumer.sig = make(chan os.Signal, 1)
+	signal.Notify(consumer.sig, syscall.SIGINT, syscall.SIGTERM)
+
+	consumer.setState(csReady)
+
+	return consumer, nil
+}
+
 // commit commits the offset of the specified message.
 //
 // If readNext is true, the offset is committed as is (the consumer will read
@@ -59,12 +114,10 @@ type consumer struct {
 // If the commit fails, a FatalError is returned (the consumer will stop
 // processing messages)
 var commitOffset = func(ctx context.Context, c *consumer, msg *kafka.Message, readNext bool) error {
-	const seekTimeoutMS = 100 // FUTURE: make configurable per consumer?
-
 	offset := msg.TopicPartition
 	if readNext {
 		offset.Offset += 1
-	} else if err := c.funcs.seek(offset, seekTimeoutMS); err != nil {
+	} else if err := c.funcs.seek(offset, int(c.seekTimeout.Milliseconds())); err != nil {
 		return fmt.Errorf("commit: seek: %w", err)
 	}
 
@@ -532,53 +585,3 @@ func (c *consumer) configure(cfg *Config, opts ...ConsumerOption) error {
 
 	return nil
 }
-
-// NewConsumer creates a new Kafka consumer with the specified configuration
-// and options.  The consumer must be started before it can receive messages.
-func NewConsumer(config *Config, opts ...ConsumerOption) (Consumer, error) {
-	handle := func(err error) (Consumer, error) {
-		return nil, fmt.Errorf("kafka.NewConsumer: %w", err)
-	}
-
-	consumer := &consumer{
-		handlers:    map[string]Handler{},
-		readTimeout: 1 * time.Second,
-	}
-
-	// clone the config and apply options
-	config = config.clone()
-	if err := consumer.configure(config, opts...); err != nil {
-		return handle(err)
-	}
-
-	// create the confluent consumer
-	var err error
-	if consumer.Consumer, err = createConsumer(&config.config); err != nil {
-		return handle(fmt.Errorf("createConsumer: %w", err))
-	}
-	consumer.funcs.close = consumer.Consumer.Close
-	consumer.funcs.commitOffsets = consumer.Consumer.CommitOffsets
-	consumer.funcs.readMessage = consumer.Consumer.ReadMessage
-	consumer.funcs.seek = consumer.Consumer.Seek
-	consumer.funcs.subscribe = consumer.Consumer.SubscribeTopics
-
-	// start a go routine to read events from the logs channel to prevent
-	// it filling up;
-	consumer.logsChannel = consumer.Consumer.Logs()
-	go func() {
-		for {
-			if _, ok := <-consumer.logsChannel; !ok {
-				consumer.logsChannel = nil
-				return
-			}
-		}
-	}()
-
-	// establish a ctrl-c signal channel for running in interactive console
-	consumer.sig = make(chan os.Signal, 1)
-	signal.Notify(consumer.sig, syscall.SIGINT, syscall.SIGTERM)
-
-	consumer.setState(csReady)
-
-	return consumer, nil
-}

consumerOption.go

@@ -66,14 +66,48 @@ func MessageDecryption(fn CypherFunc) ConsumerOption {
 	}
 }
 
-// Message
+// ReadTimeoutNever is a special value for the ReadTimeout option
+// that indicates the consumer should never time out when reading
+// messages.
+//
+// Use this with caution; a consumer that never times out may be
+// ejected from the consumer group if it does not receive a message
+// within the configured session timeout.
+const ReadTimeoutNever = time.Duration(-1)
+
+// ReadTimeout sets the read timeout for the consumer.
+// If not set, a default of 1s is used.
+func ReadTimeout(timeout time.Duration) ConsumerOption {
+	return func(cfg *Config, consumer *consumer) error {
+		if timeout < 0 && timeout != ReadTimeoutNever {
+			return ErrInvalidReadTimeout
+		}
+		consumer.readTimeout = timeout
+		return nil
+	}
+}
+
+// SeekTimeout sets the timeout for seek operations on the consumer.
+// If not set, a default of 100ms is used.
+func SeekTimeout(timeout time.Duration) ConsumerOption {
+	return func(cfg *Config, consumer *consumer) error {
+		if timeout <= 0 {
+			return ErrInvalidSeekTimeout
+		}
+		consumer.seekTimeout = timeout
+		return nil
+	}
+}
+
+// TopicHandler registers a handler for a specific topic.
 func TopicHandler[T comparable](topic T, h Handler) ConsumerOption {
 	return func(_ *Config, c *consumer) error {
 		c.handlers[fmt.Sprintf("%v", topic)] = h
 		return nil
 	}
 }
 
+// TopicHandlers registers multiple handlers for specific topics.
 func TopicHandlers[T comparable](handlers map[T]Handler) ConsumerOption {
 	return func(_ *Config, c *consumer) error {
 		for topic, h := range handlers {
@@ -82,13 +116,3 @@ func TopicHandlers[T comparable](handlers map[T]Handler) ConsumerOption {
 		return nil
 	}
 }
-
-func ReadTimeout(timeout time.Duration) ConsumerOption {
-	return func(cfg *Config, consumer *consumer) error {
-		if timeout < 0 && timeout != -1 {
-			return ErrInvalidReadTimeout
-		}
-		consumer.readTimeout = timeout
-		return nil
-	}
-}

consumerOption_test.go

@@ -58,6 +58,10 @@ func TestConsumerOptions(t *testing.T) {
 						Config:   &Config{config: kafka.ConfigMap{}},
 						consumer: &consumer{readTimeout: 1 * time.Hour},
 					}},
+					{name: "SeekTimeout", sut: SeekTimeout(250 * time.Millisecond), result: result{
+						Config:   &Config{config: kafka.ConfigMap{}},
+						consumer: &consumer{seekTimeout: 250 * time.Millisecond},
+					}},
 				}
 				for _, tc := range testcases {
 					t.Run(tc.name, func(t *testing.T) {
@@ -102,6 +106,7 @@ func TestConsumerConfigurationErrorHandling(t *testing.T) {
 		{name: "auto offset reset (invalid)", sut: AutoOffsetReset("invalid"), result: cfgerr},
 		{name: "group id", sut: ConsumerGroupID("group"), result: cfgerr},
 		{name: "read timeout", sut: ReadTimeout(-2), result: ErrInvalidReadTimeout},
+		{name: "seek timeout", sut: SeekTimeout(-2), result: ErrInvalidSeekTimeout},
 	}
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {

errors.go

@@ -12,7 +12,8 @@ var (
 	ErrConsumerNotRunning   = errors.New("consumer is not running")
 	ErrConsumerNotStarted   = errors.New("consumer has not been started")
 	ErrInvalidOperation     = errors.New("invalid operation")
-	ErrInvalidReadTimeout   = errors.New("invalid read timeout (must be -1 or >= 0)")
+	ErrInvalidReadTimeout   = errors.New("invalid read timeout (must be kafka.ReadTimeoutNever or >= 0ms)")
+	ErrInvalidSeekTimeout   = errors.New("invalid seek timeout (must be > 0ms)")
 	ErrNoHandler            = errors.New("no handler for topic")
 	ErrNoHandlersConfigured = errors.New("no handlers configured")
 	ErrReprocessMessage     = errors.New("message will be reprocessed")