Repository: wso2/micro-integrator
Issue: JMS Topic subscription is not removed when node looses connection to the cluster.

**Description:**
When using JMS inbound endpoint in a clustered scenario, the JMS topic subscription is not removed when the connection to the cluster is lost. Ideally the subscription should be removed as we are stopping the underlying running task and need to be re established when the task is resumed again. 

Changed files:
components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/AbstractQuartzTaskManager.java
components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/jms/JMSTask.java
components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/LocalTaskActionListener.java

diff --git a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/jms/JMSTask.java b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/jms/JMSTask.java
index 18520cc59..ec2bf6262 100644
--- a/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/jms/JMSTask.java
+++ b/components/mediation/inbound-endpoints/org.wso2.micro.integrator.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/jms/JMSTask.java
@@ -63,13 +63,15 @@ public class JMSTask extends InboundTask implements LocalTaskActionListener {
     /**
      * {@inheritDoc}
      * <p>
-     * Destroys the JMS task upon deletion of the local task.
+     * Destroys the JMS task upon removal of the local task.
      *
      * @param taskName the name of the task that was deleted
      */
     @Override
-    public void notifyLocalTaskDeletion(String taskName) {
+    public void notifyLocalTaskRemoval(String taskName) {
         destroy();
-        logger.debug("Destroyed JMS task due to deletion of task: " + taskName);
+        if (logger.isDebugEnabled()) {
+            logger.debug("Destroyed JMS task due to deletion of task: " + taskName);
+        }
     }
 }
diff --git a/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/AbstractQuartzTaskManager.java b/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/AbstractQuartzTaskManager.java
index 19e6af510..69281ced6 100644
--- a/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/AbstractQuartzTaskManager.java
+++ b/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/AbstractQuartzTaskManager.java
@@ -145,7 +145,7 @@ public abstract class AbstractQuartzTaskManager implements TaskManager {
                 //notify the listeners of the task deletion
                 LocalTaskActionListener listener = localTaskActionListeners.get(taskName);
                 if (null != listener) {
-                    listener.notifyLocalTaskDeletion(taskName);
+                    listener.notifyLocalTaskRemoval(taskName);
                 }
             }
         } catch (SchedulerException e) {
@@ -159,6 +159,11 @@ public abstract class AbstractQuartzTaskManager implements TaskManager {
         String taskGroup = this.getTenantTaskGroup();
         try {
             this.getScheduler().pauseJob(new JobKey(taskName, taskGroup));
+            //notify the listeners of the task pause
+            LocalTaskActionListener listener = localTaskActionListeners.get(taskName);
+            if (null != listener) {
+                listener.notifyLocalTaskRemoval(taskName);
+            }
             log.info("Task paused: [" + this.getTaskType() + "][" + taskName + "]");
         } catch (SchedulerException e) {
             throw new TaskException("Error in pausing task with name: " + taskName, TaskException.Code.UNKNOWN, e);
diff --git a/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/LocalTaskActionListener.java b/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/LocalTaskActionListener.java
index e55ac9e30..04e8e8e77 100644
--- a/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/LocalTaskActionListener.java
+++ b/components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/LocalTaskActionListener.java
@@ -27,9 +27,9 @@ import org.wso2.micro.integrator.ntask.core.TaskManager;
 public interface LocalTaskActionListener {
 
     /**
-     * Method to notify when a local task is deleted.
+     * Method to notify when a local task is removed, it can be due to pause or delete.
      *
      * @param taskName
      */
-    void notifyLocalTaskDeletion(String taskName);
+    void notifyLocalTaskRemoval(String taskName);
 }