[flink] branch master updated: [FLINK-12641][coordination] Release partitions on every terminal job state

[chesnay]

This is an automated email from the ASF dual-hosted git repository.

chesnay 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 60ab0ae  [FLINK-12641][coordination] Release partitions on every 
terminal job state
60ab0ae is described below

commit 60ab0ae88aa7c3305befe487220c4264deff895e
Author: Chesnay Schepler <ches...@apache.org>
AuthorDate: Sat Jun 29 17:06:18 2019 +0200

    [FLINK-12641][coordination] Release partitions on every terminal job state
---
 .../src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java 
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 19168a5..f638117 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -897,10 +897,8 @@ public class JobMaster extends 
FencedRpcEndpoint<JobMasterId> implements JobMast
 
                if (newJobStatus.isGloballyTerminalState()) {
                        // other terminal job states are handled by the 
executions
-                       if (newJobStatus == JobStatus.FINISHED) {
-                               runAsync(() -> registeredTaskManagers.keySet()
-                                       
.forEach(partitionTracker::stopTrackingAndReleasePartitionsFor));
-                       }
+                       runAsync(() -> registeredTaskManagers.keySet()
+                               
.forEach(partitionTracker::stopTrackingAndReleasePartitionsFor));
 
                        final ArchivedExecutionGraph archivedExecutionGraph = 
schedulerNG.requestJob();
                        scheduledExecutorService.execute(() -> 
jobCompletionActions.jobReachedGloballyTerminalState(archivedExecutionGraph));