Compatibility with Apache Kafka 4.0

[ijuma]

Apache Kafka 4.0 will remove a number of very old protocol API versions as specified by KIP-896. I was trying to understand if this client will work correctly with it and it was not clear, particularly when it comes to the consumer group apis.

So, I thought it would be simplest to ask the project maintainers. :) Note that the 4.0 branch of Apache Kafka contains the KIP-896 changes in case you would like to test your client.

[erushing]

I believe kafka-go will work with Kafka 4.0 in general because Kafka 2.x protocol support was added to kafka-go and that is not going away with Kafka 4. You will not get the very latest protocol versions with kafka-go yet, but the existing protocols should work, generally speaking. It appears that even if Apache removed support for all <2.1 protocol versions, kafka-go would still function similarly to today.

We don't yet have plans to test Kafka 4 as it is still most likely 1-2 months from release.

[ijuma]

Thanks for the response. Let me ask a more concrete question, the following file seems to show V1, but not V2+. While KIP-896 removes both V0 and V1 for the JoinGroup request. Won't that be an issue?

https://github.com/segmentio/kafka-go/blob/main/joingroup.go

[erushing]

I believe this is actually the file you would want to look at to determine which versions are supported. You can see the various fields handled differently for various versions, as appropriate.
https://github.com/segmentio/kafka-go/blob/main/protocol/joingroup/joingroup.go#L9

The way the code is written in your link, there is a V1 hardcoded request struct for whatever reason and then the more generic JoinGroupRequest struct should accommodate up to v7, as far as I understand this code. I don't think the struct names V1, etc. are indicative of what exactly is supported anymore. If you look at Kafka Protocol here (https://kafka.apache.org/protocol.html#The_Messages_JoinGroup), the fields up to v7 are accounted for in the kafka-go code. I spot checked several of these myself before my first message and did not find any APIs that didn't accommodate Kafka 2.1.

The V1 code is linked to this extremely old PR.
#81
It appears at the time (2018), there were V2 structs, but later on in this 2nd PR, a more flexible approach was introduced and it supports up to V7. This is the style I see for all the APIs I checked, but I can't guarantee 100% compatibility at this time.
#943

[ijuma]

Thanks. I was unsure what is actually used given that some stuff was reverted in #1027. It is indeed great if the current version of the client works with Apache Kafka 4.0. The other thing that would be useful to understand is the oldest segment io version that works with Apache Kafka 4.0. Is there a way to know that?

[milindl]

I tested kafka-go with Apache Kafka 4.0 (building and running the branch mentioned above).

The latest version (0.4.47) doesn't seem to work with Kafka 4.0 as it sends a JoinGroupRequest v1 (code ref). I checked this with wireshark, too, to see what the client was actually sending out rather than just basing it on my reading of the code.

It fails with a message like this in the logs:

Failed to join group test-group: EOF

This is what my code looks like

func ConsumeUntilEnd() {
	var err error

	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     []string{"localhost:9092"},
		Topic:       "test-topic-1part",
		GroupID:     "test-group",
		Logger:      kafka.LoggerFunc(logf), // just logs everything with a newline
		ErrorLogger: kafka.LoggerFunc(logf),
	})

	done := false
	go func() {
		// Sleep for 5s
		time.Sleep(5 * time.Second)
		done = true
	}()

	for !done {
		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
		defer cancel()
		m, err := r.ReadMessage(ctx)
		if err != nil && ctx.Err() != context.DeadlineExceeded {
			panic(err)
		} else if ctx.Err() == context.DeadlineExceeded {
			continue
		}

		fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
	}

	fmt.Println("Closing...")
	err = r.Close()

	if err != nil {
		panic("could not close reader " + err.Error())
	}
}

Hopefully this is helpful and I'm not making a mistake in the testing code.

[ijuma]

Thanks @milindl. For the project maintainers, would it be possible to fix this before the Apache Kafka 4.0 release? It would be very useful to have a working client version when the release lands.

[nhaq-confluent]

@petedannemann Would you be the right person to comment on this?

[petedannemann]

We will be happy to review a PR for supporting Kafka 4.0 if it is needed

[ijuma]

We will leave it to the maintainers/community of this project to work on a patch. We wanted to make sure you were aware that the client will not work with Apache Kafka 4.0 without some (seemingly minor) changes.