Hi Dan, Could you provide the code snippet such that we can reproduce the bug here?
Dan Hill <quietgol...@gmail.com> 于2021年12月20日周一 07:18写道： > Hi. > > I was curious if anyone else has hit this exception. I'm using the > IntervalJoinOperator to two streams of protos. I registered the protos > with a kryo serializer. I started hitting this issue which looks like the > operator is trying to deserialize a bad set of bytes that it serialized. > I'm not doing anything weird or custom with the code. It's a pretty simple > interval join. > > Has anyone hit this before? How have people solved this? I skimmed the > operator code and don't see an easy way to exclude the bad serialized > bytes. I could fork the interval join code and have a route that writes > badly serialized > > A couple ideas: > 1. I could fork the interval join code and have a route to handle bad > serialization. > 2. Maybe there's a weird case where the bytes become empty and this is an > exception given for an empty array of bytes? > > Could this be a version issue? My Flink version is v1.12.3 and > Twitter/chill v0.9.4. > > Thoughts? > > > java.lang.RuntimeException: Could not create class > com.myexample.proto.MyJoinedOutput > > at > com.twitter.chill.protobuf.ProtobufSerializer.read(ProtobufSerializer.java:76) > ~[blob_p-b0913791f2636dfd2781bd693704accd5ae305e8-26c95e4b9b0bf7dc8409a4d2bd4c57bf:?] > > at > com.twitter.chill.protobuf.ProtobufSerializer.read(ProtobufSerializer.java:40) > ~[blob_p-b0913791f2636dfd2781bd693704accd5ae305e8-26c95e4b9b0bf7dc8409a4d2bd4c57bf:?] > > at > com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.deserialize(KryoSerializer.java:354) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.api.operators.co.IntervalJoinOperator$BufferEntrySerializer.deserialize(IntervalJoinOperator.java:452) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.api.operators.co.IntervalJoinOperator$BufferEntrySerializer.deserialize(IntervalJoinOperator.java:401) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.api.common.typeutils.base.ListSerializer.deserialize(ListSerializer.java:137) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.api.common.typeutils.base.ListSerializer.deserialize(ListSerializer.java:42) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.contrib.streaming.state.RocksDBMapState.deserializeUserValue(RocksDBMapState.java:393) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.contrib.streaming.state.RocksDBMapState.get(RocksDBMapState.java:127) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.runtime.state.UserFacingMapState.get(UserFacingMapState.java:47) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.api.operators.co.IntervalJoinOperator.addToBuffer(IntervalJoinOperator.java:282) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.api.operators.co.IntervalJoinOperator.processElement(IntervalJoinOperator.java:234) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.api.operators.co.IntervalJoinOperator.processElement1(IntervalJoinOperator.java:194) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.processRecord1(StreamTwoInputProcessorFactory.java:199) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.lambda$create$0(StreamTwoInputProcessorFactory.java:164) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:277) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:204) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:174) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor.processInput(StreamTwoInputProcessor.java:95) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:396) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:617) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:581) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) > [flink-dist_2.12-1.12.3.jar:1.12.3] > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) > [flink-dist_2.12-1.12.3.jar:1.12.3] > > at java.lang.Thread.run(Thread.java:748) [?:1.8.0_292] > > Caused by: com.esotericsoftware.kryo.KryoException: java.io.EOFException: > No more bytes left. > > at > org.apache.flink.api.java.typeutils.runtime.NoFetchingInput.readBytes(NoFetchingInput.java:128) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at com.esotericsoftware.kryo.io.Input.readBytes(Input.java:314) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > com.twitter.chill.protobuf.ProtobufSerializer.read(ProtobufSerializer.java:73) > ~[blob_p-b0913791f2636dfd2781bd693704accd5ae305e8-26c95e4b9b0bf7dc8409a4d2bd4c57bf:?] > > ... 27 more > > Caused by: java.io.EOFException: No more bytes left. > > at > org.apache.flink.api.java.typeutils.runtime.NoFetchingInput.readBytes(NoFetchingInput.java:128) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at com.esotericsoftware.kryo.io.Input.readBytes(Input.java:314) > ~[flink-dist_2.12-1.12.3.jar:1.12.3] > > at > com.twitter.chill.protobuf.ProtobufSerializer.read(ProtobufSerializer.java:73) > ~[blob_p-b0913791f2636dfd2781bd693704accd5ae305e8-26c95e4b9b0bf7dc8409a4d2bd4c57bf:?] > > ... 27 more > -- best, Zhipeng
