This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit aff06f86705abbe70723a28e336dd8d6d8fecd93 Author: Sergey Nuyanzin <[email protected]> AuthorDate: Fri Sep 4 07:34:04 2026 +0200 Revert "[FLINK-39977][runtime] Recovery of merged channel state handles" This reverts commit 85d60af8b4413c2d223982bf7bc1e17f0cfedae6. --- .../SubtaskFileMergingManagerRestoreOperation.java | 11 +--- .../FileMergingSnapshotManagerTestBase.java | 67 +--------------------- 2 files changed, 4 insertions(+), 74 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/SubtaskFileMergingManagerRestoreOperation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/SubtaskFileMergingManagerRestoreOperation.java index 9dd8c59d028..36c064efb84 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/SubtaskFileMergingManagerRestoreOperation.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/SubtaskFileMergingManagerRestoreOperation.java @@ -35,8 +35,6 @@ import org.apache.flink.util.Preconditions; import java.util.function.Function; import java.util.stream.Stream; -import static org.apache.flink.runtime.state.ChannelStateHelper.collectUniqueDisposableInChannelState; - /** * Restore operation that restores file-merging information belonging to one subtask for {@link * FileMergingSnapshotManager}. @@ -91,15 +89,10 @@ public class SubtaskFileMergingManagerRestoreOperation { subtaskState.getRawOperatorState().stream()) .flatMap(this::getChildrenStreamHandles); - Stream<StreamStateHandle> channelStateHandles = - collectUniqueDisposableInChannelState( - Stream.of( - subtaskState.getInputChannelState(), - subtaskState.getUpstreamOutputBufferState(), - subtaskState.getResultSubpartitionState())); + // TODO support channel state restore for unaligned checkpoint. Stream<SegmentFileStateHandle> segmentStateHandles = - Stream.of(keyedStateHandles, operatorStateHandles, channelStateHandles) + Stream.of(keyedStateHandles, operatorStateHandles) .flatMap(Function.identity()) .filter( handle -> diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java index 8abb223e699..185300ff641 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java @@ -27,10 +27,7 @@ import org.apache.flink.core.fs.FileSystem; import org.apache.flink.core.fs.Path; import org.apache.flink.core.fs.local.LocalFileSystem; import org.apache.flink.runtime.checkpoint.OperatorSubtaskState; -import org.apache.flink.runtime.checkpoint.StateObjectCollection; import org.apache.flink.runtime.checkpoint.TaskStateSnapshot; -import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo; -import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo; import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SpaceStat; import org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SubtaskKey; import org.apache.flink.runtime.clusterframework.types.ResourceID; @@ -40,14 +37,10 @@ import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; import org.apache.flink.runtime.state.CheckpointedStateScope; import org.apache.flink.runtime.state.IncrementalKeyedStateHandle; import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle; -import org.apache.flink.runtime.state.InputChannelStateHandle; import org.apache.flink.runtime.state.KeyGroupRange; import org.apache.flink.runtime.state.KeyGroupRangeOffsets; import org.apache.flink.runtime.state.KeyGroupsStateHandle; -import org.apache.flink.runtime.state.MergedInputChannelStateHandle; import org.apache.flink.runtime.state.OperatorStateHandle; -import org.apache.flink.runtime.state.OutputStateHandle; -import org.apache.flink.runtime.state.ResultSubpartitionStateHandle; import org.apache.flink.runtime.state.filemerging.FileMergingOperatorStreamStateHandle; import org.apache.flink.runtime.state.filemerging.SegmentFileStateHandle; import org.apache.flink.runtime.state.filesystem.FileMergingCheckpointStateOutputStream; @@ -58,7 +51,6 @@ import org.junit.jupiter.api.io.TempDir; import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -547,8 +539,8 @@ public abstract class FileMergingSnapshotManagerTestBase { assertThat(stateFiles.size()).isEqualTo(1); Set<LogicalFile> restoreFileSet = stateFiles.get(checkpointId); assertThat(restoreFileSet).isNotNull(); - // 2 operators * (2 keyed state + 2 operator state + 3 channel state delegates) - assertThat(restoreFileSet.size()).isEqualTo(14); + // 2 operators * (2 keyed state + 2 operator state) + assertThat(restoreFileSet.size()).isEqualTo(8); assertThat(fmsm.spaceStat).isEqualTo(oldSpaceStat); for (LogicalFile file : restoreFileSet) { assertThat(fmsm.getLogicalFile(file.getFileId())).isEqualTo(file); @@ -749,66 +741,11 @@ public abstract class FileMergingSnapshotManagerTestBase { CheckpointedStateScope.EXCLUSIVE, closeableRegistry)); - SegmentFileStateHandle inputChannelStateHandle = - buildOneSegmentFileHandle( - checkpointId, fmsm, CheckpointedStateScope.EXCLUSIVE, closeableRegistry); - InputChannelStateHandle inputChannel1 = - new InputChannelStateHandle( - 0, - new InputChannelInfo(0, 0), - inputChannelStateHandle, - Collections.singletonList(0L), - 16L); - InputChannelStateHandle inputChannel2 = - new InputChannelStateHandle( - 0, - new InputChannelInfo(0, 1), - inputChannelStateHandle, - Collections.singletonList(16L), - 16L); - - SegmentFileStateHandle upstreamOutputBufferStateHandle = - buildOneSegmentFileHandle( - checkpointId, fmsm, CheckpointedStateScope.EXCLUSIVE, closeableRegistry); - InputChannelStateHandle upstreamOutputBuffer = - new InputChannelStateHandle( - 0, - new InputChannelInfo(1, 0), - upstreamOutputBufferStateHandle, - Collections.singletonList(0L), - upstreamOutputBufferStateHandle.getStateSize()); - - SegmentFileStateHandle resultSubpartitionStateHandle = - buildOneSegmentFileHandle( - checkpointId, fmsm, CheckpointedStateScope.EXCLUSIVE, closeableRegistry); - ResultSubpartitionStateHandle resultSubpartition1 = - new ResultSubpartitionStateHandle( - 0, - new ResultSubpartitionInfo(0, 0), - resultSubpartitionStateHandle, - Collections.singletonList(0L), - 16L); - ResultSubpartitionStateHandle resultSubpartition2 = - new ResultSubpartitionStateHandle( - 0, - new ResultSubpartitionInfo(0, 1), - resultSubpartitionStateHandle, - Collections.singletonList(16L), - 16L); - return OperatorSubtaskState.builder() .setManagedKeyedState(keyedStateHandle1) .setRawKeyedState(keyedStateHandle2) .setManagedOperatorState(operatorStateHandle1) .setRawOperatorState(operatorStateHandle2) - .setInputChannelState( - StateObjectCollection.singleton( - MergedInputChannelStateHandle.fromChannelHandles( - Arrays.asList(inputChannel1, inputChannel2)))) - .setUpstreamOutputBufferState(StateObjectCollection.singleton(upstreamOutputBuffer)) - .setResultSubpartitionState( - new StateObjectCollection<OutputStateHandle>( - Arrays.asList(resultSubpartition1, resultSubpartition2))) .build(); }
