Kafka Reader unable to read events from kafka when using StartOffset: kafka.LastOffset and GroupID when groupID is new

**Describe the bug**

When using Kafka Reader to read messages from kafka with StartOffset set to last offset and group id was never used before the Reader just doesn't read any messages

**Kafka Version**

> * What version(s) of Kafka are you testing against?
7.6.0
> * What version of kafka-go are you using?
v0.4.47

**To Reproduce**

> Resources to reproduce the behavior:

```yaml
# docker compose file generated with https://github.com/sknop/kafka-docker-composer
services:
    controller-1:
        image: confluentinc/cp-server:7.6.0
        hostname: controller-1
        container_name: controller-1

        environment:
            KAFKA_NODE_ID: 1
            CLUSTER_ID: Nk018hRAQFytWskYqtQduw
            KAFKA_PROCESS_ROLES: controller
            KAFKA_LISTENERS: CONTROLLER://controller-1:19091
            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT
            KAFKA_INTER_BROKER_LISTENER_NAME: CONTROLLER
            KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
            KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:19091
            KAFKA_JMX_PORT: 9999
            KAFKA_JMX_HOSTNAME: controller-1
            KAFKA_BROKER_RACK: rack-0
            KAFKA_DEFAULT_REPLICATION_FACTOR: 2
            KAFKA_OFFSET_REPLICATION_FACTOR: 2
            KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 2
            KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 2
            KAFKA_CONFLUENT_METADATA_TOPIC_REPLICATION_FACTOR: 2
            KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 2
            KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
            KAFKA_CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 2
            KAFKA_OPTS: -javaagent:/tmp/jmx_prometheus_javaagent-0.20.0.jar=8091:/tmp/kafka_config.yml
            KAFKA_CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093
        cap_add:
            - NET_ADMIN
        ports:
            - 19091:19091
        volumes:
            - $PWD/volumes/jmx_prometheus_javaagent-0.20.0.jar:/tmp/jmx_prometheus_javaagent-0.20.0.jar
            - $PWD/volumes/kafka_config.yml:/tmp/kafka_config.yml

    kafka-1:
        image: confluentinc/cp-server:7.6.0
        hostname: kafka-1
        container_name: kafka-1

        healthcheck:
            test: curl -fail --silent http://kafka-1:8090/kafka/v3/clusters/ --output /dev/null || exit 1
            interval: 10s
            retries: 10
            start_period: 20s
        depends_on:
            - controller-1
        environment:
            KAFKA_LISTENERS: PLAINTEXT://kafka-1:19092, EXTERNAL://0.0.0.0:9091
            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:19092, EXTERNAL://localhost:9091
            KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
            KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
            KAFKA_JMX_PORT: 10001
            KAFKA_JMX_HOSTNAME: localhost
            KAFKA_BROKER_RACK: rack-0
            KAFKA_OPTS: -javaagent:/tmp/jmx_prometheus_javaagent-0.20.0.jar=8091:/tmp/kafka_config.yml
            KAFKA_MIN_INSYNC_REPLICAS: 1
            KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
            KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 2
            KAFKA_CONFLUENT_CLUSTER_LINK_ENABLE: False
            KAFKA_CONFLUENT_REPORTERS_TELEMETRY_AUTO_ENABLE: False
            KAFKA_NODE_ID: 2
            CLUSTER_ID: Nk018hRAQFytWskYqtQduw
            KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:19091
            KAFKA_PROCESS_ROLES: broker
            KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
            KAFKA_CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093
            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
        cap_add:
            - NET_ADMIN
        ports:
            - 9091:9091
            - 10001:10001
            - 10101:8091
            - 10201:8090
        volumes:
            - $PWD/volumes/jmx_prometheus_javaagent-0.20.0.jar:/tmp/jmx_prometheus_javaagent-0.20.0.jar
            - $PWD/volumes/kafka_config.yml:/tmp/kafka_config.yml

    kafka-2:
        image: confluentinc/cp-server:7.6.0
        hostname: kafka-2
        container_name: kafka-2

        healthcheck:
            test: curl -fail --silent http://kafka-2:8090/kafka/v3/clusters/ --output /dev/null || exit 1
            interval: 10s
            retries: 10
            start_period: 20s
        depends_on:
            - controller-1
        environment:
            KAFKA_LISTENERS: PLAINTEXT://kafka-2:19093, EXTERNAL://0.0.0.0:9092
            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:19093, EXTERNAL://localhost:9092
            KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
            KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
            KAFKA_JMX_PORT: 10002
            KAFKA_JMX_HOSTNAME: localhost
            KAFKA_BROKER_RACK: rack-0
            KAFKA_OPTS: -javaagent:/tmp/jmx_prometheus_javaagent-0.20.0.jar=8091:/tmp/kafka_config.yml
            KAFKA_MIN_INSYNC_REPLICAS: 1
            KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
            KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 2
            KAFKA_CONFLUENT_CLUSTER_LINK_ENABLE: False
            KAFKA_CONFLUENT_REPORTERS_TELEMETRY_AUTO_ENABLE: False
            KAFKA_NODE_ID: 3
            CLUSTER_ID: Nk018hRAQFytWskYqtQduw
            KAFKA_CONTROLLER_QUORUM_VOTERS: 1@controller-1:19091
            KAFKA_PROCESS_ROLES: broker
            KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
            KAFKA_CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093
            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
        cap_add:
            - NET_ADMIN
        ports:
            - 9092:9092
            - 10002:10002
            - 10102:8091
            - 10202:8090
        volumes:
            - $PWD/volumes/jmx_prometheus_javaagent-0.20.0.jar:/tmp/jmx_prometheus_javaagent-0.20.0.jar
            - $PWD/volumes/kafka_config.yml:/tmp/kafka_config.yml

    kafka-ui:
        container_name: kafka-ui
        image: provectuslabs/kafka-ui:latest
        ports:
            - 8080:8080
        depends_on:
            - controller-1
            - kafka-1
            - kafka-2
        environment:
            KAFKA_CLUSTERS_0_NAME: local
            KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: controller-1:19091,kafka-1:19092,kafka-2:19093
            KAFKA_CLUSTERS_0_METRICS_PORT: 9999
            DYNAMIC_CONFIG_ENABLED: 'true'




    portainer:
        image: portainer/portainer-ce:2.20.2
        container_name: portainer
        restart: unless-stopped
        ports:
        - 9000:9000
        volumes:
        - /var/run/docker.sock:/var/run/docker.sock
        - portainer_data:/data

volumes:
  portainer_data:
```

