anyone have any clues about this deserialisation p...
# troubleshootingf
Felix Angell
08/25/2023, 3:22 PManyone have any clues about this deserialisation problem in flink? i've omitted some details to make the logs more concise:
Copy code
2023-08-25 16:14:17
java.io.IOException: Failed to deserialize consumer record due to
Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
...
Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
...
Caused by: com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException
...
Serialization trace:
links_ (xxxxxxxxxxx.messages.common.CommonOuter$MessageMetadata)
Caused by: java.lang.UnsupportedOperationException
at java.base/java.util.Collections$UnmodifiableCollection.add(Collections.java:1060)
at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)
at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
... 35 more
...Felix Angell
08/25/2023, 3:23 PMlots of link on SO seem to mention i need a kyro dependency or something or other but this is typically for avro based Serde which isn't relevant in this case
Felix Angell
08/25/2023, 3:51 PMlooks like i have to specify a custom serializer?
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/third_party_serializers/
Felix Angell
08/25/2023, 3:52 PMi am trying to pass the GeneratedMessageV3 around in operators which i think is the problem. i have a map function to take byte[] -> GeneratedMessageV3 (the child of GeneratedMessageV3 protobuf msg not literally GeneratedMessageV3)
Felix Angell
08/25/2023, 3:58 PMso i either need to configure some kind of custom serializer for this? or potentially pass around everything as bytes π¬
a
Ari Huttunen
08/25/2023, 5:43 PMAri Huttunen
08/25/2023, 5:45 PMIntegrating Kafka and Flink should become easier because https://www.confluent.io/blog/cloud-kafka-meets-cloud-flink-with-confluent-and-immerok/
f
Felix Angell
08/26/2023, 1:52 PMwill give this a look! cheers Ari
π 1
Felix Angell
08/26/2023, 2:04 PMnot too sure where to go from here π€ i think the problem seems to be passing around grpc/protobuf messages to operators and the serialisation is incompatible with Kyro
Felix Angell
08/26/2023, 2:39 PMmy deserializer is super simple for reading kafka records.
KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:56)looks like this is invoked on reading a kafka record i.e. on my KafkaSource? i have something that basically passes the message as byte[]
Felix Angell
08/26/2023, 2:40 PM.setValueOnlyDeserializer(new KafkaBytesDeserializer())implements DeserializationSchema<byte[]>Felix Angell
08/26/2023, 3:24 PMhttps://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/third_party_serializers/
so i have a Map function which takes
byte[] -> MyProtobufMessageFelix Angell
08/26/2023, 3:24 PMi've added a kyro serializer for protobuf using twitters' 'chill-protobuf' dep. no dice though so far.
2 Views