FetchMessage Why submit offset automatically

[yangtongbing]

I tried several ways, including setting CommitInterval, to commit automatically, and didn't commit offset through CommitMessages as expected. Here is my sample code.

package main

import (
	"context"
	"fmt"
	kafka "github.com/segmentio/kafka-go"
)

func main() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{"127.0.0.1:9092"},
		Topic:    "test",
		GroupID:  "9",
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
	})
	defer r.Close()
	ctx := context.Background()
	for {
		m, err := r.FetchMessage(ctx)
		if err != nil {
			break
		}
		fmt.Printf("message at offset %d：%s\n", m.Offset, string(m.Value))
		if shouldCommit(m) {
			err = r.CommitMessages(ctx, m)
			if err != nil {
				fmt.Println("commit failed", err)
			}
		}
	}
}

func shouldCommit(message kafka.Message) bool {
	return false
}

[yangtongbing]

go version go1.22.4 darwin/arm64
kafka-go version is v0.4.47

[luffyao]

why did you think the FetchMessage function will submit offset automatically? according to code, it will not submit offset , and also it doesn't appear this issue in our env.

[yangtongbing]

In my own tests, instead of committing, shouldCommit returning false will still automatically consume the next offset

[oggyunao]

plz use QueueCapacity:1,and try again..

r := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{"127.0.0.1:9092"},
Topic: "test",
GroupID: "9",
MinBytes: 10e3, // 10KB
MaxBytes: 10e6, // 10MB
QueueCapacity:1,
})

[yangtongbing]

Ok, I'll try

[yangtongbing]

I have tested it and configured QueueCapacity equal to 1 to automatically commit offset without CommitMessages being explicitly called