```go
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"example.com/kafka/util/configurator"
	timeutil "example.com/kafka/util/time-util"
	"github.com/segmentio/kafka-go"
)

var (
	cfgFileName string
)

func init() {
	flag.StringVar(&cfgFileName, "cfg", "/home/veljko/Desktop/Faks/Master Studije/Prva Godina/2. Semestar/Big Data инфраструктуре и сервиси/Projekat/kod/Kafka/Code/kafka/doc/consumer.conf", "File path to the file containing the configuration")
}

type Message struct {
	ID      string `json:"id"`
	Type    int    `json:"type"`
	Kind    int    `json:"kind"`
	Content string `json:"content"`
}

func main() {
	flag.Parse()
	defer timeutil.MeasureExecutionTime(time.Now())

	var config configurator.KafkaConsumerConfig

	// Load configuration
	err := configurator.LoadConfiguration(cfgFileName, &config)
	if err != nil {
		log.Fatalf("Failed to load configuration: %s", err)
	}

	// Create new Kafka reader
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     strings.Split(config.BootstrapServers, ","),
		GroupID:     config.GroupID,
		Topic:       config.Topic,
		StartOffset: kafka.LastOffset,
		MaxBytes:    config.MaxBytes, // 10MB max message size
	})

	// Close the reader when main function exits
	defer r.Close()

	// Handle OS signals for graceful shutdown
	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

	// Start consuming messages
	for {
		select {
		case <-sigchan:
			slog.Info("Received shutdown signal. Shutting down consumer...")
			return
		default:
			// Read a message from Kafka
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			msg, err := r.ReadMessage(ctx)
			cancel()

			if err != nil {
				slog.Error(fmt.Sprintf("Error reading message: %v", err))
				continue
			}

			// Process the message
			var message Message
			err = json.Unmarshal(msg.Value, &message)
			if err != nil {
				slog.Error(fmt.Sprintf("Error decoding message: %v", err))
				continue
			}

			// Handle the message
			slog.Info(fmt.Sprintf("Received message: %+v\n", message))
		}
	}
}
```
Config for go program
```json
{
    "topic":"json_test_topic",
    "bootstrapServers":"localhost:9090,localhost:9091,localhost:9092",
    "groupId": "test-group",
    "batchBytes": 1000000
}
```

**Expected Behavior**

Program should consume messages from kafka and output them to console

**Observed Behavior**
Nothing happens
```
2024/05/13 22:22:21 ERROR Error reading message: fetching message: context deadline exceeded
^C2024/05/13 22:22:26 ERROR Error reading message: fetching message: context deadline exceeded
2024/05/13 22:22:26 INFO Received shutdown signal. Shutting down consumer...
2024/05/13 22:22:34 INFO Elapsed time: 18.00940981s
```

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Kafka Reader unable to read events from kafka when using StartOffset: kafka.LastOffset and GroupID when groupID is new #1291

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Kafka Reader unable to read events from kafka when using StartOffset: kafka.LastOffset and GroupID when groupID is new #1291

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions