[FLINK-38267] Only call channel state rescaling logic for exchange with channel state to avoid UnsupportedOperationException

[1996fanrui]

Backport #26931 to 2.1

[FLINK-38267] Only call channel state rescaling logic for exchange with channel state to avoid UnsupportedOperationException

[flinkbot]

CI report:

• e415483 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[1996fanrui]

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=69481&view=logs&j=5c8e7682-d68f-54d1-16a2-a09310218a49&t=9d734c8c-6253-55e6-3bce-47e7cdf68ac4&l=8852

The CI is timed out in 2.1, and after analysis, it only happens for UnalignedCheckpointRescaleWithMixedExchangesITCase#createMultiInputDAG. And I am sure it is a test issue instead of production code issue.

The reason is : Co-Map task has 2 inputs exchanges, one is rebalance, another one :

• is forward when source2Parallelism is equal to the parallelism of Co-Map
• is rebalance when source2Parallelism is not equal to the parallelism of Co-Map
• All of parallelism are generated randomly
• Timeout only happens when it is forward, for this case, Co-Map task[1] will disable unaligned checkpoint for all inputs, it causes no inflight buffers during checkpoint. But our test expects waitForCheckpointWithInflightBuffers.

It means FLINK-38267 bug only happens for multiple outputs. For MultiInput cases, the unaligned checkpoint is disabled for all exchanges if any exchanges is disallow unaligned checkpoint, so it is not needed to distinguish specific gate.

So I'd like to delete UnalignedCheckpointRescaleWithMixedExchangesITCase#createMultiInputDAG or introducing an rebalance after forwardedStream to prevent forward exchange for Co-Map.

cc @gaborgsomogyi

[1]

flink/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java

Line 275 in 831f3ab

for (StreamEdge edge : node.getInEdges()) {

[gaborgsomogyi]

Thanks, I'm intended to have a look tomorrow morning.

[gaborgsomogyi]

LGTM.

[1996fanrui]

Thanks @gaborgsomogyi for the review, merging...