优化 PerJob 和 Application 作业的JID提交检测

TagawaAkirayo · Mar 9, 2022 · 0bef2d4 · 0bef2d4
1 parent be15074
commit 0bef2d4
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 2 deletions.
diff --git a/dlink-daemon/src/main/java/com/dlink/daemon/entity/TaskWorker.java b/dlink-daemon/src/main/java/com/dlink/daemon/entity/TaskWorker.java
@@ -17,7 +17,7 @@ public TaskWorker(TaskQueue queue) {
 
     @Override
     public void run() {
-        log.info("TaskWorker run");
+//        log.info("TaskWorker run");
         while (running) {
             DaemonTask daemonTask = queue.dequeue();
             if (daemonTask != null) {
@@ -31,7 +31,7 @@ public void run() {
     }
 
     public void shutdown() {
-        log.info(Thread.currentThread().getName() + "TaskWorker shutdown");
+//        log.info(Thread.currentThread().getName() + "TaskWorker shutdown");
         running = false;
     }
 }
diff --git a/dlink-gateway/src/main/java/com/dlink/gateway/kubernetes/KubernetesApplicationGateway.java b/dlink-gateway/src/main/java/com/dlink/gateway/kubernetes/KubernetesApplicationGateway.java
@@ -53,6 +53,15 @@ public GatewayResult submitJar() {
             ClusterClientProvider<String> clusterClientProvider = kubernetesClusterDescriptor.deployApplicationCluster(clusterSpecification, applicationConfiguration);
             ClusterClient<String> clusterClient = clusterClientProvider.getClusterClient();
             Collection<JobStatusMessage> jobStatusMessages = clusterClient.listJobs().get();
+            int counts = 10;
+            while (jobStatusMessages.size() == 0 && counts > 0) {
+                Thread.sleep(1000);
+                counts--;
+                jobStatusMessages = clusterClient.listJobs().get();
+                if (jobStatusMessages.size() > 0) {
+                    break;
+                }
+            }
             if (jobStatusMessages.size() > 0) {
                 List<String> jids = new ArrayList<>();
                 for (JobStatusMessage jobStatusMessage : jobStatusMessages) {

diff --git a/dlink-gateway/src/main/java/com/dlink/gateway/yarn/YarnApplicationGateway.java b/dlink-gateway/src/main/java/com/dlink/gateway/yarn/YarnApplicationGateway.java
@@ -8,6 +8,7 @@
 import com.dlink.gateway.result.GatewayResult;
 import com.dlink.gateway.result.YarnResult;
 import com.dlink.utils.LogUtil;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.client.deployment.ClusterSpecification;
 import org.apache.flink.client.deployment.application.ApplicationConfiguration;
 import org.apache.flink.client.program.ClusterClient;
@@ -67,12 +68,22 @@ public GatewayResult submitJar() {
                     applicationConfiguration);
             ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();
             Collection<JobStatusMessage> jobStatusMessages = clusterClient.listJobs().get();
+            int counts = 10;
+            while (jobStatusMessages.size() == 0 && counts > 0) {
+                Thread.sleep(1000);
+                counts--;
+                jobStatusMessages = clusterClient.listJobs().get();
+                if (jobStatusMessages.size() > 0) {
+                    break;
+                }
+            }
             if (jobStatusMessages.size() > 0) {
                 List<String> jids = new ArrayList<>();
                 for (JobStatusMessage jobStatusMessage : jobStatusMessages) {
                     jids.add(jobStatusMessage.getJobId().toHexString());
                 }
                 result.setJids(jids);
+                logger.info("JIDS =" + StringUtils.join(jids, ","));
             }
             ApplicationId applicationId = clusterClient.getClusterId();
             result.setAppId(applicationId.toString());

diff --git a/dlink-gateway/src/main/java/com/dlink/gateway/yarn/YarnPerJobGateway.java b/dlink-gateway/src/main/java/com/dlink/gateway/yarn/YarnPerJobGateway.java
@@ -56,6 +56,15 @@ public GatewayResult submitJobGraph(JobGraph jobGraph) {
             result.setAppId(applicationId.toString());
             result.setWebURL(clusterClient.getWebInterfaceURL());
             Collection<JobStatusMessage> jobStatusMessages = clusterClient.listJobs().get();
+            int counts = 10;
+            while (jobStatusMessages.size() == 0 && counts > 0) {
+                Thread.sleep(1000);
+                counts--;
+                jobStatusMessages = clusterClient.listJobs().get();
+                if (jobStatusMessages.size() > 0) {
+                    break;
+                }
+            }
             if (jobStatusMessages.size() > 0) {
                 List<String> jids = new ArrayList<>();
                 for (JobStatusMessage jobStatusMessage : jobStatusMessages) {

diff --git a/dlink-web/src/pages/Welcome.tsx b/dlink-web/src/pages/Welcome.tsx
@@ -710,6 +710,9 @@ export default (): React.ReactNode => {
                 <li>
                   <Link>优化 K8S Application 提交配置</Link>
                 </li>
+                <li>
+                  <Link>优化 PerJob 和 Application 作业的JID提交检测</Link>
+                </li>
               </ul>
             </Paragraph>
           </Timeline.Item>