[FLINK-36278] Unify container creation in Kafka tests

Using KafkaUtil will guarantee unified log settings.

Author: AHeise

flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.api.common.time.Deadline;
 import org.apache.flink.connector.kafka.testutils.DockerImageVersions;
+import org.apache.flink.connector.kafka.testutils.KafkaUtil;
 import org.apache.flink.connector.testframe.container.FlinkContainers;
 import org.apache.flink.connector.testframe.container.TestcontainersSettings;
 import org.apache.flink.test.resources.ResourceTestUtils;
@@ -44,7 +45,6 @@
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.Network;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.utility.DockerImageName;
 
 import java.nio.file.Path;
@@ -61,7 +61,6 @@
 /** End-to-end test for SQL client using Avro Confluent Registry format. */
 public class SQLClientSchemaRegistryITCase {
     private static final Logger LOG = LoggerFactory.getLogger(SQLClientSchemaRegistryITCase.class);
-    private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
 
     public static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
     public static final String INTER_CONTAINER_REGISTRY_ALIAS = "registry";
@@ -78,10 +77,9 @@ public class SQLClientSchemaRegistryITCase {
 
     @ClassRule
     public static final KafkaContainer KAFKA =
-            new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+            KafkaUtil.createKafkaContainer()
                     .withNetwork(NETWORK)
-                    .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS)
-                    .withLogConsumer(LOG_CONSUMER);
+                    .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
 
     @ClassRule
     public static final SchemaRegistryContainer REGISTRY =

flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java

@@ -62,7 +62,6 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 
-import static org.apache.flink.connector.kafka.testutils.DockerImageVersions.KAFKA;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -78,7 +77,7 @@ class SmokeKafkaITCase {
 
     @Container
     public static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA)
+            createKafkaContainer()
                     .withEmbeddedZookeeper()
                     .withNetwork(NETWORK)
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java

@@ -47,7 +47,6 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
-import static org.apache.flink.connector.kafka.testutils.DockerImageVersions.KAFKA;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -61,7 +60,7 @@ class FlinkKafkaInternalProducerITCase {
 
     @Container
     private static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA).withEmbeddedZookeeper();
+            createKafkaContainer().withEmbeddedZookeeper();
 
     @Test
     void testInitTransactionId() {

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java

@@ -101,7 +101,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.LongStream;
 
-import static org.apache.flink.connector.kafka.testutils.DockerImageVersions.KAFKA;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
@@ -124,7 +123,7 @@ public class KafkaSinkITCase extends TestLogger {
 
     @ClassRule
     public static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA)
+            createKafkaContainer()
                     .withEmbeddedZookeeper()
                     .withNetwork(NETWORK)
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLogITCase.java

@@ -44,7 +44,6 @@
 import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.Ongoing;
 import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.PrepareAbort;
 import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.PrepareCommit;
-import static org.apache.flink.connector.kafka.testutils.DockerImageVersions.KAFKA;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -57,7 +56,7 @@ public class KafkaTransactionLogITCase extends TestLogger {
 
     @ClassRule
     public static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA).withEmbeddedZookeeper();
+            createKafkaContainer().withEmbeddedZookeeper();
 
     private final List<Producer<byte[], Integer>> openProducers = new ArrayList<>();
 

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterTestBase.java

@@ -55,7 +55,6 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.function.Consumer;
 
-import static org.apache.flink.connector.kafka.testutils.DockerImageVersions.KAFKA;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 
 /** Test base for KafkaWriter. */
@@ -73,7 +72,7 @@ public abstract class KafkaWriterTestBase {
     protected TriggerTimeService timeService;
 
     protected static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA)
+            createKafkaContainer()
                     .withEmbeddedZookeeper()
                     .withNetwork(NETWORK)
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java

@@ -51,22 +51,38 @@ private KafkaUtil() {}
      * This method helps to set commonly used Kafka configurations and aligns the internal Kafka log
      * levels with the ones used by the capturing logger.
      *
-     * @param dockerImageVersion describing the Kafka image
      * @return configured Kafka container
      */
-    public static KafkaContainer createKafkaContainer(String dockerImageVersion) {
-        return createKafkaContainer(dockerImageVersion, "Kafka");
+    public static KafkaContainer createKafkaContainer() {
+        return createKafkaContainer("Kafka");
     }
 
     /**
      * This method helps to set commonly used Kafka configurations and aligns the internal Kafka log
      * levels with the ones used by the capturing logger, and set the prefix of logger.
      */
-    public static KafkaContainer createKafkaContainer(
-            String dockerImageVersion, String containerName) {
+    public static KafkaContainer createKafkaContainer(String containerName) {
+        Logger logger = createLogger(containerName);
 
-        Logger logger = LoggerFactory.getLogger("container." + containerName);
+        String logLevel = inferLogLevel(logger);
 
+        Slf4jLogConsumer logConsumer = new Slf4jLogConsumer(logger, true);
+        return new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+                .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+                .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
+                .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
+                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
+                .withEnv("KAFKA_LOG4J_ROOT_LOGLEVEL", logLevel)
+                .withEnv("KAFKA_LOG4J_LOGGERS", "state.change.logger=" + logLevel)
+                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
+                .withEnv(
+                        "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
+                        String.valueOf(Duration.ofHours(2).toMillis()))
+                .withEnv("KAFKA_LOG4J_TOOLS_ROOT_LOGLEVEL", logLevel)
+                .withLogConsumer(logConsumer);
+    }
+
+    public static String inferLogLevel(Logger logger) {
         String logLevel;
         if (logger.isTraceEnabled()) {
             logLevel = "TRACE";
@@ -81,21 +97,11 @@ public static KafkaContainer createKafkaContainer(
         } else {
             logLevel = "OFF";
         }
+        return logLevel;
+    }
 
-        Slf4jLogConsumer logConsumer = new Slf4jLogConsumer(logger, true);
-        return new KafkaContainer(DockerImageName.parse(dockerImageVersion))
-                .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
-                .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
-                .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
-                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
-                .withEnv("KAFKA_LOG4J_ROOT_LOGLEVEL", logLevel)
-                .withEnv("KAFKA_LOG4J_LOGGERS", "state.change.logger=" + logLevel)
-                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
-                .withEnv(
-                        "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
-                        String.valueOf(Duration.ofHours(2).toMillis()))
-                .withEnv("KAFKA_LOG4J_TOOLS_ROOT_LOGLEVEL", logLevel)
-                .withLogConsumer(logConsumer);
+    public static Logger createLogger(String containerName) {
+        return LoggerFactory.getLogger("container." + containerName);
     }
 
     /**

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -434,7 +434,7 @@ private KafkaContainer createKafkaContainer(
             int brokerID, @Nullable GenericContainer<?> zookeeper) {
         String brokerName = String.format("Kafka-%d", brokerID);
         KafkaContainer broker =
-                KafkaUtil.createKafkaContainer(DockerImageVersions.KAFKA, brokerName)
+                KafkaUtil.createKafkaContainer(brokerName)
                         .withNetworkAliases(brokerName)
                         .withEnv("KAFKA_BROKER_ID", String.valueOf(brokerID))
                         .withEnv("KAFKA_MESSAGE_MAX_BYTES", String.valueOf(50 * 1024 * 1024))

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/internals/metrics/KafkaMetricMutableWrapperTest.java

@@ -42,7 +42,6 @@
 import java.util.function.Function;
 import java.util.stream.Stream;
 
-import static org.apache.flink.connector.kafka.testutils.DockerImageVersions.KAFKA;
 import static org.apache.flink.connector.kafka.testutils.KafkaUtil.createKafkaContainer;
 
 @Testcontainers
@@ -55,7 +54,7 @@ class KafkaMetricMutableWrapperTest {
 
     @Container
     public static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA)
+            createKafkaContainer()
                     .withEmbeddedZookeeper()
                     .withNetwork(NETWORK)
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);

flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java

@@ -19,7 +19,7 @@
 package org.apache.flink.streaming.connectors.kafka.table;
 
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.connector.kafka.testutils.DockerImageVersions;
+import org.apache.flink.connector.kafka.testutils.KafkaUtil;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.test.util.AbstractTestBase;
@@ -41,8 +41,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.KafkaContainer;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.utility.DockerImageName;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -65,15 +63,8 @@ public abstract class KafkaTableTestBase extends AbstractTestBase {
 
     @ClassRule
     public static final KafkaContainer KAFKA_CONTAINER =
-            new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA)) {
-                @Override
-                protected void doStart() {
-                    super.doStart();
-                    if (LOG.isInfoEnabled()) {
-                        this.followOutput(new Slf4jLogConsumer(LOG));
-                    }
-                }
-            }.withEmbeddedZookeeper()
+            KafkaUtil.createKafkaContainer()
+                    .withEmbeddedZookeeper()
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS)
                     .withEnv(
                             "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",