[build] Optimize Kafka container to use KRaft instead of relying on ZooKeeper (#5167)

Author: liyubin117

paimon-e2e-tests/src/test/java/org/apache/paimon/tests/E2eTestBase.java

@@ -106,13 +106,9 @@ public void before() throws Exception {
                         .withStartupTimeout(Duration.ofMinutes(3))
                         .withLocalCompose(true);
         if (withKafka) {
-            List<String> kafkaServices = Arrays.asList("zookeeper", "kafka");
-            services.addAll(kafkaServices);
-            for (String s : kafkaServices) {
-                environment.withLogConsumer(s + "-1", new Slf4jLogConsumer(LOG));
-            }
-            environment.waitingFor(
-                    "kafka-1", buildWaitStrategy(".*Recorded new ZK controller.*", 2));
+            services.add("kafka");
+            environment.withLogConsumer("kafka-1", new Slf4jLogConsumer(LOG));
+            environment.waitingFor("kafka-1", buildWaitStrategy(".*Kafka Server started.*", 1));
         }
         if (withHive) {
             List<String> hiveServices =

paimon-e2e-tests/src/test/resources-filtered/docker-compose.yaml

@@ -76,39 +76,33 @@ services:
   # Kafka services
   # ----------------------------------------
 
-  zookeeper:
-    image: confluentinc/cp-zookeeper:7.8.0
-    networks:
-      testnetwork:
-        aliases:
-          - zookeeper
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-      ZOOKEEPER_TICK_TIME: 2000
-    expose:
-      - "2181"
-
   kafka:
     image: confluentinc/cp-kafka:7.8.0
     networks:
       testnetwork:
         aliases:
           - kafka
     environment:
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_INTERNAL://kafka:29092
+      KAFKA_NODE_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
+      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://kafka:9092'
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_JMX_PORT: 9101
+      KAFKA_JMX_HOSTNAME: localhost
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
+      KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
       KAFKA_TRANSACTION_MAX_TIMEOUT_MS: 7200000
       # Disable log deletion to prevent records from being deleted during test run
       KAFKA_LOG_RETENTION_MS: -1
+      CLUSTER_ID: Pkv3OFSRNQcNATKEMDV9Ql
     expose:
       - "9092"
-    depends_on:
-      - zookeeper
 
   # ----------------------------------------
   # Hive services, copied and modified from https://github.com/big-data-europe/docker-hive