Flink application ClassCastException

Question

I have a flink application which reads from kafka and sinks it to kafka.

Application runs with no issues when i run it from Intellij IDEA, but gives ClassCastException when I submit the shadowJar to flink cluster. Can I get some help in figuring out what Im doing wrong here ?

Exception trace :

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)

Code used :

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");

Build file : 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

The root cause is Flink loads dependencies dynamically from Java Classpath and also UserCode Classpath. Some classes can be loaded by different class loaders then their type is assigned to each other. You can simply fix this by:

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

More details in docs