ITaskQueue should have "hasTask" function (#1744)

* issue https://github.com/apache/incubator-dolphinscheduler/issues/1742

* remove unnecessary catch

* fix exception type

* fix bad code smell

* remove log exceptions

* catch exception

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/queue/ITaskQueue.java

@@ -30,6 +30,13 @@ public interface ITaskQueue {
      */
     List<String> getAllTasks(String key);
 
+    /**
+     * check if has a task
+     * @param key queue name
+     * @return true if has; false if not
+     */
+    boolean hasTask(String key);
+
     /**
      * check task exists in the task queue or not
      *

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/queue/TaskQueueZkImpl.java

@@ -75,6 +75,21 @@ public class TaskQueueZkImpl implements ITaskQueue {
         return new ArrayList<>();
     }
 
+    /**
+     * check if has a task
+     * @param key queue name
+     * @return true if has; false if not
+     */
+    @Override
+    public boolean hasTask(String key) {
+        try {
+            return zookeeperOperator.hasChildren(key);
+        } catch (Exception e) {
+            logger.error("check has task in tasks queue exception",e);
+        }
+        return false;
+    }
+
     /**
      * check task exists in the task queue or not
      *

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/zk/ZookeeperOperator.java

@@ -27,6 +27,7 @@ import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
@@ -139,6 +140,16 @@ public class ZookeeperOperator implements InitializingBean {
         }
     }
 
+    public boolean hasChildren(final String key){
+        Stat stat ;
+        try {
+            stat = zkClient.checkExists().forPath(key);
+            return stat.getNumChildren() >= 1;
+        } catch (Exception ex) {
+            throw new IllegalStateException(ex);
+        }
+    }
+
     public boolean isExisted(final String key) {
         try {
             return zkClient.checkExists().forPath(key) != null;

dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/queue/TaskQueueZKImplTest.java

@@ -64,7 +64,16 @@ public class TaskQueueZKImplTest extends BaseTaskQueueTest  {
         allTasks = tasksQueue.getAllTasks(Constants.DOLPHINSCHEDULER_TASKS_QUEUE);
         assertEquals(allTasks.size(),0);
     }
-
+    @Test
+    public void hasTask(){
+        init();
+        boolean hasTask = tasksQueue.hasTask(Constants.DOLPHINSCHEDULER_TASKS_QUEUE);
+        assertTrue(hasTask);
+        //delete all
+        tasksQueue.delete();
+        hasTask = tasksQueue.hasTask(Constants.DOLPHINSCHEDULER_TASKS_QUEUE);
+        assertFalse(hasTask);
+    }
     /**
      * test check task exists in the task queue or not
      */

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/FetchTaskThread.java

@@ -150,8 +150,9 @@ public class FetchTaskThread implements Runnable{
                 }
 
                 //whether have tasks, if no tasks , no need lock  //get all tasks
-                List<String> tasksQueueList = taskQueue.getAllTasks(Constants.DOLPHINSCHEDULER_TASKS_QUEUE);
-                if (CollectionUtils.isEmpty(tasksQueueList)){
+                boolean hasTask = taskQueue.hasTask(Constants.DOLPHINSCHEDULER_TASKS_QUEUE);
+
+                if (!hasTask){
                     Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                     continue;
                 }