Run suspension task from master-node.

components/org.wso2.carbon.identity.account.suspension.notification.task/src/main/java/org/wso2/carbon/identity/account/suspension/notification/task/AccountValidatorThread.java

@@ -18,6 +18,7 @@
 package org.wso2.carbon.identity.account.suspension.notification.task;
 
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.context.PrivilegedCarbonContext;
@@ -88,6 +89,11 @@ private void handleTask(String tenantDomain) {
             log.debug("Handling idle account suspension task for tenant: " + tenantDomain);
         }
 
+        // Run the task only from master node in cluster setup.
+        if (isMasterNodeExclusiveExecutionEnabled() && !isHazelcastMasterNode()) {
+            return;
+        }
+
         Property[] identityProperties;
         try {
             // Start Tenant flow
@@ -167,6 +173,27 @@ private void handleTask(String tenantDomain) {
         }
     }
 
+    /**
+     * Check whether suspension task configured to run only in master node or not.
+     *
+     * @return true or false based on the deployment config.
+     */
+    private boolean isMasterNodeExclusiveExecutionEnabled() {
+
+        String clusterModeEnabledValue = IdentityUtil.getProperty(NotificationConstants.EXECUTE_TASK_IN_MASTER_NODE);
+        return StringUtils.isNotBlank(clusterModeEnabledValue) ? Boolean.parseBoolean(clusterModeEnabledValue) : false;
+    }
+
+    /**
+     * Check whether current node is master node in the Hazelcast cluster.
+     *
+     * @return true or false based on the Hazelcast cluster.
+     */
+    private boolean isHazelcastMasterNode() {
+
+        return NotificationTaskDataHolder.getInstance().getClusteringAgent().isCoordinator();
+    }
+
     /**
      * Notify users about account inactivity via Email.
      */

components/org.wso2.carbon.identity.account.suspension.notification.task/src/main/java/org/wso2/carbon/identity/account/suspension/notification/task/internal/NotificationTaskDataHolder.java

@@ -16,6 +16,7 @@
 
 package org.wso2.carbon.identity.account.suspension.notification.task.internal;
 
+import org.apache.axis2.clustering.ClusteringAgent;
 import org.osgi.framework.BundleContext;
 import org.wso2.carbon.identity.account.suspension.notification.task.NotificationReceiversRetrievalFactory;
 import org.wso2.carbon.identity.account.suspension.notification.task.util.NotificationConstants;
@@ -43,6 +44,7 @@ public class NotificationTaskDataHolder {
     private String notificationTriggerTime;
     private String schedulerDelay;
     private String notificationSendingThreadPoolSize = "1";
+    private ClusteringAgent clusteringAgent;
 
     public int getNotificationSendingThreadPoolSize() {
         return Integer.parseInt(notificationSendingThreadPoolSize);
@@ -112,4 +114,14 @@ public void setRealmService(RealmService realmService) {
     public RealmService getRealmService() {
         return realmService;
     }
+
+    public ClusteringAgent getClusteringAgent() {
+
+        return clusteringAgent;
+    }
+
+    public void setClusteringAgent(ClusteringAgent clusteringAgent) {
+
+        this.clusteringAgent = clusteringAgent;
+    }
 }

components/org.wso2.carbon.identity.account.suspension.notification.task/src/main/java/org/wso2/carbon/identity/account/suspension/notification/task/internal/NotificationTaskServiceComponent.java

@@ -36,6 +36,7 @@
 import org.osgi.service.component.annotations.Reference;
 import org.osgi.service.component.annotations.ReferenceCardinality;
 import org.osgi.service.component.annotations.ReferencePolicy;
+import org.wso2.carbon.utils.ConfigurationContextService;
 
 /**
  * Notification scheduler. Check for users who requires a notification for relogin
@@ -156,5 +157,20 @@ protected void unsetRealmService(RealmService realmService) {
             log.debug("RealmService is unset in the Application Authentication Framework bundle");
         }
     }
+
+    @Reference(
+            name = "config.context.service",
+            cardinality = ReferenceCardinality.OPTIONAL,
+            policy = ReferencePolicy.DYNAMIC,
+            unbind = "unsetClusteringAgent")
+    protected void setClusteringAgent(ConfigurationContextService configurationContextService) {
+
+        NotificationTaskDataHolder.getInstance().setClusteringAgent(
+                configurationContextService.getServerConfigContext().getAxisConfiguration().getClusteringAgent());
+    }
+
+    protected void unsetClusteringAgent(ConfigurationContextService configurationContextService) {
+        NotificationTaskDataHolder.getInstance().setClusteringAgent(null);
+    }
 }
 

components/org.wso2.carbon.identity.account.suspension.notification.task/src/main/java/org/wso2/carbon/identity/account/suspension/notification/task/util/NotificationConstants.java

@@ -27,6 +27,7 @@ public class NotificationConstants {
     public static final String SUSPENSION_NOTIFICATION_TRIGGER_TIME= "suspension.notification.trigger.time";
     public static final String SUSPENSION_NOTIFICATION_DELAYS="suspension.notification.delays";
     public static final String USE_IDENTITY_CLAIM_FOR_LAST_LOGIN_TIME = "AccountSuspension.UseIdentityClaims";
+    public static final String EXECUTE_TASK_IN_MASTER_NODE = "AccountSuspension.ExecuteTaskOnMasterNode";
     public static final String TRIGGER_TIME_FORMAT = "HH:mm:ss";
     public static final long SCHEDULER_DELAY = 24; // In hours
     public static final String SUSPENSION_NOTIFICATION_THREAD_POOL_SIZE = "suspension.notification.thread.pool.size";