flink on yarn job status bug fixed. (apache#394)

* flink on yarn job status bug fixed apache#391
ambition119 · Oct 28, 2021 · 3ef8219 · 3ef8219
1 parent d9da5fc
commit 3ef8219
Show file tree

Hide file tree

Showing 6 changed files with 37 additions and 20 deletions.
diff --git a/...console-service/src/main/java/com/streamxhub/streamx/console/core/entity/Application.java b/...console-service/src/main/java/com/streamxhub/streamx/console/core/entity/Application.java
@@ -390,9 +390,12 @@ public AppInfo httpYarnAppInfo() throws Exception {
                 String url = String.format(format, HadoopUtils.getRMWebAppURL(false), appId);
                 return httpGetDoResult(url, AppInfo.class);
             } catch (IOException e) {
-                log.warn(e.getMessage());
-                String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId);
-                return httpGetDoResult(url, AppInfo.class);
+                try {
+                    String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId);
+                    return httpGetDoResult(url, AppInfo.class);
+                } catch (IOException e1) {
+                    throw e1;
+                }
             }
         }
         return null;
@@ -406,9 +409,12 @@ public JobsOverview httpJobsOverview() throws Exception {
                 String url = String.format(format, HadoopUtils.getRMWebAppURL(false), appId);
                 return httpGetDoResult(url, JobsOverview.class);
             } catch (IOException e) {
-                log.warn(e.getMessage());
-                String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId);
-                return httpGetDoResult(url, JobsOverview.class);
+                try {
+                    String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId);
+                    return httpGetDoResult(url, JobsOverview.class);
+                } catch (Exception e1) {
+                    throw e1;
+                }
             }
         }
         return null;
@@ -421,9 +427,12 @@ public Overview httpOverview() throws IOException {
             String url = String.format(format, HadoopUtils.getRMWebAppURL(false), appId);
             return httpGetDoResult(url, Overview.class);
         } catch (IOException e) {
-            log.warn(e.getMessage());
-            String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId);
-            return httpGetDoResult(url, Overview.class);
+            try {
+                String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId);
+                return httpGetDoResult(url, Overview.class);
+            } catch (Exception e1) {
+                throw e1;
+            }
         }
     }
 
@@ -434,9 +443,12 @@ public CheckPoints httpCheckpoints() throws IOException {
             String url = String.format(format, HadoopUtils.getRMWebAppURL(false), appId, jobId);
             return httpGetDoResult(url, CheckPoints.class);
         } catch (IOException e) {
-            log.warn(e.getMessage());
-            String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId, jobId);
-            return httpGetDoResult(url, CheckPoints.class);
+            try {
+                String url = String.format(format, HadoopUtils.getRMWebAppURL(true), appId, jobId);
+                return httpGetDoResult(url, CheckPoints.class);
+            } catch (Exception e1) {
+                throw e1;
+            }
         }
     }
 

diff --git a/...onsole-service/src/main/java/com/streamxhub/streamx/console/core/enums/FlinkAppState.java b/...onsole-service/src/main/java/com/streamxhub/streamx/console/core/enums/FlinkAppState.java
@@ -136,6 +136,10 @@ public enum FlinkAppState implements Serializable {
      */
     POS_TERMINATED (21),
 
+    /**
+     *  job SUCCEEDED on yarn
+     */
+    SUCCEEDED(22),
     /**
      * yarn 中检查到被killed
      */

diff --git a/...rc/main/java/com/streamxhub/streamx/console/core/service/impl/ApplicationServiceImpl.java b/...rc/main/java/com/streamxhub/streamx/console/core/service/impl/ApplicationServiceImpl.java
@@ -962,11 +962,8 @@ public void cancel(Application appParam) {
                 );
 
                 StopResponse stopResponse = FlinkSubmitHelper.stop(stopInfo);
-
-                assert stopResponse != null;
-
-                String savePointDir = stopResponse.savePointDir();
-                if (savePointDir != null) {
+                if (stopResponse != null && stopResponse.savePointDir() != null) {
+                    String savePointDir = stopResponse.savePointDir();
                     log.info("savePoint path:{}", savePointDir);
                     log.info("savePoint path:{}", savePointDir);
                     SavePoint savePoint = new SavePoint();

diff --git a/...ole-service/src/main/java/com/streamxhub/streamx/console/core/task/FlinkTrackingTask.java b/...ole-service/src/main/java/com/streamxhub/streamx/console/core/task/FlinkTrackingTask.java
@@ -525,6 +525,9 @@ private void getFromYarnRestApi(Application application, StopFrom stopFrom) thro
                         cleanSavepoint(application);
                         application.setEndTime(new Date());
                     }
+                    if(FlinkAppState.SUCCEEDED.equals(flinkAppState)) {
+                        flinkAppState = FlinkAppState.FINISHED;
+                    }
                     application.setState(flinkAppState.getValue());
                     //能运行到这一步,说明到YARN REST api中成功查询到信息
                     cleanOptioning(optionState, application.getId());

diff --git a/...-flink-proxy/src/main/scala/com/streamxhub/streamx/flink/proxy/ChildFirstClassLoader.java b/...-flink-proxy/src/main/scala/com/streamxhub/streamx/flink/proxy/ChildFirstClassLoader.java
@@ -211,4 +211,4 @@ private Class<?> loadClassWithoutExceptionHandling(String name, boolean resolve)
         return c;
     }
 
-}
+}
diff --git a/...ink-submit-api/src/main/scala/com/streamxhub/streamx/flink/submit/FlinkSubmitHelper.scala b/...ink-submit-api/src/main/scala/com/streamxhub/streamx/flink/submit/FlinkSubmitHelper.scala
@@ -61,8 +61,9 @@ object FlinkSubmitHelper extends Logger {
       val method = submitClass.getDeclaredMethod("stop", requestClass)
       method.setAccessible(true)
       val obj = method.invoke(null, FlinkShimsProxy.getObject(classLoader, stopRequest))
-      require(obj != null)
-      FlinkShimsProxy.getObject(this.getClass.getClassLoader, obj).asInstanceOf[StopResponse]
+      if (obj == null) null; else {
+        FlinkShimsProxy.getObject(this.getClass.getClassLoader, obj).asInstanceOf[StopResponse]
+      }
     })
   }
-Original file line number
+Diff line change
@@ Expand Up @@
             return c;
         }
-    }
+    }