https://flink.apache.org/ logo
Join Slack
Powered by
Hi, I am trying to upgrade scala from 2.11 to 2.12...
# troubleshooting
a
Hi, I am trying to upgrade scala from 2.11 to 2.12 and currently on flink 1.13.5. I am facing 
NPE
when kryo is trying to serialise 
PriorityQueue
. Stack trace in thread
Copy code
(dfac0af332dc) Caused by: com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException
(0d4c023239a5) 	at myclass.saveAdEventQueue(myclass.scala:252)
2023-04-28 07:45:36 [stdout] -  (2270ee21b641) 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor.processInput(StreamTwoInputProcessor.java:96)
 (dfac0af332dc) Serialization trace:
(0d4c023239a5) 	at myclass.processElement1(myclass.scala:114)
2023-04-28 07:45:36 [stdout] -  (e99cb36abe55) 	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
 (dfac0af332dc) f$2 (scala.math.Ordering$$anon$10)
(0d4c023239a5) 	at myclass.processElement1(myclass.scala:30)
2023-04-28 07:45:36 [stdout] -  (eb23772b6965) 	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
 (dfac0af332dc) $outer (scala.math.Ordering$$anon$4)
(0d4c023239a5) 	at org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator.processElement1(KeyedCoProcessOperator.java:78)
 (dfac0af332dc) ord (scala.collection.mutable.PriorityQueue)
2023-04-28 07:45:36 [stdout] -  (bccfc625df95) 	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.processRecord1(StreamTwoInputProcessorFactory.java:213)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:82)
2023-04-28 07:45:36 [stdout] -  (d5dddeacb336) 	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.lambda$create$0(StreamTwoInputProcessorFactory.java:178)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:291)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
2023-04-28 07:45:36 [stdout] -  (2c8b5c9efd77) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
2023-04-28 07:45:36 [stdout] -  (2270ee21b641) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
2023-04-28 07:45:36 [stdout] -  (e99cb36abe55) 	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor.processInput(StreamTwoInputProcessor.java:96)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
2023-04-28 07:45:36 [stdout] -  (eb23772b6965) 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
2023-04-28 07:45:36 [stdout] -  (bccfc625df95) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684)
 (dfac0af332dc) 	at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
2023-04-28 07:45:36 [stdout] -  (d5dddeacb336) 	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
 (dfac0af332dc) 	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:316)
(0d4c023239a5) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623)
Copy code
Caused by: com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException
 (0d4c023239a5) 	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:316)
2023-04-28 07:45:36 [stdout]  (9fd86b86024c) 	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
2023-04-28 07:45:36 [stdout] - 10-69-116-54-uswest2aprod (bccfc625df95) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
 (0d4c023239a5) 	at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeValueInternal(AbstractRocksDBState.java:158)
2023-04-28 07:45:36 [stdout]  (9fd86b86024c) 	at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor.processInput(StreamTwoInputProcessor.java:96)
 (0d4c023239a5) 	at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeValue(AbstractRocksDBState.java:180)
2023-04-28 07:45:36 [stdout]  (9fd86b86024c) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423)
2023-04-28 07:45:36 [stdout] - 10-69-116-54-uswest2aprod (d5dddeacb336) 	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
 (0d4c023239a5) 	at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeValue(AbstractRocksDBState.java:168)
2023-04-28 07:45:36 [stdout]  (9fd86b86024c) 	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
 (0d4c023239a5) 	at org.apache.flink.contrib.streaming.state.RocksDBValueState.update(RocksDBValueState.java:107)
2023-04-28 07:45:36 [stdout] - 10-69-116-54-uswest2aprod (2c8b5c9efd77) 	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
2023-04-28 07:45:36 [stdout]  (9fd86b86024c) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684)
 (0d4c023239a5) 	... 20 more
2023-04-28 07:45:36 [stdout] - 10-69-116-54-uswest2aprod (2270ee21b641) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639)
2023-04-28 07:45:36 [stdout]  (9fd86b86024c) 	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639)
m
I don’t think you can upgrade from a Scala 2.11 to 2.12 application. That’s a binary incompatible change
a
Skipped the state, so that it starts afresh, but still facing the issue
Hi, can someone please help me resolve this ?
m
It looks like you're trying to store state on a key that doesn't exist, resulting in a NPE. You would need to debug your incoming data and the computations that you're applying on that.
a
Thank you for your response. Here is the PQ I am trying to store, logging it before exception is thrown. Also, why would the key wont exist just from scala upgrade, with no computations changed
Copy code
PriorityQueue(AdEventInfoWithTimestamp(AdEventOpportunityGroup(List(AdEventInfo(Impression:2023-05-10T11:59:43.042Z,8737da5d3dd07209,above_search,0,AdEventComputedInfo(None,true,List())), AdEventInfo(Impression:2023-05-10T11:59:43.042Z,8737da5d3dd07209,above_search,1,AdEventComputedInfo(None,true,List())), AdEventInfo(Impression:2023-05-10T11:59:43.042Z,8737da5d3dd07209,below_search,0,AdEventComputedInfo(Some(AdClickInfo(Click:2023-05-10T12:16:49.086Z,8737da5d3dd07209,below_search,0,AdClickComputedInfo(false,List(No click on client device),false))),true,List()))),y_FB18669D-AE4C-4D1E-8531-F7E65EFD6019,None),1683724609086))
Also, when tried to use java PQ, got the following error:
Copy code
Caused by: java.lang.NullPointerException
	at com.esotericsoftware.kryo.util.DefaultClassResolver.writeClass(DefaultClassResolver.java:80)
	at com.esotericsoftware.kryo.Kryo.writeClass(Kryo.java:488)
	at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:593)
	at org.apache.flink.runtime.types.PriorityQueueSerializer.write(PriorityQueueSerializer.java:65)
	at org.apache.flink.runtime.types.PriorityQueueSerializer.write(PriorityQueueSerializer.java:40)
	at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.serialize(KryoSerializer.java:316)
	at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeValueInternal(AbstractRocksDBState.java:158)
	at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeValue(AbstractRocksDBState.java:180)
	at org.apache.flink.contrib.streaming.state.AbstractRocksDBState.serializeValue(AbstractRocksDBState.java:168)
	at org.apache.flink.contrib.streaming.state.RocksDBValueState.update(RocksDBValueState.java:107)
	... 20 more
m
Also, why would the key wont exist just from scala upgrade, with no computations changed
Because the Scala binary compatibility changed from 2.11 to 2.12: that's a breaking change from a Scala perspective. There's no upgrade path for that
a
But if I am skipping the state, so that it starts afresh, that means there is no state
m
Yes because Scala is a breaking upgrade. There is nothing Flink can do is this matter
s
I'm seeing same issue from my end🤔 I'll try to change scala 2.12 back to 2.11
Open in Slack
PreviousNext
Powered by