[jira] [Commented] (FLINK-14087) throws java.lang.ArrayIndexOutOfBoundsException when emiting the data using RebalancePartitioner.

["Yun Gao (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-14087?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17031354#comment-17031354 ] 

Yun Gao commented on FLINK-14087:
---------------------------------

Hi all, we have met with a similar problem online recently: uses write the code

 
{code:java}
 a = env.addSource().rebalance();
 b = a.map()
 c = b.map() {code}
 

and the rebalance partition is then shared between the edges _a -> b_ and _a -> c_, which causes half of tasks of _b_ and _c_ do not receive records. As a temporary fix, we asked users to use the following code instead

 
{code:java}
a = env.addSource();
b = a.rebalance().map();
c = b.rebalance().map()
{code}
 

 

In summary, currently when downstream operators are connected to the same rebalance() PartitionTransformation, the partitioner is shared between all the edges at runtime. Otherwise when the downstream operators are connected to different rebalance() PartitionTransformation, each edge will have its own partitioner at runtime. I think this should also be reasonable, but it seems to be a little divergence from the intuitive thought of users.

 

Therefore, do we still think always use distinct partitioner for each edge is better ? If so, we should copy the partitioner for each edge when creating graph. The down side of the change will still be that if users really want to share the same partitioner between the edges, he would not be able to achieve this after the change.

> throws java.lang.ArrayIndexOutOfBoundsException  when emiting the data using RebalancePartitioner. 
> ---------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-14087
>                 URL: https://issues.apache.org/jira/browse/FLINK-14087
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / Network
>    Affects Versions: 1.8.0, 1.8.1, 1.9.0
>            Reporter: luojiangyu
>            Priority: Major
>         Attachments: image-2019-09-16-19-14-39-403.png, image-2019-09-16-19-15-34-639.png
>
>
> There is the condition the RecordWriter sharing the ChannelSelector instance.
> When two RecordWriter instance shared the same ChannelSelector Instance , It may throws 
> java.lang.ArrayIndexOutOfBoundsException .  For example,  two recordWriter instance shared the RebalancePartitioner instance. the RebalancePartitioner instance setup 2 number of Channels when the first RecordWriter initializing, next the some RebalancePartitioner instance setup 3 number of  channels When the second RecordWriter initializing. this  throws ArrayIndexOutOfBoundsException when the first RecordWriter instance emits the data.
> The Exception likes
> |java.lang.RuntimeException: 2 at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:112) at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:91) at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:47) at org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:673) at org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:617) at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:726) at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:699) at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104) at com.xx.flink.demo.wordcount.case3.StateTest$Source.run(StateTest.java:107) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:94) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:57) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:97) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:302) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:734) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.ArrayIndexOutOfBoundsException: 2 at org.apache.flink.runtime.io.network.api.writer.RecordWriter.getBufferBuilder(RecordWriter.java:255) at org.apache.flink.runtime.io.network.api.writer.RecordWriter.copyFromSerializerToTargetChannel(RecordWriter.java:177) at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:162) at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:128) at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:109) ... 14 more|



--
This message was sent by Atlassian Jira
(v8.3.4#803005)