使用Beam 2.45.0的Flink Runner调用KafkaIO失败求助

问题场景 #

使用Beam 2.45.0版本的Flink Runner调用KafkaIO读取数据时，抛出以下错误：

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: No translator known for org.apache.beam.runners.core.construction.SplittableParDo$PrimitiveUnboundedRead
    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:841)
    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:240)
    at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1085)
    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1163)
    at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1163)
Caused by: java.lang.IllegalStateException: No translator known for org.apache.beam.runners.core.construction.SplittableParDo$PrimitiveUnboundedRead
    at org.apache.beam.runners.core.construction.PTransformTranslation.urnForTransform(PTransformTranslation.java:283)
    at org.apache.beam.runners.flink.FlinkStreamingPipelineTranslator.visitPrimitiveTransform(FlinkStreamingPipelineTranslator.java:135)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:593)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:585)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:585)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:585)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$500(TransformHierarchy.java:240)
    at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:214)
    at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:469)
    at org.apache.beam.runners.flink.FlinkPipelineTranslator.translate(FlinkPipelineTranslator.java:38)
    at org.apache.beam.runners.flink.FlinkStreamingPipelineTranslator.translate(FlinkStreamingPipelineTranslator.java:92)
    at org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.translate(FlinkPipelineExecutionEnvironment.java:115)
    at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:104)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:323)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:309)
    at BeamPipelineKafka.main(BeamPipelineKafka.java:51)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
    ... 8 more

当前使用的KafkaIO读取代码：

PipelineOptions options = PipelineOptionsFactory.create();

options.setRunner(FlinkRunner.class);

Pipeline pipeline = Pipeline.create(options);

pipeline
        // Read from the input Kafka topic
        .apply("Read from Kafka", KafkaIO.<String, String>read()
                .withBootstrapServers("localhost:9092")
                .withTopic("input-topic")
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializer(StringDeserializer.class))
                .apply ....

疑问：是否遗漏了必要配置？或者有没有办法禁用SplittableParDo？

问题原因 #

该错误源于Beam的KafkaIO默认启用了SplittableParDo（可拆分并行读取）特性，但当前版本的Flink Runner未提供对应操作的转换器，无法识别PrimitiveUnboundedRead类型的转换。

解决办法 #

1. 直接禁用SplittableParDo #

在KafkaIO的配置链中添加disableSplitting()方法，强制使用非拆分的读取模式，绕过SplittableParDo的兼容性问题：

pipeline
        // Read from the input Kafka topic
        .apply("Read from Kafka", KafkaIO.<String, String>read()
                .withBootstrapServers("localhost:9092")
                .withTopic("input-topic")
                .withKeyDeserializer(StringDeserializer.class)
                .withValueDeserializer(StringDeserializer.class)
                .disableSplitting()) // 禁用可拆分读取，避免SplittableParDo问题
        .apply ....

2. 检查并补全依赖 #

确保项目依赖中包含完整且版本统一的Beam组件：

• 引入Flink Runner核心依赖（Beam 2.45.0对应Flink 1.17.x，需匹配你的Flink集群版本）：

  <!-- Maven依赖示例 -->
  <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-runners-flink-1.17</artifactId>
      <version>2.45.0</version>
      <scope>runtime</scope>
  </dependency>
  

• 确保KafkaIO依赖版本与Beam版本一致：

  <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-io-kafka</artifactId>
      <version>2.45.0</version>
  </dependency>
  

可通过Maven的dependency:tree命令排查版本不一致的依赖并排除，避免依赖冲突。

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