You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fapaul (via GitHub)" <gi...@apache.org> on 2024/09/19 15:17:36 UTC
Re: [PR] [FLINK-36278] Reduce log size [flink-connector-kafka]
fapaul commented on code in PR #119:
URL: https://github.com/apache/flink-connector-kafka/pull/119#discussion_r1767018924
##########
flink-connector-kafka/src/test/resources/log4j2-test.properties:
##########
@@ -27,15 +27,41 @@ appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
-logger.kafka.name = kafka
+# Overwrite the level for all Flink related loggers
+logger.flink.name = org.apache.flink
+logger.flink.level = OFF # WARN for starting debugging
+logger.flinkconnector.name = org.apache.flink.connector
+logger.flinkconnector.level = OFF # INFO/DEBUG for starting debugging
+
+# Kafka producer and consumer level
+logger.kafka.name = org.apache.kafka
logger.kafka.level = OFF
-logger.kafka2.name = state.change
+
+# For old school kafka test that spawn the test server in the same JVM
+logger.kafka2.name = kafka
Review Comment:
Which tests still spawn kafka in the same JVM?
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java:
##########
@@ -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) {
Review Comment:
Afaict this method is always called with `"flink"` as a first parameter. WDY about removing the type parameter and always set it to `flink` when called with a testClass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org