问在Flink流道上通过KafkaIO消耗事件时Apache管道失败
EN

Stack Overflow用户

提问于 2022-09-16 00:56:13

回答 1查看 66关注 0票数 0

我有一个包含几个阶段的波束管道，它通过一个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();
}

在直接运行程序上运行管道时，它不会引发任何错误。但是在我的例子中，我必须使用Flink 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

这个错误可以通过在上面的方法中注释行来解决，但是该方法，withMaxReadTime(...)不应该使用其他方法来测试/演示，就像文档一样。

管道实例化如下所示，

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

问题：

为什么会发生这个错误？
如何解决这个问题？

如果可能的话，请提供一些资源。

java

apache-flink

apache-beam

apache-beam-kafkaio

媒体处理1元起

智能、强大、全面的多媒体数据处理服务，助您提升媒体质量、降低成本，媒体处理套餐低至1元

回答 1

Stack Overflow用户

发布于 2022-09-16 12:45:22

错误似乎不是在Beam中，而是在Flink的闭包清洁器中，后者修改了用户或SDK代码的私有部分。这似乎是最近版本的Java和Flink的一个已知问题。请参阅Error message on example flink job: Unable to make field private final byte[] java.lang.String.value accessible

为什么注释行会改变事情？通常，当你从卡夫卡阅读时，你会毫无限制地阅读溪流。当您指定withMaxReadTime时，这将成为一个有界的读取。因此，对底层Flink运算符的转换是不同的。

票数 1

页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持

原文链接：

https://stackoverflow.com/questions/73742360

复制

Web应用程序测试：Web测试的8步指南

腾讯云测试服务网站数据库 sql 网络安全

本文翻译自：https://www.guru99.com/web-application-testing.html

小锋学长生活大爆炸

2020/08/13

2.6K0