使用Flink Runner运行KafkaIO消费的Beam管道失败问题排查

问题场景 #

我有一个多阶段的Apache Beam管道，通过KafkaIO消费数据，核心代码如下：

pipeline.apply("Read Data from Stream", StreamReader.read())
        .apply("Decode event and extract relevant fields", ParDo.of(new DecodeExtractFields()))
        .apply(...);

StreamReader.read()方法实现：

public static KafkaIO.Read<String, String> read() {
    return KafkaIO.<String, String>read()
            .withBootstrapServers(Constants.BOOTSTRAP_SERVER)
            .withTopics(Constants.KAFKA_TOPICS)
            .withConsumerConfigUpdates(Constants.CONSUMER_PROPERTIES)
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(StringDeserializer.class)
  //Line-A  .withMaxReadTime(Duration.standardDays(10))
            .withLogAppendTime();
}

管道实例化代码：

PipelineOptions pipelineOptions = PipelineOptionsFactory.create();
pipelineOptions.setRunner(FlinkRunner.class);
Pipeline pLine = Pipeline.create(pipelineOptions);

错误现象 #

该管道在Direct Runner上运行无报错，但切换为Flink Runner时抛出如下错误：

Exception in thread "main" java.lang.RuntimeException: Error while translating UnboundedSource: org.apache.beam.sdk.io.kafka.KafkaUnboundedSource@14b31e37
    at org.apache.beam.runners.flink.FlinkStreamingTransformTranslators$UnboundedReadSourceTranslator.translateNode(FlinkStreamingTransformTranslators.java:250)
    at org.apache.beam.runners.flink.FlinkStreamingTransformTranslators$ReadSourceTranslator.translateNode(FlinkStreamingTransformTranslators.java:336)
    at org.apache.beam.runners.flink.FlinkStreamingPipelineTranslator.applyStreamingTransform(FlinkStreamingPipelineTranslator.java:161)
....
    at Main.main(Main.java:6)
Caused by: java.lang.reflect.InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not "opens java.lang" to unnamed module @2c34f934
    at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:354)
    at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:297)
    at java.base/java.lang.reflect.Field.checkCanSetAccessible(Field.java:178)
    at java.base/java.lang.reflect.Field.setAccessible(Field.java:172)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:106)
Caused by: java.lang.reflect.InaccessibleObjectException: Unable to make field private final byte[] java.lang.String.value accessible: module java.base does not "opens java.lang" to unnamed module @2c34f934

取消注释StreamReader.read()方法中的Line-A（.withMaxReadTime(...)）可解决错误，但该方法仅用于测试/演示场景，不符合生产规范。

问题 #

1. 该错误产生的原因是什么？
2. 如何在不使用withMaxReadTime()的前提下解决该问题？

1. 错误产生的原因 #

这个错误是Java模块系统（JPMS）权限限制与Flink ClosureCleaner兼容性问题共同导致的：

• 未设置withMaxReadTime时，Kafka源为无界源，Flink Runner翻译Beam的UnboundedSource时，会调用Flink的ClosureCleaner做序列化清理。
• ClosureCleaner会尝试通过反射访问java.lang.String的私有字段value，但Java 9+的模块系统中，java.base模块默认不向未命名模块开放java.lang包的反射权限，因此抛出InaccessibleObjectException。
• 设置withMaxReadTime后，Kafka源转为有界源，Flink Runner的处理逻辑改变，不会触发该闭包清理的反射操作，因此错误消失。

2. 不使用withMaxReadTime()的解决方案 #

方式一：添加JVM参数开放模块访问权限 #

启动Beam应用时，添加以下JVM参数，允许未命名模块反射访问java.lang包：

--add-opens java.base/java.lang=ALL-UNNAMED

这是直接绕过权限限制的临时方案，能让Flink的ClosureCleaner正常工作。

方式二：升级Beam和Flink版本 #

该兼容性问题在较新版本中已被修复：

• Beam 2.30.0+ 优化了Flink Runner的闭包处理逻辑，避免触发非法反射。
• Flink 1.14+ 改进了ClosureCleaner实现，减少对JDK内部私有字段的依赖。
  建议升级到匹配的版本（如Beam 2.40.0 + Flink 1.17.x），从根源解决问题。

方式三：显式设置水位线策略 #

在KafkaIO配置中显式指定水位线策略，可能改变源的处理逻辑，间接避免触发错误：

return KafkaIO.<String, String>read()
        // 其他原有配置...
        .withLogAppendTime()
        .withWatermarkStrategy(WatermarkStrategy.forMonotonousTimestamps());

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