This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 40dca974558 [FLINK-38804][runtime] Ensure channelStateWriter is closed
after the inputGates
new c787fcf3a60 [hotfix][checkpoint] Refactor output buffers distribution
logic via ResultSubpartitionDistributor
new cb5b955212b [hotfix][checkpoint] Limit that the one buffer is only
distributed to one target InputChannel
new 686c00f8e3b [FLINK-38542][checkpoint] Recover output buffers of
upstream task on downstream task side directly
new 5e02d64bf01 [FLINK-38542][checkpoint] Randomize
UNALIGNED_ALLOW_ON_RECOVERY for testing
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../generated/checkpointing_configuration.html | 6 +
.../flink/configuration/CheckpointingOptions.java | 10 +
.../runtime/checkpoint/CheckpointCoordinator.java | 9 +-
.../runtime/checkpoint/OperatorSubtaskState.java | 26 +-
.../checkpoint/StateAssignmentOperation.java | 36 +-
.../runtime/checkpoint/TaskStateAssignment.java | 145 +++++-
.../channel/RecoveredChannelStateHandler.java | 103 ++--
.../channel/ResultSubpartitionDistributor.java | 73 +++
.../channel/SequentialChannelStateReaderImpl.java | 5 +
.../tasks/CheckpointCoordinatorConfiguration.java | 29 +-
.../flink/streaming/api/graph/StreamGraph.java | 5 +
.../checkpoint/StateAssignmentOperationTest.java | 580 +++++++++++++++++----
.../runtime/checkpoint/StateHandleDummyUtil.java | 8 +-
.../InputChannelRecoveredStateHandlerTest.java | 41 ++
.../channel/ResultSubpartitionDistributorTest.java | 139 +++++
.../streaming/util/TestStreamEnvironment.java | 2 +
.../test/state/ChangelogRecoveryCachingITCase.java | 4 +
17 files changed, 1034 insertions(+), 187 deletions(-)
create mode 100644
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ResultSubpartitionDistributor.java
create mode 100644
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ResultSubpartitionDistributorTest.java
