tillrohrmann commented on a change in pull request #9960: [FLINK-14476] Extend
PartitionTracker to support promotions
URL: https://github.com/apache/flink/pull/9960#discussion_r339103795
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTrackerImplTest.java
##########
@@ -233,6 +233,53 @@ public void testStopTrackingIssuesNoReleaseCalls() {
assertEquals(0,
shuffleMaster.externallyReleasedPartitions.size());
}
+ @Test
+ public void testReleaseOrPromote() {
+ final TestingShuffleMaster shuffleMaster = new
TestingShuffleMaster();
+
+ final Queue<Tuple4<ResourceID, JobID,
Collection<ResultPartitionID>, Collection<ResultPartitionID>>>
taskExecutorReleaseCalls = new ArrayBlockingQueue<>(4);
+ final PartitionTracker partitionTracker = new
PartitionTrackerImpl(
+ new JobID(),
+ shuffleMaster,
+ resourceId ->
Optional.of(createTaskExecutorGateway(resourceId, taskExecutorReleaseCalls))
+ );
+
+ final ResourceID taskExecutorId1 = ResourceID.generate();
+ final ResultPartitionID jobPartitionId = new
ResultPartitionID();
+ final ResultPartitionID clusterPartitionId = new
ResultPartitionID();
+
+ // any partition type that is not BLOCKING_PERSISTENT denotes a
job partition
+ final ResultPartitionDeploymentDescriptor jobPartition =
createResultPartitionDeploymentDescriptor(jobPartitionId,
ResultPartitionType.BLOCKING, true);
+ partitionTracker.startTrackingPartition(
+ taskExecutorId1,
+ jobPartition);
+
+ // BLOCKING_PERSISTENT denotes a cluster partition
+ final ResultPartitionDeploymentDescriptor clusterPartition =
createResultPartitionDeploymentDescriptor(clusterPartitionId,
ResultPartitionType.BLOCKING_PERSISTENT, true);
+ partitionTracker.startTrackingPartition(
+ taskExecutorId1,
+ clusterPartition);
+
+
partitionTracker.stopTrackingAndReleaseOrPromotePartitionsFor(taskExecutorId1);
+
+ // exactly one call should have been made to the hosting task
executor
+ assertEquals(1, taskExecutorReleaseCalls.size());
+
+ // the job partition should have been released on the shuffle
master
+ assertEquals(1,
shuffleMaster.externallyReleasedPartitions.size());
+ assertEquals(jobPartitionId,
shuffleMaster.externallyReleasedPartitions.remove());
+
+ final Tuple4<ResourceID, JobID, Collection<ResultPartitionID>,
Collection<ResultPartitionID>> taskExecutorReleaseOrPromoteCall =
taskExecutorReleaseCalls.remove();
+
+ // the job partition should be passed as a partition to release
+ assertEquals(1, taskExecutorReleaseOrPromoteCall.f2.size());
+ assertEquals(jobPartitionId,
taskExecutorReleaseOrPromoteCall.f2.iterator().next());
Review comment:
We could simplify this via
`Iterables.getFirst(taskExecutorReleaseOrPromoteCall.f2)`
----------------------------------------------------------------
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
