https://flink.apache.org/ logo
Join Slack
Powered by
Hi Im running very simple Flink Job on Flink 1.17 ...
# troubleshooting
k
Hi Im running very simple Flink Job on Flink 1.17 and I see this in logs:
Kryo serializer scala extensions are not available
I'm not using Scala in my sandbox job, pure java. What should I do to turn this off? the same thing was reported by other user on April 21st but unfortunately got no response https://apache-flink.slack.com/archives/C03G7LJTS2G/p1682093159771169
d
What exactly is the concern there? Is it just confusing? I guess we can make the message more explicit, something along the lines of 
"Falling back to default Kryo serializer, because flink-scala is not present on the classpath."
WDYT?
k
Hi, the concern is that my logs were "flooded" with this line. For every message ive processed. My sandbox pipeline was using Kryo serialization but without any scala classes.
I dont mind the "text" in the log, only that it was printed for every message
I wasnt sure if this is somethign i should be worry about
d
that sounds weird, I’d expect this to be done once 🤔 it shouldn’t really be on the per-element code path how are you using the serializer in your pipeline?
if you switch to DEBUG log level, it should also give you a stacktrace of where this is called from, can you please share it?
k
how are you using the serializer in your pipeline?
Nothings special I just have a pipeline Source -> process - > Sink Custom source, and custom sink (sink prints events in logs) Im using two custom data object, Pojo Like, but they dont have getters/setters Source is producins Object A, process is cosuming those and creates ObjectB, Sink consumes ObjectB and prints it in logs
and checkpoitns are set
d
🤔 Can you please try to get that stacktrace? I’m still bit blind here :(
👍 1
c
The message should be at most logged once per task. (Basically whenever a Kryo serializer is constructed)
🤔 1
k
hm ok, maybe this is what in fact happen. I will have to double check.
It is logged 10 times after every checkpoint My job:
Copy code
StreamExecutionEnvironment env = createStreamEnv();
env.addSource(new CheckpointCountingSource<>(5, 5, eventProducer))
 .setParallelism(1)
 .process(new FlinkBusinessLogic())
 .addSink(sink);
Copy code
12:08:34,088 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask          [] - Notify checkpoint 1 complete on task Process -> Sink: Unnamed (16/16)#0
12:08:34,088 DEBUG org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl [] - Notification of checkpoint COMPLETE 1 for task Process -> Sink: Unnamed (16/16)#0
12:08:34,088 DEBUG org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl [] - Notification of checkpoint SUBSUME -1 for task Process -> Sink: Unnamed (16/16)#0
12:08:34,138 INFO  org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer [] - Kryo serializer scala extensions are not available.
java.lang.ClassNotFoundException: org.apache.flink.runtime.types.FlinkScalaKryoInstantiator
	at jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) ~[?:?]
	at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) ~[?:?]
	at java.lang.ClassLoader.loadClass(ClassLoader.java:521) ~[?:?]
	at java.lang.Class.forName0(Native Method) ~[?:?]
	at java.lang.Class.forName(Class.java:315) ~[?:?]
	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.getKryoInstance(KryoSerializer.java:484) ~[flink-core-1.17.0.jar:1.17.0]
	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.checkKryoInitialized(KryoSerializer.java:519) ~[flink-core-1.17.0.jar:1.17.0]
	at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:391) ~[flink-core-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:190) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:43) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:53) ~[flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:337) ~[flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNonSpanningRecord(SpillingAdaptiveSpanningRecordDeserializer.java:128) ~[flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:103) ~[flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:93) ~[flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:100) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788) ~[flink-streaming-java-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) [flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) [flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) [flink-runtime-1.17.0.jar:1.17.0]
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) [flink-runtime-1.17.0.jar:1.17.0]
	at java.lang.Thread.run(Thread.java:834) [?:?]
12:08:34,138 INFO  org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer [] - Kryo serializer scala extensions are not available.
java.lang.ClassNotFoundException: org.apache.flink.runtime.types.FlinkScalaKryoInstantiator
	at jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) ~[?:?]
	at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) ~[?:?]
	at java.lang.ClassLoader.loadClass(ClassLoader.java:521) ~[?:?]
	at java.lang.Class.forName0(Native Method) ~[?:?]
Im runnign this in IDE, when I add
Copy code
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-scala_2.12</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>
the exception/log line is gone
d
are you using hashmap state backend?
k
yep, in-memory one. Its just a sandbox pipeline
d
org.apache.flink.runtime.state.heap.CopyOnWriteStateTable#stateSnapshot
seems to duplicate serializers with every checkpoint, that might be related
c
Yes that would fit since the 
kryo
instance in the Kryo serializer is transient and won't be copied
d
I’m wondering whether there’s any value in logging this at first place (on INFO level, it’s reasonable on DEBUG imo)
c
On the client-side its very valuable if you're using Scala types.
server-side, not so much
not sure if theres a good way to differentiate between the two
k
if you're using Scala types.
FYI @Chesnay Schepler I was not using any scala types in my Flink Job.
c
Well we can't know that on the client-side if the scala extensions aren't on the classpath 🤷
👍 1
k
I was rather thinking whether it is possible to know if Scala is present in submitted job.
c
I don't think so; not reliably anyway.
👍 1
k
anyways it was "fixed" when I added
Copy code
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-scala_2.12</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>
And i believe that jar is in Flink's lib folder, so I dont think I had this issue when running my job on a cluster, Only in IDE
thaks for clarification
c
TBH if all you want is for this message to disappear I'd configure a log4j filter instead.
Can you file a ticket? This shouldn't be difficult to fix on the Flink side after all.
k
sure I know the ways to hide the log message. I was just worried that maybe there is something wrong since this was "flooding" the logs.
c
I see. If you don't use Scala you can safely ignore this.
3 Views
Open in Slack
PreviousNext
Powered by