You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by za...@apache.org on 2025/03/04 11:55:24 UTC
(flink) 02/07: [FLINK-33734][checkpointing] Use interface replace InputChannelStateHandle and ResultSubpartitionStateHandle in OperatorSubtaskState
This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 32aa7973daaf70e097f247eaf5a3869b47d8e3e0
Author: wangfeifan <zo...@163.com>
AuthorDate: Thu Feb 1 16:10:33 2024 +0800
[FLINK-33734][checkpointing] Use interface replace InputChannelStateHandle and ResultSubpartitionStateHandle in OperatorSubtaskState
---
.../runtime/checkpoint/OperatorSubtaskState.java | 37 +++---
.../PrioritizedOperatorSubtaskState.java | 94 +++++++++++---
.../checkpoint/StateAssignmentOperation.java | 9 +-
.../runtime/checkpoint/TaskStateAssignment.java | 12 +-
.../channel/SequentialChannelStateReaderImpl.java | 6 +-
.../metadata/ChannelStateHandleSerializer.java | 55 ++++----
.../metadata/MetadataV2V3SerializerBase.java | 17 ++-
.../checkpoint/metadata/MetadataV3Serializer.java | 30 ++---
.../flink/runtime/state/ChannelStateHelper.java | 140 +++++++++++++++++++++
.../api/operators/OperatorSnapshotFinalizer.java | 8 +-
.../api/operators/OperatorSnapshotFutures.java | 21 ++--
.../streaming/runtime/tasks/OperatorChain.java | 4 +-
.../CheckpointCoordinatorFailureTest.java | 15 ++-
.../SequentialChannelStateReaderImplTest.java | 7 +-
.../runtime/state/ChannelPersistenceITCase.java | 7 +-
.../operators/OperatorSnapshotFinalizerTest.java | 6 +-
.../api/operators/OperatorSnapshotFuturesTest.java | 17 +--
.../operators/StreamOperatorStateHandlerTest.java | 8 +-
.../flink/streaming/util/OperatorSnapshotUtil.java | 27 ++--
.../runtime/tasks/LocalStateForwardingTest.java | 9 +-
20 files changed, 367 insertions(+), 162 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/OperatorSubtaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/OperatorSubtaskState.java
index 24ff9db5f04..a70b85175d6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/OperatorSubtaskState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/OperatorSubtaskState.java
@@ -19,12 +19,12 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.runtime.state.AbstractChannelStateHandle;
+import org.apache.flink.runtime.state.ChannelState;
import org.apache.flink.runtime.state.CompositeStateHandle;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.state.SharedStateRegistryImpl.EmptyDiscardStateObjectForRegister;
import org.apache.flink.runtime.state.SharedStateRegistryKey;
@@ -42,7 +42,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.apache.flink.runtime.state.AbstractChannelStateHandle.collectUniqueDelegates;
+import static org.apache.flink.runtime.state.ChannelStateHelper.collectUniqueDisposableInChannelState;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -87,9 +87,9 @@ public class OperatorSubtaskState implements CompositeStateHandle {
*/
private final StateObjectCollection<KeyedStateHandle> rawKeyedState;
- private final StateObjectCollection<InputChannelStateHandle> inputChannelState;
+ private final StateObjectCollection<InputStateHandle> inputChannelState;
- private final StateObjectCollection<ResultSubpartitionStateHandle> resultSubpartitionState;
+ private final StateObjectCollection<OutputStateHandle> resultSubpartitionState;
/**
* The subpartitions mappings per partition set when the output operator for a partition was
@@ -122,8 +122,8 @@ public class OperatorSubtaskState implements CompositeStateHandle {
StateObjectCollection<OperatorStateHandle> rawOperatorState,
StateObjectCollection<KeyedStateHandle> managedKeyedState,
StateObjectCollection<KeyedStateHandle> rawKeyedState,
- StateObjectCollection<InputChannelStateHandle> inputChannelState,
- StateObjectCollection<ResultSubpartitionStateHandle> resultSubpartitionState,
+ StateObjectCollection<InputStateHandle> inputChannelState,
+ StateObjectCollection<OutputStateHandle> resultSubpartitionState,
InflightDataRescalingDescriptor inputRescalingDescriptor,
InflightDataRescalingDescriptor outputRescalingDescriptor) {
@@ -151,11 +151,8 @@ public class OperatorSubtaskState implements CompositeStateHandle {
.filter(Objects::nonNull);
}
- private Stream<StateObjectCollection<? extends AbstractChannelStateHandle<?>>>
- streamChannelStates() {
- return Stream.<StateObjectCollection<? extends AbstractChannelStateHandle<?>>>of(
- inputChannelState, resultSubpartitionState)
- .filter(Objects::nonNull);
+ private Stream<StateObjectCollection<? extends ChannelState>> streamChannelStates() {
+ return Stream.of(inputChannelState, resultSubpartitionState).filter(Objects::nonNull);
}
@VisibleForTesting
@@ -189,11 +186,11 @@ public class OperatorSubtaskState implements CompositeStateHandle {
return rawKeyedState;
}
- public StateObjectCollection<InputChannelStateHandle> getInputChannelState() {
+ public StateObjectCollection<InputStateHandle> getInputChannelState() {
return inputChannelState;
}
- public StateObjectCollection<ResultSubpartitionStateHandle> getResultSubpartitionState() {
+ public StateObjectCollection<OutputStateHandle> getResultSubpartitionState() {
return resultSubpartitionState;
}
@@ -208,7 +205,7 @@ public class OperatorSubtaskState implements CompositeStateHandle {
public List<StateObject> getDiscardables() {
return Stream.concat(
streamOperatorAndKeyedStates().flatMap(Collection::stream),
- collectUniqueDelegates(streamChannelStates()))
+ collectUniqueDisposableInChannelState(streamChannelStates()))
.collect(Collectors.toList());
}
@@ -393,9 +390,9 @@ public class OperatorSubtaskState implements CompositeStateHandle {
StateObjectCollection.empty();
private StateObjectCollection<KeyedStateHandle> rawKeyedState =
StateObjectCollection.empty();
- private StateObjectCollection<InputChannelStateHandle> inputChannelState =
+ private StateObjectCollection<InputStateHandle> inputChannelState =
StateObjectCollection.empty();
- private StateObjectCollection<ResultSubpartitionStateHandle> resultSubpartitionState =
+ private StateObjectCollection<OutputStateHandle> resultSubpartitionState =
StateObjectCollection.empty();
private InflightDataRescalingDescriptor inputRescalingDescriptor =
InflightDataRescalingDescriptor.NO_RESCALE;
@@ -447,13 +444,13 @@ public class OperatorSubtaskState implements CompositeStateHandle {
}
public Builder setInputChannelState(
- StateObjectCollection<InputChannelStateHandle> inputChannelState) {
+ StateObjectCollection<InputStateHandle> inputChannelState) {
this.inputChannelState = checkNotNull(inputChannelState);
return this;
}
public Builder setResultSubpartitionState(
- StateObjectCollection<ResultSubpartitionStateHandle> resultSubpartitionState) {
+ StateObjectCollection<OutputStateHandle> resultSubpartitionState) {
this.resultSubpartitionState = checkNotNull(resultSubpartitionState);
return this;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PrioritizedOperatorSubtaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PrioritizedOperatorSubtaskState.java
index e41bcfe7338..3e1aa039f0d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PrioritizedOperatorSubtaskState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PrioritizedOperatorSubtaskState.java
@@ -19,10 +19,14 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
+import org.apache.flink.runtime.state.AbstractChannelStateHandle;
+import org.apache.flink.runtime.state.ChannelState;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.StateObject;
import org.apache.commons.lang3.BooleanUtils;
@@ -33,6 +37,8 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -73,10 +79,9 @@ public class PrioritizedOperatorSubtaskState {
/** List of prioritized snapshot alternatives for raw keyed state. */
private final List<StateObjectCollection<KeyedStateHandle>> prioritizedRawKeyedState;
- private final List<StateObjectCollection<InputChannelStateHandle>> prioritizedInputChannelState;
+ private final List<StateObjectCollection<InputStateHandle>> prioritizedInputChannelState;
- private final List<StateObjectCollection<ResultSubpartitionStateHandle>>
- prioritizedResultSubpartitionState;
+ private final List<StateObjectCollection<OutputStateHandle>> prioritizedResultSubpartitionState;
/** Checkpoint id for a restored operator or null if not restored. */
private final @Nullable Long restoredCheckpointId;
@@ -88,11 +93,9 @@ public class PrioritizedOperatorSubtaskState {
List<StateObjectCollection<OperatorStateHandle>>
prioritizedManagedOperatorState,
@Nonnull List<StateObjectCollection<OperatorStateHandle>> prioritizedRawOperatorState,
+ @Nonnull List<StateObjectCollection<InputStateHandle>> prioritizedInputChannelState,
@Nonnull
- List<StateObjectCollection<InputChannelStateHandle>>
- prioritizedInputChannelState,
- @Nonnull
- List<StateObjectCollection<ResultSubpartitionStateHandle>>
+ List<StateObjectCollection<OutputStateHandle>>
prioritizedResultSubpartitionState,
@Nullable Long restoredCheckpointId) {
@@ -182,13 +185,12 @@ public class PrioritizedOperatorSubtaskState {
}
@Nonnull
- public StateObjectCollection<InputChannelStateHandle> getPrioritizedInputChannelState() {
+ public StateObjectCollection<InputStateHandle> getPrioritizedInputChannelState() {
return lastElement(prioritizedInputChannelState);
}
@Nonnull
- public StateObjectCollection<ResultSubpartitionStateHandle>
- getPrioritizedResultSubpartitionState() {
+ public StateObjectCollection<OutputStateHandle> getPrioritizedResultSubpartitionState() {
return lastElement(prioritizedResultSubpartitionState);
}
@@ -271,10 +273,10 @@ public class PrioritizedOperatorSubtaskState {
new ArrayList<>(size);
List<StateObjectCollection<KeyedStateHandle>> rawKeyedAlternatives =
new ArrayList<>(size);
- List<StateObjectCollection<InputChannelStateHandle>> inputChannelStateAlternatives =
+ List<StateObjectCollection<InputStateHandle>> inputChannelStateAlternatives =
+ new ArrayList<>(size);
+ List<StateObjectCollection<OutputStateHandle>> resultSubpartitionStateAlternatives =
new ArrayList<>(size);
- List<StateObjectCollection<ResultSubpartitionStateHandle>>
- resultSubpartitionStateAlternatives = new ArrayList<>(size);
for (OperatorSubtaskState subtaskState : alternativesByPriority) {
@@ -309,11 +311,31 @@ public class PrioritizedOperatorSubtaskState {
resolvePrioritizedAlternatives(
jobManagerState.getInputChannelState(),
inputChannelStateAlternatives,
- eqStateApprover(InputChannelStateHandle::getInfo)),
+ channelStateApprover(
+ (InputChannelInfo i1, InputChannelInfo i2) -> {
+ if (i1.getGateIdx() == i2.getGateIdx()) {
+ return Integer.compare(
+ i1.getInputChannelIdx(),
+ i2.getInputChannelIdx());
+ } else {
+ return Integer.compare(
+ i1.getGateIdx(), i2.getGateIdx());
+ }
+ })),
resolvePrioritizedAlternatives(
jobManagerState.getResultSubpartitionState(),
resultSubpartitionStateAlternatives,
- eqStateApprover(ResultSubpartitionStateHandle::getInfo)),
+ channelStateApprover(
+ (ResultSubpartitionInfo r1, ResultSubpartitionInfo r2) -> {
+ if (r1.getPartitionIdx() == r2.getPartitionIdx()) {
+ return Integer.compare(
+ r1.getSubPartitionIdx(),
+ r2.getSubPartitionIdx());
+ } else {
+ return Integer.compare(
+ r1.getPartitionIdx(), r2.getPartitionIdx());
+ }
+ })),
restoredCheckpointId);
}
@@ -467,6 +489,44 @@ public class PrioritizedOperatorSubtaskState {
}
}
+ private static <T extends ChannelState, Info> BiFunction<T, T, Boolean> channelStateApprover(
+ Comparator<Info> comparator) {
+ return (ref, alt) -> approveChannelState(ref, alt, comparator);
+ }
+
+ private static <Info> boolean approveChannelState(
+ ChannelState ref, ChannelState alt, Comparator<Info> comparator) {
+ List<Info> refInfos = extractInfos(ref, comparator);
+ List<Info> altInfos = extractInfos(alt, comparator);
+
+ if (refInfos.size() != altInfos.size()) {
+ return false;
+ }
+
+ Iterator<Info> refInfoIterator = refInfos.iterator();
+ Iterator<Info> altInfoIterator = altInfos.iterator();
+ while (refInfoIterator.hasNext()) {
+ Info refInfo = refInfoIterator.next();
+ Info altInfo = altInfoIterator.next();
+ if (!refInfo.equals(altInfo)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /** return sorted info list extracted from the passed in handle. */
+ private static <Info> List<Info> extractInfos(
+ ChannelState handle, Comparator<Info> comparator) {
+ if (handle instanceof AbstractChannelStateHandle) {
+ return Collections.singletonList(((AbstractChannelStateHandle<Info>) handle).getInfo());
+ } else {
+ throw new IllegalStateException(
+ "Not supported InputStateHandle : " + handle.getClass());
+ }
+ }
+
private static <T, E> BiFunction<T, T, Boolean> eqStateApprover(
Function<T, E> identityExtractor) {
return (ref, alt) -> identityExtractor.apply(ref).equals(identityExtractor.apply(alt));
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
index 91085856ad4..6ce8d6a5932 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
@@ -34,6 +34,7 @@ import org.apache.flink.runtime.jobgraph.JobEdge;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.OperatorInstanceID;
import org.apache.flink.runtime.state.AbstractChannelStateHandle;
+import org.apache.flink.runtime.state.ChannelStateHelper;
import org.apache.flink.runtime.state.InputChannelStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
@@ -366,12 +367,12 @@ public class StateAssignmentOperation {
checkForUnsupportedToplogyChanges(
assignment.oldState,
- OperatorSubtaskState::getResultSubpartitionState,
+ ChannelStateHelper::extractUnmergedOutputHandles,
assignment.outputOperatorID);
final OperatorState outputState = assignment.oldState.get(assignment.outputOperatorID);
final List<List<ResultSubpartitionStateHandle>> outputOperatorState =
- splitBySubtasks(outputState, OperatorSubtaskState::getResultSubpartitionState);
+ splitBySubtasks(outputState, ChannelStateHelper::extractUnmergedOutputHandles);
final ExecutionJobVertex executionJobVertex = assignment.executionJobVertex;
final List<IntermediateDataSet> outputs =
@@ -415,7 +416,7 @@ public class StateAssignmentOperation {
checkForUnsupportedToplogyChanges(
stateAssignment.oldState,
- OperatorSubtaskState::getInputChannelState,
+ ChannelStateHelper::extractUnmergedInputHandles,
stateAssignment.inputOperatorID);
final ExecutionJobVertex executionJobVertex = stateAssignment.executionJobVertex;
@@ -425,7 +426,7 @@ public class StateAssignmentOperation {
final OperatorState inputState =
stateAssignment.oldState.get(stateAssignment.inputOperatorID);
final List<List<InputChannelStateHandle>> inputOperatorState =
- splitBySubtasks(inputState, OperatorSubtaskState::getInputChannelState);
+ splitBySubtasks(inputState, ChannelStateHelper::extractUnmergedInputHandles);
boolean hasAnyFullMapper =
executionJobVertex.getJobVertex().getInputs().stream()
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateAssignment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateAssignment.java
index c61e30407a3..75e11d147a4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateAssignment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskStateAssignment.java
@@ -51,6 +51,8 @@ import java.util.function.Function;
import java.util.stream.IntStream;
import static java.util.Collections.emptySet;
+import static org.apache.flink.runtime.state.ChannelStateHelper.castToInputStateCollection;
+import static org.apache.flink.runtime.state.ChannelStateHelper.castToOutputStateCollection;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -168,17 +170,15 @@ class TaskStateAssignment {
|| !subRawKeyedState.containsKey(instanceID),
"If an operator has no managed key state, it should also not have a raw keyed state.");
- final StateObjectCollection<InputChannelStateHandle> inputState =
- getState(instanceID, inputChannelStates);
- final StateObjectCollection<ResultSubpartitionStateHandle> outputState =
- getState(instanceID, resultSubpartitionStates);
return OperatorSubtaskState.builder()
.setManagedOperatorState(getState(instanceID, subManagedOperatorState))
.setRawOperatorState(getState(instanceID, subRawOperatorState))
.setManagedKeyedState(getState(instanceID, subManagedKeyedState))
.setRawKeyedState(getState(instanceID, subRawKeyedState))
- .setInputChannelState(inputState)
- .setResultSubpartitionState(outputState)
+ .setInputChannelState(
+ castToInputStateCollection(inputChannelStates.get(instanceID)))
+ .setResultSubpartitionState(
+ castToOutputStateCollection(resultSubpartitionStates.get(instanceID)))
.setInputRescalingDescriptor(
createRescalingDescriptor(
instanceID,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
index 88b23c3bb7a..6fe1f1a6e99 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.state.AbstractChannelStateHandle;
+import org.apache.flink.runtime.state.ChannelStateHelper;
import org.apache.flink.runtime.state.StreamStateHandle;
import java.io.Closeable;
@@ -64,7 +65,8 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR
read(
stateHandler,
groupByDelegate(
- streamSubtaskStates(), OperatorSubtaskState::getInputChannelState));
+ streamSubtaskStates(),
+ ChannelStateHelper::extractUnmergedInputHandles));
}
}
@@ -80,7 +82,7 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR
stateHandler,
groupByDelegate(
streamSubtaskStates(),
- OperatorSubtaskState::getResultSubpartitionState));
+ ChannelStateHelper::extractUnmergedOutputHandles));
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializer.java
index 1e1979cea64..150caf1f6f4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/ChannelStateHandleSerializer.java
@@ -17,11 +17,11 @@
package org.apache.flink.runtime.checkpoint.metadata;
-import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
-import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
import org.apache.flink.runtime.state.AbstractChannelStateHandle;
import org.apache.flink.runtime.state.AbstractChannelStateHandle.StateContentMetaInfo;
import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.util.function.BiConsumerWithException;
@@ -36,50 +36,51 @@ import java.util.List;
import static org.apache.flink.runtime.checkpoint.metadata.MetadataV2V3SerializerBase.deserializeStreamStateHandle;
import static org.apache.flink.runtime.checkpoint.metadata.MetadataV2V3SerializerBase.serializeStreamStateHandle;
+import static org.apache.flink.runtime.state.ChannelStateHelper.INPUT_CHANNEL_INFO_READER;
+import static org.apache.flink.runtime.state.ChannelStateHelper.INPUT_CHANNEL_INFO_WRITER;
+import static org.apache.flink.runtime.state.ChannelStateHelper.RESULT_SUBPARTITION_INFO_READER;
+import static org.apache.flink.runtime.state.ChannelStateHelper.RESULT_SUBPARTITION_INFO_WRITER;
+/**
+ * ChannelStateHandlerSerializer only support {@link InputChannelStateHandle} and {@link
+ * ResultSubpartitionStateHandle}, used in {@link MetadataV3Serializer} and {@link
+ * MetadataV4Serializer}.
+ */
class ChannelStateHandleSerializer {
- public void serialize(ResultSubpartitionStateHandle handle, DataOutputStream dataOutputStream)
- throws IOException {
+ public void serialize(OutputStateHandle handle, DataOutputStream dos) throws IOException {
+ if (!(handle instanceof ResultSubpartitionStateHandle)) {
+ throw new IllegalStateException(
+ "OutputStateHandle must be ResultSubpartitionStateHandle.");
+ }
+
serializeChannelStateHandle(
- handle,
- dataOutputStream,
- (info, out) -> {
- out.writeInt(info.getPartitionIdx());
- out.writeInt(info.getSubPartitionIdx());
- });
+ (ResultSubpartitionStateHandle) handle, dos, RESULT_SUBPARTITION_INFO_WRITER);
}
- ResultSubpartitionStateHandle deserializeResultSubpartitionStateHandle(
+ public OutputStateHandle deserializeOutputStateHandle(
DataInputStream dis, MetadataV2V3SerializerBase.DeserializationContext context)
throws IOException {
return deserializeChannelStateHandle(
- is -> new ResultSubpartitionInfo(is.readInt(), is.readInt()),
- ResultSubpartitionStateHandle::new,
- dis,
- context);
+ RESULT_SUBPARTITION_INFO_READER, ResultSubpartitionStateHandle::new, dis, context);
}
- public void serialize(InputChannelStateHandle handle, DataOutputStream dos) throws IOException {
+ public void serialize(InputStateHandle handle, DataOutputStream dos) throws IOException {
+ if (!(handle instanceof InputChannelStateHandle)) {
+ throw new IllegalStateException("InputStateHandle must be InputChannelStateHandle.");
+ }
+
serializeChannelStateHandle(
- handle,
- dos,
- (info, dataOutputStream) -> {
- dos.writeInt(info.getGateIdx());
- dos.writeInt(info.getInputChannelIdx());
- });
+ (InputChannelStateHandle) handle, dos, INPUT_CHANNEL_INFO_WRITER);
}
- InputChannelStateHandle deserializeInputChannelStateHandle(
+ public InputStateHandle deserializeInputStateHandle(
DataInputStream dis, MetadataV2V3SerializerBase.DeserializationContext context)
throws IOException {
return deserializeChannelStateHandle(
- is -> new InputChannelInfo(is.readInt(), is.readInt()),
- InputChannelStateHandle::new,
- dis,
- context);
+ INPUT_CHANNEL_INFO_READER, InputChannelStateHandle::new, dis, context);
}
private static <I> void serializeChannelStateHandle(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
index 9f26090fd2c..37d043047d5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java
@@ -30,7 +30,7 @@ import org.apache.flink.runtime.checkpoint.filemerging.LogicalFile;
import org.apache.flink.runtime.state.CheckpointedStateScope;
import org.apache.flink.runtime.state.IncrementalKeyedStateHandle.HandleAndLocalPath;
import org.apache.flink.runtime.state.IncrementalRemoteKeyedStateHandle;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyGroupRangeOffsets;
import org.apache.flink.runtime.state.KeyGroupsSavepointStateHandle;
@@ -38,7 +38,7 @@ import org.apache.flink.runtime.state.KeyGroupsStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.OperatorStreamStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.StateHandleID;
import org.apache.flink.runtime.state.StateObject;
import org.apache.flink.runtime.state.StreamStateHandle;
@@ -696,25 +696,22 @@ public abstract class MetadataV2V3SerializerBase {
// channel state (unaligned checkpoints)
// ------------------------------------------------------------------------
- protected StateObjectCollection<ResultSubpartitionStateHandle>
- deserializeResultSubpartitionStateHandle(
- DataInputStream dis, @Nullable DeserializationContext context)
- throws IOException {
+ protected StateObjectCollection<OutputStateHandle> deserializeResultSubpartitionStateHandle(
+ DataInputStream dis, @Nullable DeserializationContext context) throws IOException {
return StateObjectCollection.empty();
}
- protected StateObjectCollection<InputChannelStateHandle> deserializeInputChannelStateHandle(
+ protected StateObjectCollection<InputStateHandle> deserializeInputChannelStateHandle(
DataInputStream dis, @Nullable DeserializationContext context) throws IOException {
return StateObjectCollection.empty();
}
protected void serializeResultSubpartitionStateHandle(
- ResultSubpartitionStateHandle resultSubpartitionStateHandle, DataOutputStream dos)
+ OutputStateHandle resultSubpartitionStateHandle, DataOutputStream dos)
throws IOException {}
protected void serializeInputChannelStateHandle(
- InputChannelStateHandle inputChannelStateHandle, DataOutputStream dos)
- throws IOException {}
+ InputStateHandle inputChannelStateHandle, DataOutputStream dos) throws IOException {}
// ------------------------------------------------------------------------
// low-level state handles
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java
index fc6df97ce8d..c2fe5134e08 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV3Serializer.java
@@ -26,10 +26,10 @@ import org.apache.flink.runtime.checkpoint.OperatorState;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.jobgraph.OperatorID;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.StateObject;
import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
@@ -203,35 +203,31 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements
@VisibleForTesting
@Override
public void serializeResultSubpartitionStateHandle(
- ResultSubpartitionStateHandle handle, DataOutputStream dos) throws IOException {
+ OutputStateHandle handle, DataOutputStream dos) throws IOException {
channelStateHandleSerializer.serialize(handle, dos);
}
@VisibleForTesting
@Override
- public StateObjectCollection<ResultSubpartitionStateHandle>
- deserializeResultSubpartitionStateHandle(
- DataInputStream dis, @Nullable DeserializationContext context)
- throws IOException {
+ public StateObjectCollection<OutputStateHandle> deserializeResultSubpartitionStateHandle(
+ DataInputStream dis, @Nullable DeserializationContext context) throws IOException {
return deserializeCollection(
- dis,
- context,
- channelStateHandleSerializer::deserializeResultSubpartitionStateHandle);
+ dis, context, channelStateHandleSerializer::deserializeOutputStateHandle);
}
@VisibleForTesting
@Override
- public void serializeInputChannelStateHandle(
- InputChannelStateHandle handle, DataOutputStream dos) throws IOException {
+ public void serializeInputChannelStateHandle(InputStateHandle handle, DataOutputStream dos)
+ throws IOException {
channelStateHandleSerializer.serialize(handle, dos);
}
@VisibleForTesting
@Override
- public StateObjectCollection<InputChannelStateHandle> deserializeInputChannelStateHandle(
+ public StateObjectCollection<InputStateHandle> deserializeInputChannelStateHandle(
DataInputStream dis, @Nullable DeserializationContext context) throws IOException {
return deserializeCollection(
- dis, context, channelStateHandleSerializer::deserializeInputChannelStateHandle);
+ dis, context, channelStateHandleSerializer::deserializeInputStateHandle);
}
private <T extends StateObject> void serializeCollection(
@@ -296,14 +292,14 @@ public class MetadataV3Serializer extends MetadataV2V3SerializerBase implements
}
@VisibleForTesting
- public static StateObjectCollection<InputChannelStateHandle> deserializeInputChannelStateHandle(
+ public static StateObjectCollection<InputStateHandle> deserializeInputChannelStateHandle(
DataInputStream dis) throws IOException {
return INSTANCE.deserializeInputChannelStateHandle(dis, null);
}
@VisibleForTesting
- public StateObjectCollection<ResultSubpartitionStateHandle>
- deserializeResultSubpartitionStateHandle(DataInputStream dis) throws IOException {
+ public StateObjectCollection<OutputStateHandle> deserializeResultSubpartitionStateHandle(
+ DataInputStream dis) throws IOException {
return INSTANCE.deserializeResultSubpartitionStateHandle(dis, null);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ChannelStateHelper.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ChannelStateHelper.java
new file mode 100644
index 00000000000..ff72654ea31
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ChannelStateHelper.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.state;
+
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
+import org.apache.flink.util.function.BiConsumerWithException;
+import org.apache.flink.util.function.FunctionWithException;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Utility class for channel info serialization and conversions between single channel state handle
+ * and merged state handle.
+ */
+public final class ChannelStateHelper {
+
+ private ChannelStateHelper() {}
+
+ public static final BiConsumerWithException<
+ ResultSubpartitionInfo, DataOutputStream, IOException>
+ RESULT_SUBPARTITION_INFO_WRITER =
+ (info, out) -> {
+ out.writeInt(info.getPartitionIdx());
+ out.writeInt(info.getSubPartitionIdx());
+ };
+
+ public static final BiConsumerWithException<InputChannelInfo, DataOutputStream, IOException>
+ INPUT_CHANNEL_INFO_WRITER =
+ (info, dos) -> {
+ dos.writeInt(info.getGateIdx());
+ dos.writeInt(info.getInputChannelIdx());
+ };
+
+ public static final FunctionWithException<DataInputStream, ResultSubpartitionInfo, IOException>
+ RESULT_SUBPARTITION_INFO_READER =
+ is -> new ResultSubpartitionInfo(is.readInt(), is.readInt());
+
+ public static final FunctionWithException<DataInputStream, InputChannelInfo, IOException>
+ INPUT_CHANNEL_INFO_READER = is -> new InputChannelInfo(is.readInt(), is.readInt());
+
+ public static Stream<StreamStateHandle> collectUniqueDisposableInChannelState(
+ Stream<StateObjectCollection<? extends ChannelState>> stateCollections) {
+ return stateCollections
+ .flatMap(Collection::stream)
+ .map(
+ handle -> {
+ if (handle instanceof AbstractChannelStateHandle) {
+ return ((AbstractChannelStateHandle<?>) handle).getDelegate();
+
+ } else {
+ throw new IllegalStateException(
+ "Not Supported state handle : " + handle.getClass());
+ }
+ })
+ .distinct();
+ }
+
+ public static StateObjectCollection<InputStateHandle> castToInputStateCollection(
+ Collection<InputChannelStateHandle> handles) {
+ if (handles == null || handles.isEmpty()) {
+ return StateObjectCollection.empty();
+ }
+
+ return new StateObjectCollection<>(
+ handles.stream().map(e -> ((InputStateHandle) e)).collect(Collectors.toList()));
+ }
+
+ public static StateObjectCollection<OutputStateHandle> castToOutputStateCollection(
+ Collection<ResultSubpartitionStateHandle> handles) {
+ if (handles == null || handles.isEmpty()) {
+ return StateObjectCollection.empty();
+ }
+
+ return new StateObjectCollection<>(
+ handles.stream().map(e -> ((OutputStateHandle) e)).collect(Collectors.toList()));
+ }
+
+ public static StateObjectCollection<InputChannelStateHandle> extractUnmergedInputHandles(
+ OperatorSubtaskState subtaskState) {
+ List<InputChannelStateHandle> inputHandles =
+ subtaskState.getInputChannelState().stream()
+ .flatMap(
+ h -> {
+ if (h instanceof InputChannelStateHandle) {
+ return Stream.of((InputChannelStateHandle) h);
+ } else {
+ throw new IllegalStateException(
+ "Invalid input channel state : " + h.getClass());
+ }
+ })
+ .collect(Collectors.toList());
+ return inputHandles.isEmpty()
+ ? StateObjectCollection.empty()
+ : new StateObjectCollection<>(inputHandles);
+ }
+
+ public static StateObjectCollection<ResultSubpartitionStateHandle> extractUnmergedOutputHandles(
+ OperatorSubtaskState subtaskState) {
+ List<ResultSubpartitionStateHandle> outputHandles =
+ subtaskState.getResultSubpartitionState().stream()
+ .flatMap(
+ h -> {
+ if (h instanceof ResultSubpartitionStateHandle) {
+ return Stream.of((ResultSubpartitionStateHandle) h);
+ } else {
+ throw new IllegalStateException(
+ "Invalid output channel state : " + h.getClass());
+ }
+ })
+ .collect(Collectors.toList());
+ return outputHandles.isEmpty()
+ ? StateObjectCollection.empty()
+ : new StateObjectCollection<>(outputHandles);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizer.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizer.java
index 27051f1d57c..c308a6cb8ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizer.java
@@ -20,10 +20,10 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.util.concurrent.FutureUtils;
@@ -62,10 +62,10 @@ public class OperatorSnapshotFinalizer {
SnapshotResult<OperatorStateHandle> operatorRaw =
FutureUtils.runIfNotDoneAndGet(snapshotFutures.getOperatorStateRawFuture());
- SnapshotResult<StateObjectCollection<InputChannelStateHandle>> inputChannel =
+ SnapshotResult<StateObjectCollection<InputStateHandle>> inputChannel =
snapshotFutures.getInputChannelStateFuture().get();
- SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>> resultSubpartition =
+ SnapshotResult<StateObjectCollection<OutputStateHandle>> resultSubpartition =
snapshotFutures.getResultSubpartitionStateFuture().get();
jobManagerOwnedState =
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFutures.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFutures.java
index a8dad0f5de2..ce36d1fe5ba 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFutures.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFutures.java
@@ -21,10 +21,10 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.state.DoneFuture;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.StateObject;
@@ -51,11 +51,10 @@ public class OperatorSnapshotFutures {
@Nonnull private RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateRawFuture;
@Nonnull
- private Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
- inputChannelStateFuture;
+ private Future<SnapshotResult<StateObjectCollection<InputStateHandle>>> inputChannelStateFuture;
@Nonnull
- private Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
+ private Future<SnapshotResult<StateObjectCollection<OutputStateHandle>>>
resultSubpartitionStateFuture;
public OperatorSnapshotFutures() {
@@ -74,10 +73,10 @@ public class OperatorSnapshotFutures {
@Nonnull RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateManagedFuture,
@Nonnull RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateRawFuture,
@Nonnull
- Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
+ Future<SnapshotResult<StateObjectCollection<InputStateHandle>>>
inputChannelStateFuture,
@Nonnull
- Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
+ Future<SnapshotResult<StateObjectCollection<OutputStateHandle>>>
resultSubpartitionStateFuture) {
this.keyedStateManagedFuture = keyedStateManagedFuture;
this.keyedStateRawFuture = keyedStateRawFuture;
@@ -130,27 +129,27 @@ public class OperatorSnapshotFutures {
}
@Nonnull
- public Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
+ public Future<SnapshotResult<StateObjectCollection<InputStateHandle>>>
getInputChannelStateFuture() {
return inputChannelStateFuture;
}
public void setInputChannelStateFuture(
@Nonnull
- Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
+ Future<SnapshotResult<StateObjectCollection<InputStateHandle>>>
inputChannelStateFuture) {
this.inputChannelStateFuture = inputChannelStateFuture;
}
@Nonnull
- public Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
+ public Future<SnapshotResult<StateObjectCollection<OutputStateHandle>>>
getResultSubpartitionStateFuture() {
return resultSubpartitionStateFuture;
}
public void setResultSubpartitionStateFuture(
@Nonnull
- Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
+ Future<SnapshotResult<StateObjectCollection<OutputStateHandle>>>
resultSubpartitionStateFuture) {
this.resultSubpartitionStateFuture = resultSubpartitionStateFuture;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
index 7e17d63b183..febbcb1bdb0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
@@ -469,14 +469,14 @@ public abstract class OperatorChain<OUT, OP extends StreamOperator<OUT>>
snapshotInProgress.setInputChannelStateFuture(
channelStateWriteResult
.getInputChannelStateHandles()
- .thenApply(StateObjectCollection::new)
+ .thenApply(e -> new StateObjectCollection(e))
.thenApply(SnapshotResult::of));
}
if (op == getTailOperator()) {
snapshotInProgress.setResultSubpartitionStateFuture(
channelStateWriteResult
.getResultSubpartitionStateHandles()
- .thenApply(StateObjectCollection::new)
+ .thenApply(e -> new StateObjectCollection(e))
.thenApply(SnapshotResult::of));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java
index 52cc0d1bcf2..f10d92d1b86 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorFailureTest.java
@@ -53,9 +53,11 @@ import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
import static java.util.Collections.emptyList;
import static org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTest.assertStatsMetrics;
+import static org.apache.flink.runtime.state.ChannelStateHelper.collectUniqueDisposableInChannelState;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.Mockito.mock;
@@ -174,9 +176,18 @@ class CheckpointCoordinatorFailureTest {
verify(operatorSubtaskState.getRawOperatorState().iterator().next()).discardState();
verify(operatorSubtaskState.getManagedKeyedState().iterator().next()).discardState();
verify(operatorSubtaskState.getRawKeyedState().iterator().next()).discardState();
- verify(operatorSubtaskState.getInputChannelState().iterator().next().getDelegate())
+ verify(
+ collectUniqueDisposableInChannelState(
+ Stream.of(operatorSubtaskState.getInputChannelState()))
+ .iterator()
+ .next())
.discardState();
- verify(operatorSubtaskState.getResultSubpartitionState().iterator().next().getDelegate())
+ verify(
+ collectUniqueDisposableInChannelState(
+ Stream.of(
+ operatorSubtaskState.getResultSubpartitionState()))
+ .iterator()
+ .next())
.discardState();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
index 8aeec586b75..5f05af0d92e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.checkpoint.channel;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.MemorySegmentFactory;
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.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
@@ -70,6 +69,8 @@ import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.IntStream.range;
+import static org.apache.flink.runtime.state.ChannelStateHelper.castToInputStateCollection;
+import static org.apache.flink.runtime.state.ChannelStateHelper.castToOutputStateCollection;
import static org.assertj.core.api.Assertions.assertThat;
/** {@link SequentialChannelStateReaderImpl} Test. */
@@ -268,8 +269,8 @@ public class SequentialChannelStateReaderImplTest {
Collections.singletonMap(
new OperatorID(),
OperatorSubtaskState.builder()
- .setInputChannelState(new StateObjectCollection<>(handles.f0))
- .setResultSubpartitionState(new StateObjectCollection<>(handles.f1))
+ .setInputChannelState(castToInputStateCollection(handles.f0))
+ .setResultSubpartitionState(castToOutputStateCollection(handles.f1))
.build()));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
index 30b9aed26af..53c05b19613 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ChannelPersistenceITCase.java
@@ -21,7 +21,6 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
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.ChannelStateWriteRequestExecutorFactory;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter.ChannelStateWriteResult;
@@ -71,6 +70,8 @@ import static java.util.Collections.singletonMap;
import static org.apache.flink.runtime.checkpoint.CheckpointType.CHECKPOINT;
import static org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter.SEQUENCE_NUMBER_UNKNOWN;
import static org.apache.flink.runtime.io.network.buffer.Buffer.DataType.RECOVERY_COMPLETION;
+import static org.apache.flink.runtime.state.ChannelStateHelper.castToInputStateCollection;
+import static org.apache.flink.runtime.state.ChannelStateHelper.castToOutputStateCollection;
import static org.apache.flink.util.CloseableIterator.ofElements;
import static org.assertj.core.api.Assertions.assertThat;
@@ -305,10 +306,10 @@ class ChannelPersistenceITCase {
new OperatorID(),
OperatorSubtaskState.builder()
.setInputChannelState(
- new StateObjectCollection<>(
+ castToInputStateCollection(
t.getInputChannelStateHandles().get()))
.setResultSubpartitionState(
- new StateObjectCollection<>(
+ castToOutputStateCollection(
t.getResultSubpartitionStateHandles().get()))
.build()));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizerTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizerTest.java
index f51fdcff3af..a64905098af 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizerTest.java
@@ -23,9 +23,11 @@ import org.apache.flink.runtime.checkpoint.StateHandleDummyUtil;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.state.DoneFuture;
import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
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.SnapshotResult;
import org.apache.flink.runtime.state.StateObject;
@@ -69,11 +71,11 @@ class OperatorSnapshotFinalizerTest {
withLocalState(deepDummyCopy(operatorTemplate), deepDummyCopy(operatorTemplate));
SnapshotResult<OperatorStateHandle> rawOper =
withLocalState(deepDummyCopy(operatorTemplate), deepDummyCopy(operatorTemplate));
- SnapshotResult<StateObjectCollection<InputChannelStateHandle>> inputChannel =
+ SnapshotResult<StateObjectCollection<InputStateHandle>> inputChannel =
withLocalState(
singleton(deepDummyCopy(inputChannelTemplate)),
singleton(deepDummyCopy(inputChannelTemplate)));
- SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>> resultSubpartition =
+ SnapshotResult<StateObjectCollection<OutputStateHandle>> resultSubpartition =
withLocalState(
singleton(deepDummyCopy(resultSubpartitionTemplate)),
singleton(deepDummyCopy(resultSubpartitionTemplate)));
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFuturesTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFuturesTest.java
index e984616468b..d9c6c23283b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFuturesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFuturesTest.java
@@ -22,11 +22,13 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.state.DoneFuture;
import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyGroupRangeOffsets;
import org.apache.flink.runtime.state.KeyGroupsStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.OperatorStreamStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
@@ -102,18 +104,17 @@ class OperatorSnapshotFuturesTest {
spy(DoneFuture.of(operatorStateRawResult));
InputChannelStateHandle inputChannelRawStateHandle = mock(InputChannelStateHandle.class);
- SnapshotResult<StateObjectCollection<InputChannelStateHandle>> inputChannelStateRawResult =
+ SnapshotResult<StateObjectCollection<InputStateHandle>> inputChannelStateRawResult =
SnapshotResult.of(StateObjectCollection.singleton(inputChannelRawStateHandle));
- Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
- inputChannelStateRawFuture = spy(DoneFuture.of(inputChannelStateRawResult));
+ Future<SnapshotResult<StateObjectCollection<InputStateHandle>>> inputChannelStateRawFuture =
+ spy(DoneFuture.of(inputChannelStateRawResult));
ResultSubpartitionStateHandle resultSubpartitionRawStateHandle =
mock(ResultSubpartitionStateHandle.class);
- SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>
- resultSubpartitionStateRawResult =
- SnapshotResult.of(
- StateObjectCollection.singleton(resultSubpartitionRawStateHandle));
- Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
+ SnapshotResult<StateObjectCollection<OutputStateHandle>> resultSubpartitionStateRawResult =
+ SnapshotResult.of(
+ StateObjectCollection.singleton(resultSubpartitionRawStateHandle));
+ Future<SnapshotResult<StateObjectCollection<OutputStateHandle>>>
resultSubpartitionStateRawFuture =
spy(DoneFuture.of(resultSubpartitionStateRawResult));
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandlerTest.java
index 9cb1d4522f2..2b827aff041 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandlerTest.java
@@ -32,11 +32,11 @@ import org.apache.flink.runtime.metrics.util.InterceptingOperatorMetricGroup;
import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
@@ -77,9 +77,9 @@ class StreamOperatorStateHandlerTest {
new CancelableFuture<>();
RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateRawFuture =
new CancelableFuture<>();
- RunnableFuture<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
+ RunnableFuture<SnapshotResult<StateObjectCollection<InputStateHandle>>>
inputChannelStateFuture = new CancelableFuture<>();
- RunnableFuture<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
+ RunnableFuture<SnapshotResult<StateObjectCollection<OutputStateHandle>>>
resultSubpartitionStateFuture = new CancelableFuture<>();
OperatorSnapshotFutures operatorSnapshotResult =
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java b/flink-runtime/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
index 150ef6b8edb..7f9312f0c38 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
@@ -21,10 +21,10 @@ package org.apache.flink.streaming.util;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.checkpoint.metadata.MetadataV3Serializer;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -101,19 +101,17 @@ public class OperatorSnapshotUtil {
dos.writeInt(-1);
}
- Collection<InputChannelStateHandle> inputChannelStateHandles =
- state.getInputChannelState();
+ Collection<InputStateHandle> inputChannelStateHandles = state.getInputChannelState();
dos.writeInt(inputChannelStateHandles.size());
- for (InputChannelStateHandle inputChannelStateHandle : inputChannelStateHandles) {
+ for (InputStateHandle inputChannelStateHandle : inputChannelStateHandles) {
MetadataV3Serializer.INSTANCE.serializeInputChannelStateHandle(
inputChannelStateHandle, dos);
}
- Collection<ResultSubpartitionStateHandle> resultSubpartitionStateHandles =
+ Collection<OutputStateHandle> resultSubpartitionStateHandles =
state.getResultSubpartitionState();
dos.writeInt(inputChannelStateHandles.size());
- for (ResultSubpartitionStateHandle resultSubpartitionStateHandle :
- resultSubpartitionStateHandles) {
+ for (OutputStateHandle resultSubpartitionStateHandle : resultSubpartitionStateHandles) {
MetadataV3Serializer.INSTANCE.serializeResultSubpartitionStateHandle(
resultSubpartitionStateHandle, dos);
}
@@ -177,17 +175,16 @@ public class OperatorSnapshotUtil {
}
}
- final StateObjectCollection<InputChannelStateHandle> inputChannelStateHandles =
+ final StateObjectCollection<InputStateHandle> inputChannelStateHandles =
v == MetadataV3Serializer.VERSION
? MetadataV3Serializer.deserializeInputChannelStateHandle(dis)
: StateObjectCollection.empty();
- final StateObjectCollection<ResultSubpartitionStateHandle>
- resultSubpartitionStateHandles =
- v == MetadataV3Serializer.VERSION
- ? MetadataV3Serializer.INSTANCE
- .deserializeResultSubpartitionStateHandle(dis)
- : StateObjectCollection.empty();
+ final StateObjectCollection<OutputStateHandle> resultSubpartitionStateHandles =
+ v == MetadataV3Serializer.VERSION
+ ? MetadataV3Serializer.INSTANCE
+ .deserializeResultSubpartitionStateHandle(dis)
+ : StateObjectCollection.empty();
return OperatorSubtaskState.builder()
.setManagedOperatorState(new StateObjectCollection<>(managedOperatorState))
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
index eceda40a3f5..138e129b85c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
@@ -33,12 +33,12 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
import org.apache.flink.runtime.state.DoneFuture;
-import org.apache.flink.runtime.state.InputChannelStateHandle;
+import org.apache.flink.runtime.state.InputStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.LocalRecoveryConfig;
import org.apache.flink.runtime.state.LocalSnapshotDirectoryProviderImpl;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
+import org.apache.flink.runtime.state.OutputStateHandle;
import org.apache.flink.runtime.state.SnapshotResult;
import org.apache.flink.runtime.state.StateObject;
import org.apache.flink.runtime.state.TaskExecutorStateChangelogStoragesManager;
@@ -111,10 +111,9 @@ class LocalStateForwardingTest {
osFuture.setKeyedStateRawFuture(createSnapshotResult(KeyedStateHandle.class));
osFuture.setOperatorStateManagedFuture(createSnapshotResult(OperatorStateHandle.class));
osFuture.setOperatorStateRawFuture(createSnapshotResult(OperatorStateHandle.class));
- osFuture.setInputChannelStateFuture(
- createSnapshotCollectionResult(InputChannelStateHandle.class));
+ osFuture.setInputChannelStateFuture(createSnapshotCollectionResult(InputStateHandle.class));
osFuture.setResultSubpartitionStateFuture(
- createSnapshotCollectionResult(ResultSubpartitionStateHandle.class));
+ createSnapshotCollectionResult(OutputStateHandle.class));
OperatorID operatorID = new OperatorID();
snapshots.put(operatorID, osFuture);