[zeppelin] branch master updated: [ZEPPELIN-5412] Unable to cancel flink job due to fail to take savepoint

[zjffdu]

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

zjffdu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git


The following commit(s) were added to refs/heads/master by this push:
     new ed18c3e  [ZEPPELIN-5412] Unable to cancel flink job due to fail to 
take savepoint
ed18c3e is described below

commit ed18c3e6bfcb64322659c830dbc8ef787bdb3477
Author: Jeff Zhang <zjf...@apache.org>
AuthorDate: Thu Jun 17 07:26:27 2021 +0800

    [ZEPPELIN-5412] Unable to cancel flink job due to fail to take savepoint
    
    ### What is this PR for?
    
    Sometimes flink job is unable to be cancelled when savepoint is enabled. 
This PR would fall back to cancel job without savepoint if cancelling with 
savepoint is failed.
    
    ### What type of PR is it?
    [Bug Fix ]
    
    ### Todos
    * [ ] - Task
    
    ### What is the Jira issue?
    * https://issues.apache.org/jira/browse/ZEPPELIN-5412
    
    ### How should this be tested?
    * Manually tested
    
    ### Screenshots (if appropriate)
    
    ### Questions:
    * Does the licenses files need update? No
    * Is there breaking changes for older versions? No
    * Does this needs documentation? No
    
    Author: Jeff Zhang <zjf...@apache.org>
    
    Closes #4139 from zjffdu/ZEPPELIN-5412 and squashes the following commits:
    
    da99103249 [Jeff Zhang] logging exception
    ef3b16e956 [Jeff Zhang] [ZEPPELIN-5412] Unable to cancel flink job due to 
fail to take savepoint
---
 .../java/org/apache/zeppelin/flink/JobManager.java   | 20 +++++++++++++-------
 1 file changed, 13 insertions(+), 7 deletions(-)

diff --git 
a/flink/flink-scala-parent/src/main/java/org/apache/zeppelin/flink/JobManager.java
 
b/flink/flink-scala-parent/src/main/java/org/apache/zeppelin/flink/JobManager.java
index 9614ee9..d60ff1a 100644
--- 
a/flink/flink-scala-parent/src/main/java/org/apache/zeppelin/flink/JobManager.java
+++ 
b/flink/flink-scala-parent/src/main/java/org/apache/zeppelin/flink/JobManager.java
@@ -141,13 +141,19 @@ public class JobManager {
       } else {
         LOGGER.info("Trying to stop job of paragraph {} with save point dir: 
{}",
                 context.getParagraphId(), savePointDir);
-        String savePointPath = jobClient.stopWithSavepoint(true, 
savePointDir).get();
-        Map<String, String> config = new HashMap<>();
-        config.put(SAVEPOINT_PATH, savePointPath);
-        context.getIntpEventClient().updateParagraphConfig(
-                context.getNoteId(), context.getParagraphId(), config);
-        LOGGER.info("Job {} of paragraph {} is stopped with save point path: 
{}",
-                jobClient.getJobID(), context.getParagraphId(), savePointPath);
+        try {
+          String savePointPath = jobClient.stopWithSavepoint(true, 
savePointDir).get();
+          Map<String, String> config = new HashMap<>();
+          config.put(SAVEPOINT_PATH, savePointPath);
+          context.getIntpEventClient().updateParagraphConfig(
+                  context.getNoteId(), context.getParagraphId(), config);
+          LOGGER.info("Job {} of paragraph {} is stopped with save point path: 
{}",
+                  jobClient.getJobID(), context.getParagraphId(), 
savePointPath);
+        } catch (Exception e) {
+          LOGGER.warn("Fail to cancel job of paragraph {} with savepoint, try 
to cancel it without savepoint",
+                  context.getParagraphId(), e);
+          jobClient.cancel();
+        }
       }
       cancelled = true;
     } catch (Exception e) {