Fix the missing cluster_spec due to the null when getting clusterSpec…

… in task executor (#663)

Signed-off-by: zhangjunfan <junfan.zhang@outlook.com>

tony-core/src/main/java/com/linkedin/tony/ApplicationMaster.java

@@ -912,7 +912,7 @@ public String getClusterSpec(String taskId) throws IOException {
       if (amRuntimeAdapter.canStartTask(distributedMode, taskId)) {
         return amRuntimeAdapter.constructClusterSpec(taskId);
       }
-      return null;
+      return StringUtils.EMPTY;
     }
 
     @Override

tony-core/src/main/java/com/linkedin/tony/TaskExecutor.java

@@ -334,7 +334,13 @@ private String registerAndGetClusterSpec() throws IOException {
       throw new IOException("Errors on registering to AM, maybe due to the network failure.");
     }
 
-    return Utils.pollForeverTillNonNull(() -> proxy.getClusterSpec(taskId), DEFAULT_REQUEST_POLL_INTERVAL);
+    return Utils.pollTillConditionReached(
+            () -> proxy.getClusterSpec(taskId),
+            x -> StringUtils.isNotEmpty(x),
+            () -> null,
+            DEFAULT_REQUEST_POLL_INTERVAL,
+            0
+    );
   }
 
   public void callbackInfoToAM(String taskId, String callbackInfo) throws IOException {

tony-core/src/main/java/com/linkedin/tony/util/Utils.java

@@ -118,10 +118,6 @@ public static <T> T pollTillNonNull(Callable<T> func, int interval, int timeout)
     return pollTillConditionReached(func, Objects::nonNull, () -> null, interval, timeout);
   }
 
-  public static <T> T pollForeverTillNonNull(Callable<T> func, int interval) {
-    return pollTillNonNull(func, interval, 0);
-  }
-
   public static <T> T pollTillConditionReached(Callable<T> callFunc, Function<T, Boolean> conditionFunc,
           CallableWithoutException<T> defaultReturnedFunc, int interval, int timeout) {
     Preconditions.checkArgument(interval >= 0, "Interval must be non-negative.");
@@ -133,16 +129,16 @@ public static <T> T pollTillConditionReached(Callable<T> callFunc, Function<T, B
       while (timeout == 0 || remainingTime >= 0) {
         ret = callFunc.call();
         if (conditionFunc.apply(ret)) {
-          LOG.info("pollTillNonNull function finished within " + timeout + " seconds");
+          LOG.info("pollTillConditionReached function finished within " + timeout + " seconds");
           return ret;
         }
         Thread.sleep(interval * 1000);
         remainingTime -= interval;
       }
     } catch (Exception e) {
-      LOG.error("pollTillNonNull function threw exception", e);
+      LOG.error("pollTillConditionReached function threw exception", e);
     }
-    LOG.warn("Function didn't return non-null within " + timeout + " seconds.");
+    LOG.warn("Function didn't satisfy applied condition within " + timeout + " seconds.");
     return defaultReturnedFunc.call();
   }