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'