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.

[bcambl]

server-side log:

org.apache.kafka.common.errors.UnsupportedVersionException: Received request for api with key 11 (JoinGroup) and unsupported version 1
[2025-03-18 22:22:32,964] ERROR Closing socket for 0:0:0:0:0:0:0:1:9092-0:0:0:0:0:0:0:1:44238-0-24 because of error (kafka.network.Processor)
org.apache.kafka.common.errors.UnsupportedVersionException: Received request for api with key 11 (JoinGroup) and unsupported version 1
[2025-03-18 22:22:32,965] ERROR Exception while processing request from 0:0:0:0:0:0:0:1:9092-0:0:0:0:0:0:0:1:44238-0-24 (kafka.network.Processor)

[ijuma]

Apache Kafka 4.0 was released this week: https://kafka.apache.org/blog#apache_kafka_400_release_announcement

[CelinL123]

Is there any update for this issue? Seeing same server-side error as above when upgrading to bitnami/kafka 4.0.0

[KinNeko-De]

I also used bitnami/kafka 4.0.0 and got this error when I am using a GroupID. Reading without GroupID works fine.

The Client side error message is: fetching message: EOF

[petedannemann]

I did a quick audit of the protocol changes and this is my summary of what we are affected by. It's possible I may have missed something here.

Note all the things affected are just for Conn and things that use it like Reader. The Client appears to support Kafka 4.0.

Affected:

1. JoinGroup - affected, we need to support V2, currently we only support V1
2. CreateTopics - affected, we need to support V2, currently we only support V0
3. DeleteTopics - affected, we need to support V1, currently we only support V0

Unaffected:

1. Produce - unaffected, we support V7
2. Fetch - unaffected, we support V10
3. OffsetCommit - unaffected, we support V2
4. OffsetFetch - unaffected, we support V1
5. DescribeAcls - unaffected, we support up to V3
6. CreateAcls - unaffected, we support up to V3
7. DeleteAcls - unaffected, we support up to V3
8. DescribeConfigs - unaffected, we support up to V3

[petedannemann]

Given how we claim to support > Kafka 0.10 in the README it seems like we could just update the protocol versions for these APIs and push out a new v0.5.0. The long term best fix would be to adapt Conn to use Client, which by design negotiates API versions and selects the latest. It also has support for many more API versions. That is a big lift and had issues in the past though #1027

[petedannemann]

Can someone from Confluent comment on this issue? It seems like Kafka 4.0 memory usage is unpredictable and JVM configurations for heap memory are no longer respected. Kafka 4.0 requests fail often due to running out of heap memory regardless of how much memory Kafka is given and this makes it difficult for us to test against. I guess it's possible this may just be a Bitnami image issue, but I've haven't been able to find sufficient examples for SASL using the Apache Kafka image to get that to work
#1379 (comment)

[petedannemann]

I gave Kafka 4GB of memory and the tests now sometimes pass without running out of memory. Maybe using Kraft mode dramatically increases the memory requirements. Previously I think Kafka used like 200MB of memory during our tests

[ijuma]

It seems like Kafka 4.0 memory usage is unpredictable and JVM configurations for heap memory are no longer respected

Neither or these is expected and there haven't been such reports filed in the issue tracker either.

[petedannemann]

We have only seen the test failures due to Kafka running out of memory for tests involving consumer groups. VisualVM shows a dramatic increase of allocations from the "group-coordinator-event-processor" Kafka threads when running these tests. This memory never seems to get reclaimed and the increase occurs when we create several consumer groups

[dajac]

@petedannemann This is weird. I am not aware of any issues in 4.0 affecting the memory allocation. Is there a way to reproduce the test? Could you share the heap dump?

[Rajvira10]

func NewReader(cfg *config.Config, topic string) *kafka.Reader {
mechanism := plain.Mechanism{
Username: cfg.KafkaUsername,
Password: cfg.KafkaPassword,
}

return kafka.NewReader(kafka.ReaderConfig{
	Brokers:  []string{cfg.KafkaBroker},
	Topic:    topic,
	GroupID:  "hellomello",
	MinBytes: 10e3,
	MaxBytes: 10e8,
	MaxWait:  100 * time.Millisecond,
	Dialer: &kafka.Dialer{
		SASLMechanism: mechanism,
	},
	Logger: kafka.LoggerFunc(func(msg string, args ...interface{}) {
		fmt.Printf(msg, args...)
	},
	),
	ErrorLogger: kafka.LoggerFunc(func(msg string, args ...interface{}) {
		fmt.Printf(msg, args...)
	},
	),
})

}

Am getting the same errors mentioned above for my code and I have the latest Kafka and latest kafka-go

[petedannemann]

@dajac to reproduce this issue you can checkout the branch associated with this PR where we fixed the API deprecation issues and added a docker-compose file for running kafka 4.0.

Then run this

# Launch Kafka 4.0
docker compose -f docker_compose_versions/docker-compose-400.yml up -d 

# KAFKA_VERSION env var is required to run tests that hit Kafka running locally
export KAFKA_VERSION=4.0.0

# Memory appears to increase with each time the tests are run up to some limit
go test -count 10 -run TestReaderConsumerGroup github.com/segmentio/kafka-go

I'll try to attach a heapdump sometime soon