[jira] [Created] (FLINK-38357) ResultSubpartitionRecoveredStateHandler.getSubpartition ArrayIndexOutOfBoundException 0

["Xinglong Wang (Jira)" <ji...@apache.org>]

Xinglong Wang created FLINK-38357:
-------------------------------------

             Summary: ResultSubpartitionRecoveredStateHandler.getSubpartition ArrayIndexOutOfBoundException 0
                 Key: FLINK-38357
                 URL: https://issues.apache.org/jira/browse/FLINK-38357
             Project: Flink
          Issue Type: Bug
          Components: Runtime / Checkpointing
    Affects Versions: 2.1.0, 1.16.1
            Reporter: Xinglong Wang
         Attachments: image-2025-09-15-21-34-19-517.png

h2. Problem

StreamTask: `map -> (filter -> Sink: sink-r, map-q)`, the tailOperator/outputOperator is `Sink: sink-r`.

```

// TaskStateAssignment
outputOperatorID = operatorIDs.get(0).getGeneratedOperatorID();
 
 
// OperatorChain
this.tailOperatorWrapper = allOpWrappers.get(0);
```

While in reality, `map-q` is connected to the downstream Task.

If we cleverly change the parallelism, it can result in this situation：
 * StreamTask will change to `map -> (filter, map-q)`,
 * and `Sink: sink-r` will become a downstream StreamTask: `Sink: sink-r`.

In this situation, ResultSubpartitionRecoveredStateHandler#getSubpartition reports ArrayIndexOutOfBoundException: 0, since Sink does not have writers.

```
java.lang.ArrayIndexOutOfBoundsException: 0
at org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionRecoveredStateHandler.getSubpartition(RecoveredChannelStateHandler.java:217) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionRecoveredStateHandler.lambda$calculateMapping$1(RecoveredChannelStateHandler.java:237) ~[flink-dist-1.16.1.jar:1.16.1]
at java.util.stream.IntPipeline$4$1.accept(IntPipeline.java:250) ~[?:1.8.0_312]
at java.util.Spliterators$IntArraySpliterator.forEachRemaining(Spliterators.java:1032) ~[?:1.8.0_312]
at java.util.Spliterator$OfInt.forEachRemaining(Spliterator.java:693) ~[?:1.8.0_312]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) ~[?:1.8.0_312]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) ~[?:1.8.0_312]
at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708) ~[?:1.8.0_312]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_312]
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:566) ~[?:1.8.0_312]
at org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionRecoveredStateHandler.calculateMapping(RecoveredChannelStateHandler.java:238) ~[flink-dist-1.16.1.jar:1.16.1]
at java.util.HashMap.computeIfAbsent(HashMap.java:1128) ~[?:1.8.0_312]
at org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionRecoveredStateHandler.getMappedChannels(RecoveredChannelStateHandler.java:227) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionRecoveredStateHandler.getBuffer(RecoveredChannelStateHandler.java:182) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionRecoveredStateHandler.getBuffer(RecoveredChannelStateHandler.java:157) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.ChannelStateChunkReader.readChunk(SequentialChannelStateReaderImpl.java:200) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.SequentialChannelStateReaderImpl.readSequentially(SequentialChannelStateReaderImpl.java:109) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.SequentialChannelStateReaderImpl.read(SequentialChannelStateReaderImpl.java:95) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.checkpoint.channel.SequentialChannelStateReaderImpl.readOutputData(SequentialChannelStateReaderImpl.java:81) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:735) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:713) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:679) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:937) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:906) [flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:730) [flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550) [flink-dist-1.16.1.jar:1.16.1]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_312]
```

 
h2. Analysis

When a Task contains multiple output operators, the current implementation only treats one of them as the tailOperator/outputOperator. If the tailOperator/outputOperator is set to Sink, this leads to an error when restoring the channel state for the ResultPartition: indexing into the ResultPartitionWriter fails with an "index 0 out of bounds" exception.

 
h2. Extension

If there's multiple output operators, each connected to a different downstream tasks, the current Flink implementation's assumption of a single tailOperator/outputOperator breaks down. For example, in a job with three outputs:

!image-2025-09-15-21-34-19-517.png!

The mapping between ResultPartition and output operator needs to be more detailed to correctly retain the relationship of channel state restoration.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)