You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "pnowojski (via GitHub)" <gi...@apache.org> on 2023/05/15 13:12:21 UTC
[GitHub] [flink] pnowojski commented on a diff in pull request #22584: [FLINK-31963][state] Fix rescaling bug in recovery from unaligned checkpoints.
pnowojski commented on code in PR #22584:
URL: https://github.com/apache/flink/pull/22584#discussion_r1193812795
##########
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java:
##########
@@ -785,6 +789,130 @@ public void testOnlyUpstreamChannelStateAssignment()
}
}
+ /** FLINK-31963: Tests rescaling for stateless operators and upstream result partition state. */
+ @Test
+ public void testOnlyUpstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState upstreamOpState =
+ OperatorSubtaskState.builder()
+ .setResultSubpartitionState(
+ new StateObjectCollection<>(
+ asList(
+ createNewResultSubpartitionStateHandle(10, random),
+ createNewResultSubpartitionStateHandle(
+ 10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(upstreamOpState, null, 5, 7);
+ }
+
+ /** FLINK-31963: Tests rescaling for stateless operators and downstream input channel state. */
+ @Test
+ public void testOnlyDownstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState downstreamOpState =
+ OperatorSubtaskState.builder()
+ .setInputChannelState(
+ new StateObjectCollection<>(
+ asList(
+ createNewInputChannelStateHandle(10, random),
+ createNewInputChannelStateHandle(10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(null, downstreamOpState, 5, 5);
+ }
+
+ private void testOnlyUpstreamOrDownstreamRescalingInternal(
+ @Nullable OperatorSubtaskState upstreamOpState,
+ @Nullable OperatorSubtaskState downstreamOpState,
+ int expectedUpstreamCount,
+ int expectedDownstreamCount)
+ throws JobException, JobExecutionException {
+
+ if ((upstreamOpState == null && downstreamOpState == null)
+ || (upstreamOpState != null && downstreamOpState != null)) {
+ // Either upstream or downstream state must exist, but not both.
+ return;
+ }
+
+ // Start from parallelism 5 for both operators
+ int upstreamParallelism = 5;
+ int downstreamParallelism = 5;
+
+ // Build states
+ List<OperatorID> operatorIds = buildOperatorIds(2);
+ Map<OperatorID, OperatorState> states = new HashMap<>();
+ OperatorState upstreamState =
+ new OperatorState(operatorIds.get(0), upstreamParallelism, MAX_P);
+ OperatorState downstreamState =
+ new OperatorState(operatorIds.get(1), downstreamParallelism, MAX_P);
+
+ states.put(operatorIds.get(0), upstreamState);
+ states.put(operatorIds.get(1), downstreamState);
+
+ if (upstreamOpState != null) {
+ upstreamState.putState(0, upstreamOpState);
+ // rescale downstream 5 -> 3
+ downstreamParallelism = 3;
+ }
+
+ if (downstreamOpState != null) {
+ downstreamState.putState(0, downstreamOpState);
+ // rescale upstream 5 -> 3
+ upstreamParallelism = 3;
+ }
+
+ List<OperatorIdWithParallelism> opIdWithParallelism = new ArrayList<>(2);
+ opIdWithParallelism.add(
+ new OperatorIdWithParallelism(operatorIds.get(0), upstreamParallelism));
+ opIdWithParallelism.add(
+ new OperatorIdWithParallelism(operatorIds.get(1), downstreamParallelism));
+
+ Map<OperatorID, ExecutionJobVertex> vertices =
+ buildVertices(opIdWithParallelism, RANGE, ROUND_ROBIN);
+
+ // Run state assignment
+ new StateAssignmentOperation(0, new HashSet<>(vertices.values()), states, false)
+ .assignStates();
+
+ // Check results
+ ExecutionJobVertex upstreamExecutionJobVertex = vertices.get(operatorIds.get(0));
+ ExecutionJobVertex downstreamExecutionJobVertex = vertices.get(operatorIds.get(1));
+
+ List<TaskStateSnapshot> upstreamRescalingDescriptors =
+ getRescalingDescriptorsFromVertex(upstreamExecutionJobVertex);
+ List<TaskStateSnapshot> downstreamRescalingDescriptors =
+ getRescalingDescriptorsFromVertex(downstreamExecutionJobVertex);
+
+ checkMappings(
+ upstreamRescalingDescriptors,
+ TaskStateSnapshot::getOutputRescalingDescriptor,
+ expectedUpstreamCount);
+
+ checkMappings(
+ downstreamRescalingDescriptors,
+ TaskStateSnapshot::getInputRescalingDescriptor,
+ expectedDownstreamCount);
+ }
+
+ private void checkMappings(
+ List<TaskStateSnapshot> taskStateSnapshots,
+ Function<TaskStateSnapshot, InflightDataRescalingDescriptor> extractFun,
+ int expectedCount) {
+ Assert.assertEquals(
+ expectedCount,
+ taskStateSnapshots.stream()
+ .map(extractFun)
+ .mapToInt(
+ x -> {
+ int len = x.getOldSubtaskIndexes(0).length;
+ // Assert that there is a mapping.
+ Assert.assertTrue(len > 0);
+ return len;
+ })
+ .sum());
Review Comment:
Instead of asserting length of the mappings, should we assert the actual mappings? 🤔
##########
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java:
##########
@@ -785,6 +789,130 @@ public void testOnlyUpstreamChannelStateAssignment()
}
}
+ /** FLINK-31963: Tests rescaling for stateless operators and upstream result partition state. */
+ @Test
+ public void testOnlyUpstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState upstreamOpState =
+ OperatorSubtaskState.builder()
+ .setResultSubpartitionState(
+ new StateObjectCollection<>(
+ asList(
+ createNewResultSubpartitionStateHandle(10, random),
+ createNewResultSubpartitionStateHandle(
+ 10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(upstreamOpState, null, 5, 7);
+ }
+
+ /** FLINK-31963: Tests rescaling for stateless operators and downstream input channel state. */
+ @Test
+ public void testOnlyDownstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState downstreamOpState =
+ OperatorSubtaskState.builder()
+ .setInputChannelState(
+ new StateObjectCollection<>(
+ asList(
+ createNewInputChannelStateHandle(10, random),
+ createNewInputChannelStateHandle(10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(null, downstreamOpState, 5, 5);
+ }
+
+ private void testOnlyUpstreamOrDownstreamRescalingInternal(
+ @Nullable OperatorSubtaskState upstreamOpState,
+ @Nullable OperatorSubtaskState downstreamOpState,
+ int expectedUpstreamCount,
+ int expectedDownstreamCount)
+ throws JobException, JobExecutionException {
+
+ if ((upstreamOpState == null && downstreamOpState == null)
+ || (upstreamOpState != null && downstreamOpState != null)) {
+ // Either upstream or downstream state must exist, but not both.
+ return;
+ }
+
+ // Start from parallelism 5 for both operators
+ int upstreamParallelism = 5;
+ int downstreamParallelism = 5;
+
+ // Build states
+ List<OperatorID> operatorIds = buildOperatorIds(2);
+ Map<OperatorID, OperatorState> states = new HashMap<>();
+ OperatorState upstreamState =
+ new OperatorState(operatorIds.get(0), upstreamParallelism, MAX_P);
+ OperatorState downstreamState =
+ new OperatorState(operatorIds.get(1), downstreamParallelism, MAX_P);
+
+ states.put(operatorIds.get(0), upstreamState);
+ states.put(operatorIds.get(1), downstreamState);
+
+ if (upstreamOpState != null) {
+ upstreamState.putState(0, upstreamOpState);
+ // rescale downstream 5 -> 3
+ downstreamParallelism = 3;
+ }
+
+ if (downstreamOpState != null) {
+ downstreamState.putState(0, downstreamOpState);
+ // rescale upstream 5 -> 3
+ upstreamParallelism = 3;
+ }
+
+ List<OperatorIdWithParallelism> opIdWithParallelism = new ArrayList<>(2);
+ opIdWithParallelism.add(
+ new OperatorIdWithParallelism(operatorIds.get(0), upstreamParallelism));
+ opIdWithParallelism.add(
+ new OperatorIdWithParallelism(operatorIds.get(1), downstreamParallelism));
+
+ Map<OperatorID, ExecutionJobVertex> vertices =
+ buildVertices(opIdWithParallelism, RANGE, ROUND_ROBIN);
+
+ // Run state assignment
+ new StateAssignmentOperation(0, new HashSet<>(vertices.values()), states, false)
+ .assignStates();
+
+ // Check results
+ ExecutionJobVertex upstreamExecutionJobVertex = vertices.get(operatorIds.get(0));
+ ExecutionJobVertex downstreamExecutionJobVertex = vertices.get(operatorIds.get(1));
+
+ List<TaskStateSnapshot> upstreamRescalingDescriptors =
+ getRescalingDescriptorsFromVertex(upstreamExecutionJobVertex);
+ List<TaskStateSnapshot> downstreamRescalingDescriptors =
+ getRescalingDescriptorsFromVertex(downstreamExecutionJobVertex);
Review Comment:
Rename `RescalingDescriptors` -> `TaskStateSnapshots`? You are obtaining descriptors from the `TaskStateSnapshot` in the next step within `checkMappings`.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java:
##########
@@ -785,6 +789,130 @@ public void testOnlyUpstreamChannelStateAssignment()
}
}
+ /** FLINK-31963: Tests rescaling for stateless operators and upstream result partition state. */
+ @Test
+ public void testOnlyUpstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState upstreamOpState =
+ OperatorSubtaskState.builder()
+ .setResultSubpartitionState(
+ new StateObjectCollection<>(
+ asList(
+ createNewResultSubpartitionStateHandle(10, random),
+ createNewResultSubpartitionStateHandle(
+ 10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(upstreamOpState, null, 5, 7);
+ }
+
+ /** FLINK-31963: Tests rescaling for stateless operators and downstream input channel state. */
+ @Test
+ public void testOnlyDownstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState downstreamOpState =
+ OperatorSubtaskState.builder()
+ .setInputChannelState(
+ new StateObjectCollection<>(
+ asList(
+ createNewInputChannelStateHandle(10, random),
+ createNewInputChannelStateHandle(10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(null, downstreamOpState, 5, 5);
+ }
+
+ private void testOnlyUpstreamOrDownstreamRescalingInternal(
+ @Nullable OperatorSubtaskState upstreamOpState,
+ @Nullable OperatorSubtaskState downstreamOpState,
+ int expectedUpstreamCount,
+ int expectedDownstreamCount)
+ throws JobException, JobExecutionException {
+
+ if ((upstreamOpState == null && downstreamOpState == null)
+ || (upstreamOpState != null && downstreamOpState != null)) {
+ // Either upstream or downstream state must exist, but not both.
+ return;
+ }
Review Comment:
`checkArgument(..., "Either upstream or downstream state must exist, but not both")`?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperationTest.java:
##########
@@ -785,6 +789,130 @@ public void testOnlyUpstreamChannelStateAssignment()
}
}
+ /** FLINK-31963: Tests rescaling for stateless operators and upstream result partition state. */
+ @Test
+ public void testOnlyUpstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState upstreamOpState =
+ OperatorSubtaskState.builder()
+ .setResultSubpartitionState(
+ new StateObjectCollection<>(
+ asList(
+ createNewResultSubpartitionStateHandle(10, random),
+ createNewResultSubpartitionStateHandle(
+ 10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(upstreamOpState, null, 5, 7);
+ }
+
+ /** FLINK-31963: Tests rescaling for stateless operators and downstream input channel state. */
+ @Test
+ public void testOnlyDownstreamChannelRescaleStateAssignment()
+ throws JobException, JobExecutionException {
+ Random random = new Random();
+ OperatorSubtaskState downstreamOpState =
+ OperatorSubtaskState.builder()
+ .setInputChannelState(
+ new StateObjectCollection<>(
+ asList(
+ createNewInputChannelStateHandle(10, random),
+ createNewInputChannelStateHandle(10, random))))
+ .build();
+ testOnlyUpstreamOrDownstreamRescalingInternal(null, downstreamOpState, 5, 5);
+ }
+
+ private void testOnlyUpstreamOrDownstreamRescalingInternal(
+ @Nullable OperatorSubtaskState upstreamOpState,
+ @Nullable OperatorSubtaskState downstreamOpState,
+ int expectedUpstreamCount,
+ int expectedDownstreamCount)
+ throws JobException, JobExecutionException {
+
+ if ((upstreamOpState == null && downstreamOpState == null)
+ || (upstreamOpState != null && downstreamOpState != null)) {
+ // Either upstream or downstream state must exist, but not both.
+ return;
+ }
+
+ // Start from parallelism 5 for both operators
+ int upstreamParallelism = 5;
+ int downstreamParallelism = 5;
+
+ // Build states
+ List<OperatorID> operatorIds = buildOperatorIds(2);
+ Map<OperatorID, OperatorState> states = new HashMap<>();
+ OperatorState upstreamState =
+ new OperatorState(operatorIds.get(0), upstreamParallelism, MAX_P);
+ OperatorState downstreamState =
+ new OperatorState(operatorIds.get(1), downstreamParallelism, MAX_P);
+
+ states.put(operatorIds.get(0), upstreamState);
+ states.put(operatorIds.get(1), downstreamState);
+
+ if (upstreamOpState != null) {
+ upstreamState.putState(0, upstreamOpState);
+ // rescale downstream 5 -> 3
+ downstreamParallelism = 3;
+ }
+
+ if (downstreamOpState != null) {
+ downstreamState.putState(0, downstreamOpState);
+ // rescale upstream 5 -> 3
+ upstreamParallelism = 3;
+ }
+
+ List<OperatorIdWithParallelism> opIdWithParallelism = new ArrayList<>(2);
+ opIdWithParallelism.add(
+ new OperatorIdWithParallelism(operatorIds.get(0), upstreamParallelism));
+ opIdWithParallelism.add(
+ new OperatorIdWithParallelism(operatorIds.get(1), downstreamParallelism));
+
+ Map<OperatorID, ExecutionJobVertex> vertices =
+ buildVertices(opIdWithParallelism, RANGE, ROUND_ROBIN);
+
+ // Run state assignment
+ new StateAssignmentOperation(0, new HashSet<>(vertices.values()), states, false)
+ .assignStates();
+
+ // Check results
+ ExecutionJobVertex upstreamExecutionJobVertex = vertices.get(operatorIds.get(0));
+ ExecutionJobVertex downstreamExecutionJobVertex = vertices.get(operatorIds.get(1));
+
+ List<TaskStateSnapshot> upstreamRescalingDescriptors =
+ getRescalingDescriptorsFromVertex(upstreamExecutionJobVertex);
+ List<TaskStateSnapshot> downstreamRescalingDescriptors =
+ getRescalingDescriptorsFromVertex(downstreamExecutionJobVertex);
+
+ checkMappings(
+ upstreamRescalingDescriptors,
+ TaskStateSnapshot::getOutputRescalingDescriptor,
+ expectedUpstreamCount);
+
+ checkMappings(
+ downstreamRescalingDescriptors,
+ TaskStateSnapshot::getInputRescalingDescriptor,
+ expectedDownstreamCount);
Review Comment:
nit: instead of lambda functions I would accept a little bit of code deduplication and replace those calls with:
```
checkMappings(
upstreamTaskStateSnapshots.stream().map(TaskStateSnapshot::getOutputRescalingDescriptor),
expectedUpstreamCount);
checkMappings(
downstreamTaskStateSnapshots.stream().map(TaskStateSnapshot::getInputRescalingDescriptor),
expectedDownstreamCount);
```
##########
flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointRescaleITCase.java:
##########
@@ -459,46 +478,59 @@ public void processBroadcastElement(Long value, Context ctx, Collector<Long> out
}
}
- @Parameterized.Parameters(name = "{0} {1} from {2} to {3}, buffersPerChannel = {4}")
+ @Parameterized.Parameters(
+ name = "{0} {1} from {2} to {3}, sourceSleepMs = {4}, buffersPerChannel = {5}")
Review Comment:
I would add a comment above this line explaining why do we want to have non zero `sourceSleepMs` sometimes. That we want to test the rescaling without backpressure with only occasional a couple of captured in-flight records .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org