I'll retry the job to see if it's reproducible. The serialized state is bad so that run keeps failing.
On Sun, Dec 19, 2021 at 4:28 PM Zhipeng Zhang <zhangzhipe...@gmail.com> wrote: > 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 > >
