如何在Apache Beam中关闭Direct Runner的Kafka Consumer日志？

问题背景 #

此前遇到过类似的日志屏蔽问题，但针对的是Dataflow日志而非Direct日志。希望关闭KafkaIO读取（消费者）的大量日志，已尝试通过设置SDK Harness的日志级别来实现，但未成功，代码如下：

var kafkasLogs =
        SdkHarnessOptions.SdkHarnessLogLevelOverrides.from(
            new HashMap<>(
                Map.of(
                    "org.apache.kafka.clients.consumer.internals.SubscriptionState",
                    SdkHarnessOptions.LogLevel.ERROR.name())));
    options.setSdkHarnessLogLevelOverrides(kafkasLogs); // extends sdkharnessoptions

尝试过该代码的多种变体，均未能屏蔽消费者日志，现寻求在不影响管道其他日志的前提下关闭这些日志的方法。

解决方案 #

SDK Harness日志级别配置是Dataflow runner专属的，Direct模式下并不生效，需要针对Direct runner的日志体系调整：

方法1：通过JUL（Java Util Logging）配置文件调整 #

Direct runner默认使用JUL，可创建logging.properties文件放在类路径下，添加以下配置：

# 保留其他日志默认级别，仅调整Kafka消费者相关包的级别
org.apache.kafka.clients.consumer.level = SEVERE
org.apache.kafka.clients.consumer.internals.level = SEVERE

启动管道时指定该配置文件：

java -Djava.util.logging.config.file=logging.properties -jar your-pipeline.jar

方法2：在代码中动态设置JUL日志级别 #

若不想用配置文件，可在管道初始化最开始添加代码，直接调整Kafka相关Logger的级别：

import java.util.logging.Logger;
import java.util.logging.Level;

// 需在KafkaIO客户端初始化前执行
Logger kafkaConsumerLogger = Logger.getLogger("org.apache.kafka.clients.consumer");
kafkaConsumerLogger.setLevel(Level.SEVERE);
Logger kafkaConsumerInternalsLogger = Logger.getLogger("org.apache.kafka.clients.consumer.internals");
kafkaConsumerInternalsLogger.setLevel(Level.SEVERE);

// 后续初始化Beam管道
PipelineOptions options = PipelineOptionsFactory.create();
// ... 其他配置逻辑

方法3：单独针对特定日志类调整 #

如果SubscriptionState这类特定类日志仍然过多，可单独设置其级别：

Logger.getLogger("org.apache.kafka.clients.consumer.internals.SubscriptionState").setLevel(Level.SEVERE);

注意事项 #

• 日志级别设置逻辑需早于KafkaIO客户端初始化，否则可能不生效
• Direct runner与Dataflow的日志体系不同，不要混用Dataflow专属的SDK Harness配置
• 建议保留SEVERE级别日志，避免完全关闭导致排查问题时无有效信息

内容的提问来源于stack exchange，提问作者Lemon