You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by yu...@apache.org on 2021/08/12 02:13:56 UTC
[flink] branch master updated: [FLINK-23475][runtime][checkpoint]
Supports repartition partly finished broacast states (#16714)
This is an automated email from the ASF dual-hosted git repository.
yuanmei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 1c563a0 [FLINK-23475][runtime][checkpoint] Supports repartition partly finished broacast states (#16714)
1c563a0 is described below
commit 1c563a03e95a7baf91055ce1fec3d2a6f35c393d
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Thu Aug 12 10:13:26 2021 +0800
[FLINK-23475][runtime][checkpoint] Supports repartition partly finished broacast states (#16714)
* [FLINK-23475][runtime][checkpoint] Supports repartition partly finished broacast states
* Fix the judgement of unfinished and wrong comment / variable
* Fix the tests
* Add todo
---
.../RoundRobinOperatorStateRepartitioner.java | 93 ++++++++++++++++------
.../checkpoint/StateAssignmentOperationTest.java | 51 ++++++++++++
2 files changed, 118 insertions(+), 26 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/RoundRobinOperatorStateRepartitioner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/RoundRobinOperatorStateRepartitioner.java
index 1e64a3b..4dc11f5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/RoundRobinOperatorStateRepartitioner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/RoundRobinOperatorStateRepartitioner.java
@@ -27,6 +27,7 @@ import org.apache.flink.util.Preconditions;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.BitSet;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.List;
@@ -69,7 +70,11 @@ public class RoundRobinOperatorStateRepartitioner
Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>>
unionStates = collectUnionStates(previousParallelSubtaskStates);
- if (unionStates.isEmpty()) {
+ Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>>
+ partlyFinishedBroadcastStates =
+ collectPartlyFinishedBroadcastStates(previousParallelSubtaskStates);
+
+ if (unionStates.isEmpty() && partlyFinishedBroadcastStates.isEmpty()) {
return previousParallelSubtaskStates;
}
@@ -77,6 +82,11 @@ public class RoundRobinOperatorStateRepartitioner
mergeMapList = initMergeMapList(previousParallelSubtaskStates);
repartitionUnionState(unionStates, mergeMapList);
+
+ // TODO: Currently if some tasks is finished, we would rescale the
+ // remaining state. A better solution would be not touch the non-empty
+ // subtask state and only fix the empty ones.
+ repartitionBroadcastState(partlyFinishedBroadcastStates, mergeMapList);
} else {
// Reorganize: group by (State Name -> StreamStateHandle + Offsets)
@@ -123,14 +133,31 @@ public class RoundRobinOperatorStateRepartitioner
return mergeMapList;
}
- /** Collect union states from given parallelSubtaskStates. */
private Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>>
collectUnionStates(List<List<OperatorStateHandle>> parallelSubtaskStates) {
+ return collectStates(parallelSubtaskStates, OperatorStateHandle.Mode.UNION).entrySet()
+ .stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().entries));
+ }
- Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>>
- unionStates = new HashMap<>(parallelSubtaskStates.size());
+ private Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>>
+ collectPartlyFinishedBroadcastStates(
+ List<List<OperatorStateHandle>> parallelSubtaskStates) {
+ return collectStates(parallelSubtaskStates, OperatorStateHandle.Mode.BROADCAST).entrySet()
+ .stream()
+ .filter(e -> e.getValue().isPartiallyReported())
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().entries));
+ }
+
+ /** Collect the states from given parallelSubtaskStates with the specific {@code mode}. */
+ private Map<String, StateEntry> collectStates(
+ List<List<OperatorStateHandle>> parallelSubtaskStates, OperatorStateHandle.Mode mode) {
+
+ Map<String, StateEntry> states = new HashMap<>(parallelSubtaskStates.size());
- for (List<OperatorStateHandle> subTaskState : parallelSubtaskStates) {
+ for (int i = 0; i < parallelSubtaskStates.size(); ++i) {
+ final int subtaskIndex = i;
+ List<OperatorStateHandle> subTaskState = parallelSubtaskStates.get(i);
for (OperatorStateHandle operatorStateHandle : subTaskState) {
if (operatorStateHandle == null) {
continue;
@@ -141,28 +168,20 @@ public class RoundRobinOperatorStateRepartitioner
operatorStateHandle.getStateNameToPartitionOffsets().entrySet();
partitionOffsetEntries.stream()
- .filter(
- entry ->
- entry.getValue()
- .getDistributionMode()
- .equals(OperatorStateHandle.Mode.UNION))
+ .filter(entry -> entry.getValue().getDistributionMode().equals(mode))
.forEach(
entry -> {
- List<
- Tuple2<
- StreamStateHandle,
- OperatorStateHandle.StateMetaInfo>>
- stateLocations =
- unionStates.computeIfAbsent(
- entry.getKey(),
- k ->
- new ArrayList<>(
- parallelSubtaskStates
- .size()
- * partitionOffsetEntries
- .size()));
-
- stateLocations.add(
+ StateEntry stateEntry =
+ states.computeIfAbsent(
+ entry.getKey(),
+ k ->
+ new StateEntry(
+ parallelSubtaskStates.size()
+ * partitionOffsetEntries
+ .size(),
+ parallelSubtaskStates.size()));
+ stateEntry.addEntry(
+ subtaskIndex,
Tuple2.of(
operatorStateHandle.getDelegateStateHandle(),
entry.getValue()));
@@ -170,7 +189,7 @@ public class RoundRobinOperatorStateRepartitioner
}
}
- return unionStates;
+ return states;
}
/** Group by the different named states. */
@@ -459,4 +478,26 @@ public class RoundRobinOperatorStateRepartitioner
return byMode.get(mode);
}
}
+
+ private static final class StateEntry {
+ final List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>> entries;
+ final BitSet reportedSubtaskIndices;
+
+ public StateEntry(int estimatedEntrySize, int parallelism) {
+ this.entries = new ArrayList<>(estimatedEntrySize);
+ this.reportedSubtaskIndices = new BitSet(parallelism);
+ }
+
+ void addEntry(
+ int subtaskIndex,
+ Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo> entry) {
+ this.entries.add(entry);
+ reportedSubtaskIndices.set(subtaskIndex);
+ }
+
+ boolean isPartiallyReported() {
+ return reportedSubtaskIndices.cardinality() > 0
+ && reportedSubtaskIndices.cardinality() < reportedSubtaskIndices.size();
+ }
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java
index 33f440d..17f8b2a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java
@@ -191,6 +191,57 @@ public class StateAssignmentOperationTest extends TestLogger {
verifyOneKindPartitionableStateRescale(operatorState, operatorID);
}
+ @Test
+ public void testRepartitionBroadcastStateWithNullSubtaskState() {
+ OperatorID operatorID = new OperatorID();
+ OperatorState operatorState = new OperatorState(operatorID, 2, 4);
+
+ // Only the subtask 0 reports the states.
+ Map<String, OperatorStateHandle.StateMetaInfo> metaInfoMap1 = new HashMap<>(2);
+ metaInfoMap1.put(
+ "t-5",
+ new OperatorStateHandle.StateMetaInfo(
+ new long[] {0, 10, 20}, OperatorStateHandle.Mode.BROADCAST));
+ metaInfoMap1.put(
+ "t-6",
+ new OperatorStateHandle.StateMetaInfo(
+ new long[] {30, 40, 50}, OperatorStateHandle.Mode.BROADCAST));
+ OperatorStateHandle osh1 =
+ new OperatorStreamStateHandle(
+ metaInfoMap1, new ByteStreamStateHandle("test1", new byte[60]));
+ operatorState.putState(
+ 0, OperatorSubtaskState.builder().setManagedOperatorState(osh1).build());
+
+ verifyOneKindPartitionableStateRescale(operatorState, operatorID);
+ }
+
+ @Test
+ public void testRepartitionBroadcastStateWithEmptySubtaskState() {
+ OperatorID operatorID = new OperatorID();
+ OperatorState operatorState = new OperatorState(operatorID, 2, 4);
+
+ // Only the subtask 0 reports the states.
+ Map<String, OperatorStateHandle.StateMetaInfo> metaInfoMap1 = new HashMap<>(2);
+ metaInfoMap1.put(
+ "t-5",
+ new OperatorStateHandle.StateMetaInfo(
+ new long[] {0, 10, 20}, OperatorStateHandle.Mode.BROADCAST));
+ metaInfoMap1.put(
+ "t-6",
+ new OperatorStateHandle.StateMetaInfo(
+ new long[] {30, 40, 50}, OperatorStateHandle.Mode.BROADCAST));
+ OperatorStateHandle osh1 =
+ new OperatorStreamStateHandle(
+ metaInfoMap1, new ByteStreamStateHandle("test1", new byte[60]));
+ operatorState.putState(
+ 0, OperatorSubtaskState.builder().setManagedOperatorState(osh1).build());
+
+ // The subtask 1 report an empty snapshot.
+ operatorState.putState(1, OperatorSubtaskState.builder().build());
+
+ verifyOneKindPartitionableStateRescale(operatorState, operatorID);
+ }
+
/** Verify repartition logic on partitionable states with all modes. */
@Test
public void testReDistributeCombinedPartitionableStates() {