[FixBug] Fix Exception on Convert.toInt of YarnGateway to track the new JobManager address in flink1.20 (#4382)

cantiandashu · web-flow · commit 136d56b73a85 · 2025-09-11T13:18:18.000+08:00
diff --git a/dinky-client/dinky-client-1.14/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.14/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
 import org.apache.flink.table.catalog.Column;
@@ -31,6 +33,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -93,4 +97,8 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+    }
 }
diff --git a/dinky-client/dinky-client-1.15/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.15/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -33,6 +35,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -86,4 +90,8 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+    }
 }
diff --git a/dinky-client/dinky-client-1.16/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.16/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -33,6 +35,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -86,4 +90,8 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+    }
 }
diff --git a/dinky-client/dinky-client-1.17/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.17/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -33,6 +35,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -86,4 +90,8 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+    }
 }
diff --git a/dinky-client/dinky-client-1.18/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.18/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -33,6 +35,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -86,4 +90,8 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+    }
 }
diff --git a/dinky-client/dinky-client-1.19/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.19/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -33,6 +35,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -86,4 +90,8 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+    }
 }
diff --git a/dinky-client/dinky-client-1.20/src/main/java/org/dinky/utils/FlinkUtil.java b/dinky-client/dinky-client-1.20/src/main/java/org/dinky/utils/FlinkUtil.java
@@ -21,6 +21,8 @@
 
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.table.catalog.CatalogManager;
@@ -33,6 +35,8 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 
+import cn.hutool.core.convert.Convert;
+
 /**
  * FlinkUtil
  *
@@ -86,4 +90,11 @@ public static String cancelWithSavepoint(ClusterClient clusterClient, String job
                 .get()
                 .toString();
     }
+
+    public static int getZookeeperSessionTimeout(Configuration configuration) {
+        return Convert.toInt(configuration
+                        .get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT)
+                        .getSeconds()
+                * 1000);
+    }
 }
diff --git a/dinky-gateway/src/main/java/org/dinky/gateway/yarn/YarnGateway.java b/dinky-gateway/src/main/java/org/dinky/gateway/yarn/YarnGateway.java
@@ -39,6 +39,7 @@
 import org.dinky.gateway.result.YarnResult;
 import org.dinky.gateway.utils.RequestKerberosUrlUtils;
 import org.dinky.utils.FlinkJsonUtil;
+import org.dinky.utils.FlinkUtil;
 import org.dinky.utils.ThreadUtil;
 
 import org.apache.commons.lang3.StringUtils;
@@ -94,7 +95,6 @@
 
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.convert.Convert;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.lang.Assert;
 import cn.hutool.core.util.ReUtil;
@@ -513,7 +513,7 @@ public String getLatestJobManageHost(String appId, String oldJobManagerHost) {
                         + HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM.key()
                         + "'.");
             }
-            int sessionTimeout = Convert.toInt(configuration.get(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT));
+            int sessionTimeout = FlinkUtil.getZookeeperSessionTimeout(configuration);
             String root = configuration.getValue(HighAvailabilityOptions.HA_ZOOKEEPER_ROOT);
             String namespace = configuration.getValue(HighAvailabilityOptions.HA_CLUSTER_ID);
 
