Hi Timo,
The code example I posted doesn't really match the code that is causing
this issue. I tried to extend it a bit but couldn't make the
reproduction work there.
I am no longer using the serialized strings, but registering the custom
serializers with the runtime during bootstrap and overriding
getTypeInference to provide the raw data type.
But again, I disabled the custom serializer for the test to make sure it
is not the one causing the issues.
Regarding FLINK-20986
<https://issues.apache.org/jira/browse/FLINK-20986>, I'm not sure but I
am no longer using the old type system so everything should pass through
InternalTypeInfo and RawType. I don't see any type equality issues, and
I see the same serializer being invoked for both serialization and
deserialization.
On Thu, Jan 28, 2021 at 5:51 PM Timo Walther <twal...@apache.org
<mailto:twal...@apache.org>> wrote:
This is helpful information. So I guess the problem must be in the
flink-table module and not in flink-core. I will try to reserve some
time tomorrow to look into the code again. How did you express
RawType(Array[String])? Again with fully serialized type string?
Could it be related to
https://issues.apache.org/jira/browse/FLINK-20986
<https://issues.apache.org/jira/browse/FLINK-20986> ?
Regards,
Timo
On 28.01.21 16:30, Yuval Itzchakov wrote:
> Hi Timo,
>
> I tried replacing it with an ordinary ARRAY<STRING> DataType, which
> doesn't reproduce the issue.
> If I use a RawType(Array[String]), the problem still manifests, so I
> assume it's not directly related to a Kryo serialization of the
specific
> underlying type (io.circe.Json), but something in the way it
interacts
> with BinaryRawValueData and writing out to the network buffer
behind the
> scenes.
>
> On Thu, Jan 28, 2021 at 5:26 PM Timo Walther <twal...@apache.org
<mailto:twal...@apache.org>
> <mailto:twal...@apache.org <mailto:twal...@apache.org>>> wrote:
>
> Hi Yuval,
>
> we should definitely find the root cause of this issue. It
helps if the
> exception happens frequently to nail down the problem.
>
> Have you tried to replace the JSON object with a regular
String? If the
> exception is gone after this change. I believe it must be the
> serialization and not the network stack.
>
> Regards,
> Timo
>
>
> On 28.01.21 10:29, Yuval Itzchakov wrote:
> > Hi,
> >
> > I previously wrote about a problem I believed was caused
by Kryo
> > serialization
> >
>
(https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E <https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E>
>
<https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E <https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E>>
>
> >
>
<https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E <https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E>
>
<https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E <https://mail-archives.apache.org/mod_mbox/flink-user/202101.mbox/%3cf9c44b72-f221-8cd3-24cc-b1090345c...@apache.org%3E>>>),
>
> > which I am no longer sure is the case.
> >
> > I have a job which involves a TableScan via a custom
source operator
> > which generates a DataStream[RowData], a UDF to parse out a
> String =>
> > io.circe.Json object (which internally flows as a
> RAW('io.circe.Json')
> > data-type), and then an AggregateFunction with a
java.util.List
> > accumulator which returns one of these objects and is used
in a
> tumbling
> > window as follows:
> >
> > SELECT any_json_array_value(parse_array(resources)) as
> resources_sample
> > FROM foo
> > GROUP BY TUMBLE(event_time, INTERVAL '1' HOUR)
> >
> > It generates the following physical plan:
> >
> > optimize result:
> > Sink(table=[catalog.default-db.foo],
fields=[resources_sample])
> > +- GroupWindowAggregate(window=[TumblingGroupWindow('w$,
event_time,
> > 3600000)], select=[any_json_array_value($f1) AS
resources_sample])
> > +- Exchange(distribution=[single])
> > +- Calc(select=[event_time, parse_array(resources)
AS $f1])
> > +- WatermarkAssigner(rowtime=[event_time],
> watermark=[event_time])
> > +- TableSourceScan(table=[[catalog,
default-db, foo]],
> > fields=[resources])
> >
> > When I run my job, I receive the following exception after
10 - 30
> > seconds (it varies, which gives me a hunch this is related to
> some race
> > condition that might be happening):
> >
> > Caused by: java.io.IOException: Can't get next record for
channel
> > InputChannelInfo{gateIdx=0, inputChannelIdx=0}
> > at
> > org.apache.flink.streaming.runtime.io
<http://org.apache.flink.streaming.runtime.io>
> <http://runtime.io
<http://runtime.io>>.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:166)
> > at
> > org.apache.flink.streaming.runtime.io
<http://org.apache.flink.streaming.runtime.io>
> <http://runtime.io
<http://runtime.io>>.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
> > at
> >
>
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395)
> > at
> >
>
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191)
> > at
> >
>
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609)
> > at
> >
>
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573)
> > at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755)
> > at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:570)
> > at java.lang.Thread.run(Thread.java:748)
> > Caused by: java.io.IOException: Serializer consumed more bytes
> than the
> > record had. This indicates broken serialization. If you
are using
> custom
> > serialization types (Value or Writable), check their
serialization
> > methods. If you are using a Kryo-serialized type, check the
> > corresponding Kryo serializer.
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:339)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNonSpanningRecord(SpillingAdaptiveSpanningRecordDeserializer.java:111)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:86)
> > at
> > org.apache.flink.streaming.runtime.io
<http://org.apache.flink.streaming.runtime.io>
> <http://runtime.io
<http://runtime.io>>.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:163)
> > ... 8 more
> > Caused by: java.lang.IndexOutOfBoundsException: pos:
> 140289414591019,
> > length: 546153590, index: 43, offset: 0
> > at
> >
>
org.apache.flink.core.memory.HybridMemorySegment.get(HybridMemorySegment.java:190)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readFully(NonSpanningWrapper.java:101)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readFully(NonSpanningWrapper.java:92)
> > at
> >
>
org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer.deserialize(BinaryRowDataSerializer.java:100)
> > at
> >
>
org.apache.flink.table.runtime.typeutils.RowDataSerializer.deserialize(RowDataSerializer.java:108)
> > at
> >
>
org.apache.flink.table.runtime.typeutils.RowDataSerializer.deserialize(RowDataSerializer.java:48)
> > at
> >
>
org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:191)
> > at
> >
>
org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:46)
> > at
> >
>
org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:53)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:337)
> > ... 11 more
> > Caused by: java.lang.IndexOutOfBoundsException: pos:
> 140289414591019,
> > length: 546153590, index: 43, offset: 0
> > at
> >
>
org.apache.flink.core.memory.HybridMemorySegment.get(HybridMemorySegment.java:190)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readFully(NonSpanningWrapper.java:101)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readFully(NonSpanningWrapper.java:92)
> > at
> >
>
org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer.deserialize(BinaryRowDataSerializer.java:100)
> > at
> >
>
org.apache.flink.table.runtime.typeutils.RowDataSerializer.deserialize(RowDataSerializer.java:108)
> > at
> >
>
org.apache.flink.table.runtime.typeutils.RowDataSerializer.deserialize(RowDataSerializer.java:48)
> > at
> >
>
org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:191)
> > at
> >
>
org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:46)
> > at
> >
>
org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:53)
> > at
> > org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>
> <http://org.apache.flink.runtime.io
<http://org.apache.flink.runtime.io>>.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:337)
> > ... 11 more
> >
> > Just to say that right now I am using the built in Kryo
> serialization
> > with no custom serializers.
> >
> > Upon further investigation, it seems that
StreamElementSerializer is
> > receiving a corrupt stream, which causes it to think it
received a
> > record with timestamp:
> >
> > image.png
> >
> > As you can see, the tag is incorrectly as 0 and then a
timestamp is
> > attempted to be read, yielding an invalid value (16), and
once the
> > TypeSerialzier (BinaryRowDataSerializer) will try to
decode this,
> it'll
> > fail with the index out of bounds exception.
> >
> > The interesting thing is that, when I *disable operator
chaining*
> > completely via
> (StreamExecutionEnvironment.disableOperatorChaining), the
> > problem does not reproduce.
> >
> > I am wondering which sections of the networking +
serialization
> stack
> > may help me further investigate this issue and understand
what is
> > causing the corrupt stream to emerge, or perhaps if there are
> additional
> > logs that could assist.
> >
> > --
> > Best Regards,
> > Yuval Itzchakov.
>
>
>
> --
> Best Regards,
> Yuval Itzchakov.
--
Best Regards,
Yuval Itzchakov.
