fix some kafka 4.0 ci errors

Author: maxwolf8852

.circleci/config.yml

@@ -178,10 +178,10 @@ jobs:
       entrypoint: *entrypoint
     steps: *steps
 
-  kafka-400:
+  kafka-370-kraft:
     working_directory: *working_directory
     environment:
-      KAFKA_VERSION: "4.0.0"
+      KAFKA_VERSION: "3.7.0"
 
       # Need to skip nettest to avoid these kinds of errors:
       #  --- FAIL: TestConn/nettest (17.56s)
@@ -195,13 +195,17 @@ jobs:
       KAFKA_SKIP_NETTEST: "1"
     docker:
       - image: circleci/golang
-      - image: bitnami/kafka:4.0.0
+      - image: bitnami/kafka:3.7.0
         ports:
           - 9092:9092
           - 9093:9093
-        environment:
+        environment: &kraft-env
+          KAFKA_KRAFT_MODE: "true"
           KAFKA_CFG_NODE_ID: 1
           KAFKA_CFG_BROKER_ID: 1
+          KAFKA_CLUSTER_ID: 1
+          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
           KAFKA_CFG_PROCESS_ROLES: broker,controller
           KAFKA_CFG_ADVERTISED_HOST_NAME: 'localhost'
           KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
@@ -212,21 +216,45 @@ jobs:
           KAFKA_CFG_SASL_ENABLED_MECHANISMS: 'PLAIN,SCRAM-SHA-256,SCRAM-SHA-512'
           KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@localhost:9094
           ALLOW_PLAINTEXT_LISTENER: yes
-          KAFKA_CFG_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
-          KAFKA_OPTS: "-Djava.security.auth.login.config=/opt/bitnami/kafka/config/kafka_jaas.conf"
-          KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true'
-          KAFKA_CFG_DELETE_TOPIC_ENABLE: 'true'
+          KAFKA_CFG_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
+          KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: true
+          KAFKA_CFG_DELETE_TOPIC_ENABLE: true
           KAFKA_CFG_MESSAGE_MAX_BYTES: '200000000'
           KAFKA_CFG_AUTHORIZER_CLASS_NAME: 'org.apache.kafka.metadata.authorizer.StandardAuthorizer'
+          KAFKA_OPTS: "-Djava.security.auth.login.config=/opt/bitnami/kafka/config/kafka_jaas.conf"
           KAFKA_CFG_SUPER_USERS: User:adminscram256;User:adminscram512;User:adminplain
           KAFKA_CLIENT_USERS: adminscram256,adminscram512,adminplain
           KAFKA_CLIENT_PASSWORDS: admin-secret-256,admin-secret-512,admin-secret
           KAFKA_CLIENT_SASL_MECHANISMS: SCRAM-SHA-256,SCRAM-SHA-512,PLAIN
           KAFKA_INTER_BROKER_USER: adminscram512
           KAFKA_INTER_BROKER_PASSWORD: admin-secret-512
           KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL: SCRAM-SHA-512
+        entrypoint: *entrypoint
     steps: *steps
 
+  kafka-400:
+    working_directory: *working_directory
+    environment:
+      KAFKA_VERSION: "4.0.0"
+
+      # Need to skip nettest to avoid these kinds of errors:
+      #  --- FAIL: TestConn/nettest (17.56s)
+      #    --- FAIL: TestConn/nettest/PingPong (7.40s)
+      #      conntest.go:112: unexpected Read error: [7] Request Timed Out: the request exceeded the user-specified time limit in the request
+      #      conntest.go:118: mismatching value: got 77, want 78
+      #      conntest.go:118: mismatching value: got 78, want 79
+      # ...
+      #
+      # TODO: Figure out why these are happening and fix them (they don't appear to be new).
+      KAFKA_SKIP_NETTEST: "1"
+    docker:
+      - image: circleci/golang
+      - image: bitnami/kafka:4.0.0
+        ports:
+          - 9092:9092
+          - 9093:9093
+        environment: *kraft-env
+    steps: *steps
 workflows:
   version: 2
   run:
@@ -236,4 +264,5 @@ workflows:
     - kafka-270
     - kafka-281
     - kafka-370
+    - kafka-370-kraft
     - kafka-400

alterclientquotas_test.go

@@ -3,9 +3,11 @@ package kafka
 import (
 	"context"
 	"testing"
+	"time"
 
-	ktesting "github.com/segmentio/kafka-go/testing"
 	"github.com/stretchr/testify/assert"
+
+	ktesting "github.com/segmentio/kafka-go/testing"
 )
 
 func TestClientAlterClientQuotas(t *testing.T) {
@@ -65,6 +67,8 @@ func TestClientAlterClientQuotas(t *testing.T) {
 
 	assert.Equal(t, expectedAlterResp, *alterResp)
 
+	time.Sleep(1 * time.Second) // wait for the quota to be applie (Kafka 4.0.0+)
+
 	describeResp, err := client.DescribeClientQuotas(context.Background(), &DescribeClientQuotasRequest{
 		Components: []DescribeClientQuotasRequestComponent{
 			{

consumergroup_test.go

@@ -250,12 +250,12 @@ func TestConsumerGroup(t *testing.T) {
 			scenario: "Next returns generations",
 			function: func(t *testing.T, ctx context.Context, cg *ConsumerGroup) {
 				gen1, err := cg.Next(ctx)
-				if gen1 == nil {
-					t.Fatalf("expected generation 1 not to be nil")
-				}
 				if err != nil {
 					t.Fatalf("expected no error, but got %+v", err)
 				}
+				if gen1 == nil {
+					t.Fatalf("expected generation 1 not to be nil")
+				}
 				// returning from this function should cause the generation to
 				// exit.
 				gen1.Start(func(context.Context) {})

describegroups_test.go

@@ -32,7 +32,7 @@ func TestClientDescribeGroups(t *testing.T) {
 		Topic: topic,
 	})
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) // longer timeout for kafka-4.0
 	defer cancel()
 
 	err := w.WriteMessages(

electleaders_test.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"testing"
+	"time"
 
 	ktesting "github.com/segmentio/kafka-go/testing"
 )
@@ -26,6 +27,7 @@ func TestClientElectLeaders(t *testing.T) {
 		&ElectLeadersRequest{
 			Topic:      topic,
 			Partitions: []int{0, 1},
+			Timeout:    5 * time.Second,
 		},
 	)
 

listgroups_test.go

@@ -55,7 +55,7 @@ func TestClientListGroups(t *testing.T) {
 		Topic: topic,
 	})
 
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) // longer timeout for kafka-4.0
 	defer cancel()
 
 	err := w.WriteMessages(

writer_test.go

@@ -856,7 +856,8 @@ func testWriterAutoCreateTopic(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
 		err = w.WriteMessages(ctx, msg)
-		if errors.Is(err, LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
+		if errors.Is(err, LeaderNotAvailable) || errors.Is(err, UnknownTopicOrPartition) ||
+			errors.Is(err, context.DeadlineExceeded) {
 			time.Sleep(time.Millisecond * 250)
 			continue
 		}
@@ -924,7 +925,8 @@ func testWriterSasl(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
 		err = w.WriteMessages(ctx, msg)
-		if errors.Is(err, LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {
+		if errors.Is(err, LeaderNotAvailable) || errors.Is(err, UnknownTopicOrPartition) ||
+			errors.Is(err, context.DeadlineExceeded) {
 			time.Sleep(time.Millisecond * 250)
 			continue
 		}