chore: fix tests

Author: JivusAyrus

.github/workflows/router-ci.yaml

@@ -150,6 +150,22 @@ jobs:
           password: ${{secrets.DOCKER_PASSWORD}}
         ports:
           - 6379:6379
+      kafka:
+        image: bitnami/kafka:3.7.0
+        env:
+          KAFKA_ENABLE_KRAFT: yes
+          KAFKA_CFG_PROCESS_ROLES: controller,broker
+          KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
+          KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+          KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+          KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
+          KAFKA_CFG_TRANSACTION_PARTITION_VERIFICATION_ENABLE: false
+          KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+          KAFKA_CFG_NODE_ID: 1
+          ALLOW_PLAINTEXT_LISTENER: yes
+          KAFKA_KRAFT_CLUSTER_ID: XkpGZQ27R3eTl3OdTm2LYA # 16 byte base64-encoded UUID
+        ports:
+          - '9092:9092'
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/go

router-tests/testenv/testenv.go

@@ -364,22 +364,36 @@ func createTestEnv(t testing.TB, cfg *Config) (*Environment, error) {
 	}
 
 	if cfg.EnableKafka {
-		kafkaStarted.Add(1)
-		go func() {
-			defer kafkaStarted.Done()
+		if os.Getenv("CI") == "true" {
+			cfg.KafkaSeeds = []string{"localhost:9092"}
 
-			var kafkaSetupErr error
-			kafkaSetup, kafkaSetupErr = setupKafkaServer(t)
-			if kafkaSetupErr != nil || kafkaSetup == nil {
-				t.Fatalf("could not setup kafka: %s", kafkaSetupErr.Error())
-				return
+			client, err := kgo.NewClient(
+				kgo.SeedBrokers(cfg.KafkaSeeds...),
+			)
+			if err != nil {
+				return nil, err
 			}
 
-			kafkaClient = kafkaSetup.Client
+			kafkaClient = client
 			kafkaAdminClient = kadm.NewClient(kafkaClient)
+		} else {
+			kafkaStarted.Add(1)
+			go func() {
+				defer kafkaStarted.Done()
+
+				var kafkaSetupErr error
+				kafkaSetup, kafkaSetupErr = setupKafkaServer(t)
+				if kafkaSetupErr != nil || kafkaSetup == nil {
+					t.Fatalf("could not setup kafka: %s", kafkaSetupErr.Error())
+					return
+				}
 
-			cfg.KafkaSeeds = kafkaSetup.Brokers
-		}()
+				kafkaClient = kafkaSetup.Client
+				kafkaAdminClient = kadm.NewClient(kafkaClient)
+
+				cfg.KafkaSeeds = kafkaSetup.Brokers
+			}()
+		}
 	}
 
 	if cfg.EnableNats {