Merge pull request #4447 from KalinduGandara/issue_4399

Fix Message processor duplication when coordination DB delay happen

Author: KalinduGandara

components/mediation/tasks/org.wso2.micro.integrator.ntask.core/src/main/java/org/wso2/micro/integrator/ntask/core/impl/AbstractQuartzTaskManager.java

@@ -19,6 +19,9 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.commons.util.MiscellaneousUtil;
+import org.apache.synapse.core.SynapseEnvironment;
+import org.apache.synapse.message.processor.MessageProcessor;
 import org.quartz.CronScheduleBuilder;
 import org.quartz.Job;
 import org.quartz.JobBuilder;
@@ -37,6 +40,7 @@
 import org.quartz.TriggerListener;
 import org.quartz.impl.matchers.GroupMatcher;
 import org.quartz.spi.OperableTrigger;
+import org.wso2.micro.integrator.core.util.MicroIntegratorBaseUtils;
 import org.wso2.micro.integrator.ntask.common.TaskConstants;
 import org.wso2.micro.integrator.ntask.common.TaskException;
 import org.wso2.micro.integrator.ntask.coordination.TaskCoordinationException;
@@ -179,6 +183,15 @@ protected synchronized void pauseLocalTaskTemporarily(String taskName) throws Ta
             if (null != listener) {
                 listener.notifyLocalTaskPause(taskName);
             }
+            if (MiscellaneousUtil.isTaskOfMessageProcessor(taskName)) {
+                SynapseEnvironment synapseEnvironment = MicroIntegratorBaseUtils.getSynapseEnvironment();
+                String messageProcessorName = MiscellaneousUtil.getMessageProcessorName(taskName);
+                MessageProcessor messageProcessor = synapseEnvironment.getSynapseConfiguration()
+                        .getMessageProcessors().get(messageProcessorName);
+                if (messageProcessor != null) {
+                    messageProcessor.pauseMessageProcessorTemporarily();
+                }
+            }
             log.info("Task temporarily paused: [" + this.getTaskType() + "][" + taskName + "]");
         } catch (SchedulerException e) {
             throw new TaskException("Error in temporarily pausing task with name: " + taskName,