Flink应用程序ClassCastException

Question

我有一个flink应用程序，它从kafka中读取并将其吸收到kafka。

当我从Intellij运行应用程序时，没有问题，但是当我将ClassCastException提交给flink集群时，它会给出shadowJar。我能在找出我在这里做错了什么方面得到一些帮助吗？

异常跟踪：

Caused by: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.clients.consumer.OffsetResetStrategy to field org.apache.flink.connector.kafka.source.enumerator.initializer.ReaderHandledOffsetsInitializer.offsetResetStrategy of type org.apache.kafka.clients.consumer.OffsetResetStrategy in instance of org.apache.flink.connector.kafka.source.enumerator.initializer.ReaderHandledOffsetsInitializer
    at java.base/java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2205)
    at java.base/java.io.ObjectStreamClass$FieldReflector.checkObjectFieldValueTypes(ObjectStreamClass.java:2168)
    at java.base/java.io.ObjectStreamClass.checkObjFieldValueTypes(ObjectStreamClass.java:1422)
    at java.base/java.io.ObjectInputStream.defaultCheckFieldValues(ObjectInputStream.java:2517)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2424)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2233)
    at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1692)
    at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2501)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2395)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2233)

所用代码：

KafkaSource<String> source = KafkaSource.<String>builder()
                    .setBootstrapServers("localhost:9092")
                    .setTopics("topic")
                    .setGroupId("grp")
                    .setStartingOffsets(OffsetsInitializer.earliest())
                    .setValueOnlyDeserializer(new SimpleStringSchema())
                    .build();


            DataStream<String> eventStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
                    .name("event-stream").sinkTo("kafka");

构建文件: flinkVersion = 1.15.0

 //flinkShadowJar "org.apache.flink:flink-connector-kafka:${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java
    implementation group: 'org.apache.flink', name: 'flink-streaming-java', version: "${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.flink/flink-java
    implementation group: 'org.apache.flink', name: 'flink-java', version: "${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.flink/flink-core
    implementation group: 'org.apache.flink', name: 'flink-core', version: "${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.flink/flink-clients
    implementation group: 'org.apache.flink', name: 'flink-clients', version: "${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.kafka/kafka
   // flinkShadowJar group: 'org.apache.kafka', name: 'kafka_2.12', version: "${kafkaVersion}"

    flinkShadowJar "org.apache.avro:avro:1.11.0"
    flinkShadowJar group: 'org.apache.flink', name: 'flink-avro-confluent-registry', version: "${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka
    flinkShadowJar group: 'org.apache.flink', name: 'flink-connector-kafka', version: "${flinkVersion}"
  //  flinkShadowJar group: 'org.apache.flink', name: 'flink-connector-base', version: "${flinkVersion}"

    // https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-core
    implementation group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.17.2'

    // https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-slf4j-impl
    implementation group: 'org.apache.logging.log4j', name: 'log4j-slf4j-impl', version: '2.17.2'

    // https://mvnrepository.com/artifact/org.apache.logging.log4j/log4j-api
    implementation group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.17.2'

Answer 1

根本原因是Flink从Java和UserCode Classpath动态加载依赖项。有些类可以由不同的类加载器加载，然后将它们的类型分配给彼此。您可以简单地通过以下方法来解决这个问题：

$ echo 'classloader.resolve-order: parent-first' >> /opt/flink/conf/flink-conf.yaml

关于文档的更多细节