Fix issues with unnecessary task deletion

components/task-mgt/task-manager/io.entgra.task.mgt.core/src/main/java/io/entgra/task/mgt/core/util/TaskManagementUtil.java

@@ -59,8 +59,7 @@ public class TaskManagementUtil {
         try {
             int serverHashIdx = TaskManagerDataHolder.getInstance().getHeartBeatService()
                     .getServerCtxInfo().getLocalServerHashIdx();
-            return TaskMgtConstants.Task.DYNAMIC_TASK_TYPE + TaskMgtConstants.Task.NAME_SEPARATOR + dynamicTaskId
-                    + TaskMgtConstants.Task.NAME_SEPARATOR + serverHashIdx;
+            return generateTaskId(dynamicTaskId, serverHashIdx);
         } catch (HeartBeatManagementException e) {
             String msg = "Failed to generate task id for a dynamic task " + dynamicTaskId;
             log.error(msg, e);
@@ -68,6 +67,11 @@ public class TaskManagementUtil {
         }
     }
 
+    public static String generateTaskId(int dynamicTaskId, int serverHashIdx) {
+        return TaskMgtConstants.Task.DYNAMIC_TASK_TYPE + TaskMgtConstants.Task.NAME_SEPARATOR + dynamicTaskId
+                + TaskMgtConstants.Task.NAME_SEPARATOR + serverHashIdx;
+    }
+
     public static String generateTaskPropsMD5(Map<String, String> taskProperties) throws TaskManagementException {
         taskProperties.remove(TaskMgtConstants.Task.TENANT_ID_PROP);
         taskProperties.remove(TaskMgtConstants.Task.LOCAL_HASH_INDEX);

components/task-mgt/task-watcher/io.entgra.task.mgt.watcher/src/main/java/io/entgra/task/mgt/watcher/IoTSStartupHandler.java

@@ -41,6 +41,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
 
@@ -70,7 +71,6 @@ public class IoTSStartupHandler implements ServerStartupObserver {
         if (log.isDebugEnabled()) {
             log.debug("Comparing Tasks from carbon nTask manager and entgra task manager");
         }
-        TaskManager taskManager;
         TaskService nTaskService = TaskWatcherDataHolder.getInstance().getnTaskService();
         if (nTaskService == null) {
             String msg = "Unable to load TaskService from the carbon nTask core";
@@ -80,6 +80,25 @@ public class IoTSStartupHandler implements ServerStartupObserver {
         try {
             List<DynamicTask> dynamicTasks = TaskWatcherDataHolder.getInstance().getTaskManagementService()
                     .getAllDynamicTasks();
+
+            scheduleMissingTasks(nTaskService, dynamicTasks);
+            deleteObsoleteTasks(nTaskService, dynamicTasks);
+
+            if (log.isDebugEnabled()) {
+                log.debug("Task Comparison Completed and all tasks in current node are updated");
+            }
+        } catch (TaskException e) {
+            String msg = "Error occurred while accessing carbon nTask manager.";
+            log.error(msg, e);
+        } catch (TaskManagementException e) {
+            String msg = "Error occurred while retrieving all active tasks from entgra task manager";
+            log.error(msg, e);
+        }
+
+    }
+
+    private static void scheduleMissingTasks(TaskService nTaskService, List<DynamicTask> dynamicTasks)
+            throws TaskException, TaskManagementException {
         Map<Integer, List<DynamicTask>> tenantedDynamicTasks = new HashMap<>();
         List<DynamicTask> dts;
         for (DynamicTask dt : dynamicTasks) {
@@ -91,6 +110,7 @@ public class IoTSStartupHandler implements ServerStartupObserver {
             dts.add(dt);
             tenantedDynamicTasks.put(dt.getTenantId(), dts);
         }
+        TaskManager taskManager;
         for (Integer tenantId : tenantedDynamicTasks.keySet()) {
             if (tenantId == -1) {
                 log.warn("Found " + tenantedDynamicTasks.get(tenantId).size() +
@@ -166,6 +186,10 @@ public class IoTSStartupHandler implements ServerStartupObserver {
             }
             PrivilegedCarbonContext.endTenantFlow();
         }
+    }
+
+    private static void deleteObsoleteTasks(TaskService nTaskService, List<DynamicTask> dynamicTasks)
+            throws TaskManagementException, TaskException {
 
         List<Tenant> tenants = new ArrayList<>();
         try {
@@ -183,6 +207,16 @@ public class IoTSStartupHandler implements ServerStartupObserver {
             return;
         }
 
+        TaskManager taskManager;
+        Set<Integer> hashIds;
+        try {
+            hashIds = TaskWatcherDataHolder.getInstance().getHeartBeatService().getActiveServers().keySet();
+        } catch (HeartBeatManagementException e) {
+            String msg = "Unexpected exception when getting hash indexes of active servers";
+            log.error(msg, e);
+            throw new TaskManagementException(msg, e);
+        }
+
         for (Tenant tenant : tenants) {
             PrivilegedCarbonContext.startTenantFlow();
             PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantId(tenant.getId(), true);
@@ -195,12 +229,17 @@ public class IoTSStartupHandler implements ServerStartupObserver {
             for (TaskInfo taskInfo : tasks) {
                 boolean isExist = false;
                 for (DynamicTask dt : dynamicTasks) {
+                    for (int hid : hashIds) {
                         if (tenant.getId() == dt.getTenantId() &&
-                                taskInfo.getName().equals(TaskManagementUtil.generateTaskId(dt.getDynamicTaskId()))) {
+                                taskInfo.getName().equals(TaskManagementUtil.generateTaskId(dt.getDynamicTaskId(), hid))) {
                             isExist = true;
                             break;
                         }
                     }
+                    if (isExist) {
+                        break;
+                    }
+                }
                 if (!isExist) {
                     taskManager.deleteTask(taskInfo.getName());
                     if (log.isDebugEnabled()) {
@@ -210,17 +249,6 @@ public class IoTSStartupHandler implements ServerStartupObserver {
             }
             PrivilegedCarbonContext.endTenantFlow();
         }
-
-            if (log.isDebugEnabled()) {
-                log.debug("Task Comparison Completed and all tasks in current node are updated");
-            }
-        } catch (TaskException e) {
-            String msg = "Error occurred while accessing carbon nTask manager.";
-            log.error(msg, e);
-        } catch (TaskManagementException e) {
-            String msg = "Error occurred while retrieving all active tasks from entgra task manager";
-            log.error(msg, e);
     }
 
-    }
 }