failed to close batch:[7] Request Timed Out: the request exceeded the user-specified time limit in the request

[mj-nehme]

I asked this question on Stack Overflow, but didn't get any response. I am copy-pasting it here with some relevant modifications.

The Problem In a nutshell:

It's my first use of Kafka-go. I installed everything locally with the default configuration on a clean machine and used the very first two code samples: Produce Messages and Consume Messages as follows. But I got the following error:

failed to close batch:[7] Request Timed Out: the request exceeded the user-specified time limit in the request

The Problem With More Details:

I downloaded Kafka-go and created a very simple project that only produces three messages, consumes them, and closes the batch (+connect+disconnect).

Please note that I just copy-pasted the code from the main page.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

func main() {

	ProduceMessages()
	ConsumeMessages()
}

func ProduceMessages() {
	// to produce messages
	topic := "my-topic"
	partition := 0

	conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
	_, err = conn.WriteMessages(
		kafka.Message{Value: []byte("one!")},
		kafka.Message{Value: []byte("two!")},
		kafka.Message{Value: []byte("three!")},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	if err := conn.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}
}

func ConsumeMessages() {
	// to consume messages
	topic := "my-topic"
	partition := 0

	conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	conn.SetReadDeadline(time.Now().Add(10 * time.Second))
	batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

	b := make([]byte, 10e3) // 10KB max per message
	for {
		n, err := batch.Read(b)
		if err != nil {
			break
		}
		fmt.Println(string(b[:n]))
	}

	if err := batch.Close(); err != nil {
		log.Fatal("failed to close batch:", err)
	}

	if err := conn.Close(); err != nil {
		log.Fatal("failed to close connection:", err)
	}
}

However, whenever trying to close the batch session, I am getting the following result (after waiting a bunch of seconds):

one!
two!
three!

Well, that's nice so far, but this is always followed by the following error.

2022/12/10 10:35:50 failed to close batch:[7] Request Timed Out: the request exceeded the user-specified time limit in the request
exit status 1

For your consideration, I used the following env:

• Go 1.19.3
• minimum required go in the project 1.19
• github.com/klauspost/compress v1.15.12 // indirect
• github.com/pierrec/lz4/v4 v4.1.17 // indirect
• github.com/segmentio/kafka-go v0.4.38 // indirect
• Kafka kafka_2.13-3.3.1

Question:

Why is it failing to close the batch (why is it timed out)?