[FLINK-36278] Reduce log size by avoiding container logs by default

Currently, container logs appear under an o.a.f logger and thus are visible on CI. This results in compressed log size >40MB for a run and often leads to download errors.

This PR reroutes container logs to a special container logger. It also uses a custom format to significantly reduce the size of each log line. The logs for containers are disabled by default.

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;
@@ -40,11 +41,8 @@
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.rules.Timeout;
-import org.slf4j.Logger;
-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;
@@ -60,9 +58,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";
     private static final Path sqlAvroJar = ResourceTestUtils.getResource(".*avro.jar");
@@ -78,10 +73,9 @@ public class SQLClientSchemaRegistryITCase {
 
     @ClassRule
     public static final KafkaContainer KAFKA =
-            new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
+            KafkaUtil.createKafkaContainer(SQLClientSchemaRegistryITCase.class)
                     .withNetwork(NETWORK)
-                    .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS)
-                    .withLogConsumer(LOG_CONSUMER);
+                    .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
 
     @ClassRule
     public static final SchemaRegistryContainer REGISTRY =
@@ -92,7 +86,11 @@ public class SQLClientSchemaRegistryITCase {
                     .dependsOn(KAFKA);
 
     public final TestcontainersSettings testcontainersSettings =
-            TestcontainersSettings.builder().network(NETWORK).logger(LOG).dependsOn(KAFKA).build();
+            TestcontainersSettings.builder()
+                    .network(NETWORK)
+                    .logger(KafkaUtil.getLogger("flink", SQLClientSchemaRegistryITCase.class))
+                    .dependsOn(KAFKA)
+                    .build();
 
     public final FlinkContainers flink =
             FlinkContainers.builder().withTestcontainersSettings(testcontainersSettings).build();

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

@@ -45,8 +45,6 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.extension.RegisterExtension;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.junit.jupiter.Container;
@@ -62,7 +60,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;
 
@@ -71,20 +68,22 @@
 @Testcontainers
 class SmokeKafkaITCase {
 
-    private static final Logger LOG = LoggerFactory.getLogger(SmokeKafkaITCase.class);
     private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
     private static final Network NETWORK = Network.newNetwork();
     private static final String EXAMPLE_JAR_MATCHER = "flink-streaming-kafka-test.*";
 
     @Container
     public static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA, LOG)
+            createKafkaContainer(SmokeKafkaITCase.class)
                     .withEmbeddedZookeeper()
                     .withNetwork(NETWORK)
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
 
     public static final TestcontainersSettings TESTCONTAINERS_SETTINGS =
-            TestcontainersSettings.builder().logger(LOG).dependsOn(KAFKA_CONTAINER).build();
+            TestcontainersSettings.builder()
+                    .logger(KafkaUtil.getLogger("flink", SmokeKafkaITCase.class))
+                    .dependsOn(KAFKA_CONTAINER)
+                    .build();
 
     @RegisterExtension
     public static final FlinkContainers FLINK =

flink-connector-kafka-e2e-tests/flink-end-to-end-tests-common-kafka/src/test/resources/log4j2-test.properties

@@ -32,3 +32,27 @@ appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
 #logger.yarn.name = org.testcontainers.shaded.com.github.dockerjava.core
 #logger.yarn.level = WARN
 #logger.yarn.appenderRef.console.ref = TestLogger
+
+# Logger configuration for containers, by default this is off
+# If you want to investigate test failures, overwrite the level as above
+logger.container.name = container
+logger.container.level = OFF
+logger.container.additivity = false  # This prevents messages from being logged by the root logger
+logger.container.appenderRef.containerappender.ref = ContainerLogger
+
+logger.kafkacontainer.name = container.kafka
+logger.kafkacontainer.level = OFF
+
+logger.flinkcontainer.name = container.flink
+logger.flinkcontainer.level = OFF
+
+logger.flinkenv.name = org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment
+logger.flinkenv.level = OFF
+logger.flinkenv.additivity = false  # This prevents messages from being logged by the root logger
+logger.flinkenv.appenderRef.containerappender.ref = ContainerLogger
+
+appender.containerappender.name = ContainerLogger
+appender.containerappender.type = CONSOLE
+appender.containerappender.target = SYSTEM_ERR
+appender.containerappender.layout.type = PatternLayout
+appender.containerappender.layout.pattern = [%c{1}] %m%n

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

@@ -34,8 +34,6 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
@@ -47,7 +45,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;
@@ -56,12 +53,9 @@
 @ExtendWith(TestLoggerExtension.class)
 class FlinkKafkaInternalProducerITCase {
 
-    private static final Logger LOG =
-            LoggerFactory.getLogger(FlinkKafkaInternalProducerITCase.class);
-
     @Container
     private static final KafkaContainer KAFKA_CONTAINER =
-            createKafkaContainer(KAFKA, LOG).withEmbeddedZookeeper();
+            createKafkaContainer(FlinkKafkaInternalProducerITCase.class).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, LOG)
+            createKafkaContainer(KafkaSinkITCase.class)
                     .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, LOG).withEmbeddedZookeeper();
