zentol commented on a change in pull request #8630: [FLINK-12667][runtime] Add
JobID to TaskExecutorGateway#releasePartitions
URL: https://github.com/apache/flink/pull/8630#discussion_r292811579
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java
##########
@@ -293,6 +299,81 @@ public void
testSlotAllocationCancellationWhenExecutionCancelled() throws Except
assertThat(canceledSlotRequests, equalTo(slotRequests));
}
+ /**
+ * Tests that the partitions are released in case of a execution
cancellation after the execution is already finished.
+ */
+ @Test
+ public void testPartitionReleaseOnCancelAfterFinished() throws
Exception {
+ testPartitionReleaseAfterFinished(Execution::cancel);
+ }
+
+ /**
+ * Tests that the partitions are released in case of a execution
suspension after the execution is already finished.
+ */
+ @Test
+ public void testPartitionReleaseOnSuspendAfterFinished() throws
Exception {
+ testPartitionReleaseAfterFinished(Execution::suspend);
+ }
+
+ private void testPartitionReleaseAfterFinished(Consumer<Execution>
postFinishedExecutionAction) throws Exception {
+ final Tuple2<JobID, Collection<ResultPartitionID>>
releasedPartitions = Tuple2.of(null, null);
+ final SimpleAckingTaskManagerGateway taskManagerGateway = new
SimpleAckingTaskManagerGateway();
+
taskManagerGateway.setReleasePartitionsConsumer(releasedPartitions::setFields);
+
+ final SimpleSlot slot = new SimpleSlot(
+ new SingleSlotTestingSlotOwner(),
+ new LocalTaskManagerLocation(),
+ 0,
+ taskManagerGateway);
+
+ final JobVertex producerVertex = createNoOpJobVertex();
+ final JobVertex consumerVertex = createNoOpJobVertex();
+ consumerVertex.connectNewDataSetAsInput(producerVertex,
DistributionPattern.ALL_TO_ALL, ResultPartitionType.BLOCKING);
+
+ final ProgrammedSlotProvider slotProvider = new
ProgrammedSlotProvider(1);
+ slotProvider.addSlot(producerVertex.getID(), 0,
CompletableFuture.completedFuture(slot));
+
+ ExecutionGraph executionGraph =
ExecutionGraphTestUtils.createSimpleTestGraph(
+ new JobID(),
+ slotProvider,
+ new NoRestartStrategy(),
+ producerVertex,
+ consumerVertex);
+
+
executionGraph.start(TestingComponentMainThreadExecutorServiceAdapter.forMainThread());
+
+ ExecutionJobVertex executionJobVertex =
executionGraph.getJobVertex(producerVertex.getID());
+ ExecutionVertex executionVertex =
executionJobVertex.getTaskVertices()[0];
+
+ final Execution execution =
executionVertex.getCurrentExecutionAttempt();
+
+ execution.allocateAndAssignSlotForExecution(
+ slotProvider,
+ false,
+ LocationPreferenceConstraint.ALL,
+ Collections.emptySet(),
+ TestingUtils.infiniteTime());
+
+ execution.deploy();
+ execution.switchToRunning();
+
+ // simulate a case where a cancel/suspend call is too slow and
the task is already finished
+ // in this case we have to explicitly release the finished
partition
+ // if the task were canceled properly the TM would release the
partition automatically
+ execution.markFinished();
+ postFinishedExecutionAction.accept(execution);
+
+ assertEquals(executionGraph.getJobID(), releasedPartitions.f0);
+ assertEquals(executionVertex.getProducedPartitions().size(),
releasedPartitions.f1.size());
+ for (ResultPartitionID partitionId : releasedPartitions.f1) {
+ IntermediateResultPartition intermediateResultPartition
= executionVertex
+ .getProducedPartitions()
+ .get(partitionId.getPartitionId());
+ assertNotNull(intermediateResultPartition);
Review comment:
this check is done to ensure that the ids of all released partitions are
actually valid. without it the test would pass even if completely random
partitions were passed to the task executor.
I'll add a comment to clarify this; I had to think for a bit myself as to
what we're checking here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
