GJL commented on a change in pull request #8804: [FLINK-12883][WIP][runtime]
Add elaborated partition release logic
URL: https://github.com/apache/flink/pull/8804#discussion_r295813852
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
##########
@@ -1634,6 +1641,28 @@ public boolean updateState(TaskExecutionState state) {
}
}
+ private void maybeReleasePartitions(final Execution attempt, final
TaskExecutionState state) {
+ final ExecutionVertexID finishedExecutionVertex =
attempt.getVertex().getID();
+
+ if (state.getExecutionState() == ExecutionState.FINISHED) {
+ final List<IntermediateResultPartitionID>
releasablePartitions =
partitionReleaseStrategy.vertexFinished(finishedExecutionVertex);
+ releasePartitions(releasablePartitions);
+ } else {
+
partitionReleaseStrategy.vertexUnfinished(finishedExecutionVertex);
+ }
+ }
+
+ private void releasePartitions(final List<? extends
IntermediateResultPartitionID> releasablePartitions) {
+ for (final IntermediateResultPartitionID resultPartition :
releasablePartitions) {
+ releasePartitions(resultPartition);
+ }
+ }
+
+ private void releasePartitions(final IntermediateResultPartitionID
resultPartition) {
Review comment:
We will have to look up the `ExecutionVertex` of the
`IntermediateResultPartitionID`, which could be done via the
`SchedulingTopology`.
----------------------------------------------------------------
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