+            createKafkaContainer(KafkaTransactionLogITCase.class).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, LOG)
+            createKafkaContainer(KafkaWriterTestBase.class)
                     .withEmbeddedZookeeper()
                     .withNetwork(NETWORK)
                     .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);

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

@@ -18,8 +18,6 @@
 
 package org.apache.flink.connector.kafka.testutils;
 
-import org.apache.flink.util.StringUtils;
-
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -49,24 +47,34 @@ public class KafkaUtil {
 
     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
-     * @param logger to derive the log level from
-     * @return configured Kafka container
-     */
-    public static KafkaContainer createKafkaContainer(String dockerImageVersion, Logger logger) {
-        return createKafkaContainer(dockerImageVersion, logger, null);
+    /** This method helps to set commonly used Kafka configurations and sets up the logger. */
+    public static KafkaContainer createKafkaContainer(Class<?> testCase) {
+        return createKafkaContainer(getContainerName("kafka", testCase));
     }
 
-    /**
-     * 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, Logger logger, String loggerPrefix) {
+    /** This method helps to set commonly used Kafka configurations and sets up the logger. */
+    public static KafkaContainer createKafkaContainer(String containerName) {
+        Logger logger = getLogger(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);
+    }
+
+    private static String inferLogLevel(Logger logger) {
         String logLevel;
         if (logger.isTraceEnabled()) {
             logLevel = "TRACE";
@@ -81,24 +89,19 @@ public static KafkaContainer createKafkaContainer(
         } else {
             logLevel = "OFF";
         }
+        return logLevel;
+    }
 
-        Slf4jLogConsumer logConsumer = new Slf4jLogConsumer(logger);
-        if (!StringUtils.isNullOrWhitespaceOnly(loggerPrefix)) {
-            logConsumer.withPrefix(loggerPrefix);
-        }
-        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 getLogger(String containerName) {
+        return LoggerFactory.getLogger("container." + containerName);
+    }
+
+    public static Logger getLogger(String type, Class<?> testClass) {
+        return getLogger(getContainerName(type, testClass));
+    }
+
+    private static String getContainerName(String type, Class<?> testClass) {
+        return type + "." + testClass.getSimpleName();
     }
 
     /**

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

@@ -54,7 +54,6 @@ protected Properties createProperties() {
         Properties properties = new Properties();
         properties.putAll(standardProps);
         properties.putAll(secureProps);
-        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-client-id");
         properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "producer-transaction-id");
         properties.put(FlinkKafkaProducer.KEY_DISABLE_METRICS, "true");
         return properties;

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, LOG, brokerName)
+                KafkaUtil.createKafkaContainer(brokerName)
                         .withNetworkAliases(brokerName)
                         .withEnv("KAFKA_BROKER_ID", String.valueOf(brokerID))
                         .withEnv("KAFKA_MESSAGE_MAX_BYTES", String.valueOf(50 * 1024 * 1024))