added cluster support for Scheduler

mifosng-db/migrations/core_db/V103__cluster_support_for_batch_jobs.sql

@@ -0,0 +1,14 @@
+ALTER TABLE `job`
+	ADD COLUMN `is_misfired` TINYINT(1) NOT NULL DEFAULT '0' AFTER `scheduler_group`;
+	
+CREATE TABLE `scheduler_detail` (
+	`id` SMALLINT(2) NOT NULL AUTO_INCREMENT,
+	`is_suspended` TINYINT(1) NOT NULL DEFAULT '0',
+	`execute_misfired_jobs` TINYINT(1) NOT NULL DEFAULT '1',
+	`reset_scheduler_on_bootup` TINYINT(1) NOT NULL DEFAULT '1',
+	PRIMARY KEY (`id`)
+)
+COLLATE='utf8_general_ci'
+ENGINE=InnoDB;
+
+INSERT INTO `scheduler_detail` (`is_suspended`, `execute_misfired_jobs`, `reset_scheduler_on_bootup`) VALUES (FALSE,TRUE,TRUE);

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java

@@ -28,6 +28,7 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
     @Column(name = "cron_expression")
     private String cronExpression;
 
+    @SuppressWarnings("unused")
     @Column(name = "create_time")
     @Temporal(TemporalType.TIMESTAMP)
     private Date createTime;
@@ -38,6 +39,7 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
     @Column(name = "group_name")
     private String groupName;
 
+    @SuppressWarnings("unused")
     @Column(name = "previous_run_start_time")
     @Temporal(TemporalType.TIMESTAMP)
     private Date previousRunStartTime;
@@ -49,6 +51,7 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
     @Column(name = "job_key")
     private String jobKey;
 
+    @SuppressWarnings("unused")
     @Column(name = "initializing_errorlog")
     private String errorLog;
 
@@ -58,12 +61,16 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
     @Column(name = "currently_running")
     private boolean currentlyRunning;
 
+    @SuppressWarnings("unused")
     @Column(name = "updates_allowed")
     private boolean updatesAllowed;
 
     @Column(name = "scheduler_group")
     private Short schedulerGroup;
 
+    @Column(name = "is_misfired")
+    private boolean triggerMisfired;
+
     protected ScheduledJobDetail() {
 
     }
@@ -104,6 +111,10 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
         this.previousRunStartTime = previousRunStartTime;
     }
 
+    public Date getNextRunTime() {
+        return this.nextRunTime;
+    }
+
     public void updateNextRunTime(final Date nextRunTime) {
         this.nextRunTime = nextRunTime;
     }
@@ -147,4 +158,12 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
         return actualChanges;
     }
 
+    public boolean isTriggerMisfired() {
+        return this.triggerMisfired;
+    }
+
+    public void updateTriggerMisfired(boolean triggerMisfired) {
+        this.triggerMisfired = triggerMisfired;
+    }
+
 }

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetailRepository.java

@@ -1,7 +1,10 @@
 package org.mifosplatform.infrastructure.jobs.domain;
 
+import javax.persistence.LockModeType;
+
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+import org.springframework.data.jpa.repository.Lock;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
 
@@ -14,4 +17,9 @@ public interface ScheduledJobDetailRepository extends JpaRepository<ScheduledJob
     @Query("from ScheduledJobDetail jobDetail where jobDetail.id=:jobId")
     ScheduledJobDetail findByJobId(@Param("jobId") Long jobId);
 
+    
+    @Lock(value=LockModeType.PESSIMISTIC_WRITE)
+    @Query("from ScheduledJobDetail jobDetail where jobDetail.jobKey = :jobKey")
+    ScheduledJobDetail findByJobKeyWithLock(@Param("jobKey") String jobKey);
+
 }

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobRunHistory.java

@@ -18,29 +18,37 @@ public class ScheduledJobRunHistory extends AbstractPersistable<Long> {
 
     @ManyToOne
     @JoinColumn(name = "job_id")
+    @SuppressWarnings("unused")
     private ScheduledJobDetail scheduledJobDetail;
 
     @Column(name = "version")
+    @SuppressWarnings("unused")
     private Long version;
 
     @Column(name = "start_time")
     @Temporal(TemporalType.TIMESTAMP)
+    @SuppressWarnings("unused")
     private Date startTime;
 
     @Column(name = "end_time")
     @Temporal(TemporalType.TIMESTAMP)
+    @SuppressWarnings("unused")
     private Date endTime;
 
     @Column(name = "status")
+    @SuppressWarnings("unused")
     private String status;
 
     @Column(name = "error_message")
+    @SuppressWarnings("unused")
     private String errorMessage;
 
     @Column(name = "trigger_type")
+    @SuppressWarnings("unused")
     private String triggerType;
 
     @Column(name = "error_log")
+    @SuppressWarnings("unused")
     private String errorLog;
 
     public ScheduledJobRunHistory() {

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetail.java

@@ -0,0 +1,49 @@
+package org.mifosplatform.infrastructure.jobs.domain;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Table;
+
+import org.springframework.data.jpa.domain.AbstractPersistable;
+
+@Entity
+@Table(name = "scheduler_detail")
+public class SchedulerDetail extends AbstractPersistable<Long> {
+
+    @Column(name = "execute_misfired_jobs")
+    private boolean executeInstructionForMisfiredJobs;
+
+    @Column(name = "is_suspended")
+    private boolean suspended;
+
+    @Column(name = "reset_scheduler_on_bootup")
+    private boolean resetSchedulerOnBootup;
+
+    protected SchedulerDetail() {
+
+    }
+
+    public boolean isExecuteInstructionForMisfiredJobs() {
+        return this.executeInstructionForMisfiredJobs;
+    }
+
+    public void updateExecuteInstructionForMisfiredJobs(boolean executeInstructionForMisfiredJobs) {
+        this.executeInstructionForMisfiredJobs = executeInstructionForMisfiredJobs;
+    }
+
+    public boolean isSuspended() {
+        return this.suspended;
+    }
+
+    public void updateSuspendedState(boolean suspended) {
+        this.suspended = suspended;
+    }
+
+    public boolean isResetSchedulerOnBootup() {
+        return this.resetSchedulerOnBootup;
+    }
+
+    public void updateResetSchedulerOnBootup(boolean resetSchedulerOnBootup) {
+        this.resetSchedulerOnBootup = resetSchedulerOnBootup;
+    }
+}

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetailRepository.java

@@ -0,0 +1,8 @@
+package org.mifosplatform.infrastructure.jobs.domain;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+
+public interface SchedulerDetailRepository extends JpaRepository<SchedulerDetail, Long>, JpaSpecificationExecutor<SchedulerDetail> {
+
+}

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java

@@ -4,7 +4,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Properties;
 import java.util.TimeZone;
 
@@ -15,6 +14,7 @@ import org.mifosplatform.infrastructure.core.exception.PlatformInternalServerExc
 import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil;
 import org.mifosplatform.infrastructure.jobs.annotation.CronMethodParser;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail;
+import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetail;
 import org.mifosplatform.infrastructure.jobs.exception.JobNotFoundException;
 import org.mifosplatform.infrastructure.security.service.TenantDetailsService;
 import org.quartz.JobDataMap;
@@ -24,6 +24,7 @@ import org.quartz.JobListener;
 import org.quartz.Scheduler;
 import org.quartz.SchedulerException;
 import org.quartz.Trigger;
+import org.quartz.TriggerListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -50,8 +51,6 @@ public class JobRegisterServiceImpl implements JobRegisterService {
 
     private final TenantDetailsService tenantDetailsService;
 
-    private final HashMap<Long, Boolean> schedulerStatus = new HashMap<Long, Boolean>(2);
-
     private final HashMap<String, Scheduler> schedulers = new HashMap<String, Scheduler>(4);
 
     @Autowired
@@ -68,24 +67,32 @@ public class JobRegisterServiceImpl implements JobRegisterService {
         for (MifosPlatformTenant tenant : allTenants) {
             ThreadLocalContextUtil.setTenant(tenant);
             List<ScheduledJobDetail> scheduledJobDetails = this.schedularWritePlatformService.retrieveAllJobs();
-            schedulerStatus.put(tenant.getId(), true);
             for (ScheduledJobDetail jobDetails : scheduledJobDetails) {
                 scheduleJob(jobDetails);
+                jobDetails.updateTriggerMisfired(false);
                 this.schedularWritePlatformService.saveOrUpdate(jobDetails);
             }
+            SchedulerDetail schedulerDetail = this.schedularWritePlatformService.retriveSchedulerDetail();
+            if (schedulerDetail.isResetSchedulerOnBootup()) {
+                schedulerDetail.updateSuspendedState(false);
+                this.schedularWritePlatformService.updateSchedulerDetail(schedulerDetail);
+            }
         }
     }
 
-    public void executeJob(final ScheduledJobDetail scheduledJobDetail) {
+    public void executeJob(final ScheduledJobDetail scheduledJobDetail, String triggerType) {
         try {
             final JobDataMap jobDataMap = new JobDataMap();
-            jobDataMap.put(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE, SchedulerServiceConstants.TRIGGER_TYPE_APPLICATION);
+            if (triggerType == null) {
+                triggerType = SchedulerServiceConstants.TRIGGER_TYPE_APPLICATION;
+            }
+            jobDataMap.put(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE, triggerType);
             jobDataMap.put(SchedulerServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier());
             String key = scheduledJobDetail.getJobKey();
             JobKey jobKey = constructJobKey(key);
             String schedulerName = getSchedulerName(scheduledJobDetail);
             Scheduler scheduler = schedulers.get(schedulerName);
-            if (scheduler == null || scheduler.isInStandbyMode() || !scheduler.checkExists(jobKey)) {
+            if (scheduler == null || !scheduler.checkExists(jobKey)) {
                 JobDetail jobDetail = createJobDetail(scheduledJobDetail);
                 String tempSchedulerName = "temp" + scheduledJobDetail.getId();
                 List<Class<? extends JobListener>> listenerClasses = new ArrayList<Class<? extends JobListener>>(2);
@@ -101,7 +108,6 @@ public class JobRegisterServiceImpl implements JobRegisterService {
             }
 
         } catch (Exception e) {
-            e.printStackTrace();
             throw new PlatformInternalServerException("error.msg.sheduler.job.execution.failed", "Job execution failed for job with id:"
                     + scheduledJobDetail.getId(), scheduledJobDetail.getId());
         }
@@ -128,32 +134,48 @@ public class JobRegisterServiceImpl implements JobRegisterService {
 
     @Override
     public void pauseScheduler() {
-        schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false);
-        for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
-            try {
-                String schedulerName = schedulerEntry.getValue().getSchedulerName();
-                if (isCurrentTenantsScheduler(schedulerName)) {
-                    schedulerEntry.getValue().standby();
-                }
-            } catch (SchedulerException e) {
-                logger.error(e.getMessage(), e);
-            }
+        SchedulerDetail schedulerDetail = this.schedularWritePlatformService.retriveSchedulerDetail();
+        if (!schedulerDetail.isSuspended()) {
+            schedulerDetail.updateSuspendedState(true);
+            this.schedularWritePlatformService.updateSchedulerDetail(schedulerDetail);
         }
     }
 
     @Override
     public void startScheduler() {
-        for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
-            try {
-                String schedulerName = schedulerEntry.getValue().getSchedulerName();
-                if (isCurrentTenantsScheduler(schedulerName)) {
-                    schedulerEntry.getValue().start();
+        SchedulerDetail schedulerDetail = this.schedularWritePlatformService.retriveSchedulerDetail();
+        if (schedulerDetail.isSuspended()) {
+            schedulerDetail.updateSuspendedState(false);
+            this.schedularWritePlatformService.updateSchedulerDetail(schedulerDetail);
+            if (schedulerDetail.isExecuteInstructionForMisfiredJobs()) {
+                List<ScheduledJobDetail> scheduledJobDetails = this.schedularWritePlatformService.retrieveAllJobs();
+                for (ScheduledJobDetail jobDetail : scheduledJobDetails) {
+                    if (jobDetail.isTriggerMisfired()) {
+                        if (jobDetail.isActiveSchedular()) {
+                            executeJob(jobDetail, SchedulerServiceConstants.TRIGGER_TYPE_CRON);
+                        }
+                        String schedulerName = getSchedulerName(jobDetail);
+                        Scheduler scheduler = schedulers.get(schedulerName);
+                        if (scheduler != null) {
+                            String key = jobDetail.getJobKey();
+                            JobKey jobKey = constructJobKey(key);
+                            try {
+                                List<? extends Trigger> triggers = scheduler.getTriggersOfJob(jobKey);
+                                for (Trigger trigger : triggers) {
+                                    if (trigger.getNextFireTime() != null && trigger.getNextFireTime().after(jobDetail.getNextRunTime())) {
+                                        jobDetail.updateNextRunTime(trigger.getNextFireTime());
+                                    }
+                                }
+                            } catch (SchedulerException e) {
+                                logger.error(e.getMessage(), e);
+                            }
+                        }
+                        jobDetail.updateTriggerMisfired(false);
+                        this.schedularWritePlatformService.saveOrUpdate(jobDetail);
+                    }
                 }
-            } catch (SchedulerException e) {
-                logger.error(e.getMessage(), e);
             }
         }
-        schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true);
     }
 
     @Override
@@ -166,23 +188,12 @@ public class JobRegisterServiceImpl implements JobRegisterService {
     public void executeJob(final Long jobId) {
         ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId);
         if (scheduledJobDetail == null) { throw new JobNotFoundException(String.valueOf(jobId)); }
-        executeJob(scheduledJobDetail);
+        executeJob(scheduledJobDetail, null);
     }
 
     @Override
     public boolean isSchedulerRunning() {
-        return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId());
-    }
-
-    private boolean isCurrentTenantsScheduler(final String schedulerName) {
-        int beginIndex = SchedulerServiceConstants.SCHEDULER.length();
-        int endIndex = schedulerName.length();
-        if (schedulerName.contains(SchedulerServiceConstants.SCHEDULER_GROUP)) {
-            endIndex = schedulerName.indexOf(SchedulerServiceConstants.SCHEDULER_GROUP);
-        }
-        String tenantId = schedulerName.substring(beginIndex, endIndex);
-        MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
-        return tenant.getId() == Long.valueOf(tenantId);
+        return !this.schedularWritePlatformService.retriveSchedulerDetail().isSuspended();
     }
 
     private void scheduleJob(final ScheduledJobDetail scheduledJobDetails) {
@@ -195,9 +206,6 @@ public class JobRegisterServiceImpl implements JobRegisterService {
             JobDetail jobDetail = createJobDetail(scheduledJobDetails);
             Trigger trigger = createTrigger(scheduledJobDetails, jobDetail);
             Scheduler scheduler = getScheduler(scheduledJobDetails);
-            if (!this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId()) && !scheduler.isInStandbyMode()) {
-                scheduler.standby();
-            }
             scheduler.scheduleJob(jobDetail, trigger);
             scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey()));
             scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
@@ -253,6 +261,9 @@ public class JobRegisterServiceImpl implements JobRegisterService {
             listenerClasses.add(SchedulerJobListener.class);
         }
         schedulerFactoryBean.setGlobalJobListeners(getGlobalListener(listenerClasses));
+        TriggerListener globalTriggerListener = applicationContext.getBean("schedulerTriggerListener", TriggerListener.class);
+        TriggerListener[] globalTriggerListeners = { globalTriggerListener };
+        schedulerFactoryBean.setGlobalTriggerListeners(globalTriggerListeners);
         Properties quartzProperties = new Properties();
         quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads));
         schedulerFactoryBean.setQuartzProperties(quartzProperties);

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformService.java

@@ -6,6 +6,7 @@ import org.mifosplatform.infrastructure.core.api.JsonCommand;
 import org.mifosplatform.infrastructure.core.data.CommandProcessingResult;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory;
+import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetail;
 
 public interface SchedularWritePlatformService {
 
@@ -21,6 +22,12 @@ public interface SchedularWritePlatformService {
 
     public ScheduledJobDetail findByJobId(Long jobId);
 
-    CommandProcessingResult updateJobDetail(Long jobId, JsonCommand command);
+    public CommandProcessingResult updateJobDetail(Long jobId, JsonCommand command);
+
+    public SchedulerDetail retriveSchedulerDetail();
+
+    public void updateSchedulerDetail(final SchedulerDetail schedulerDetail);
+
+    public boolean processJobDetailForExecution(String jobKey, String triggerType);
 
 }

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformServiceJpaRepositoryImpl.java

@@ -1,5 +1,6 @@
 package org.mifosplatform.infrastructure.jobs.service;
 
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
@@ -11,6 +12,8 @@ import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetailRepository;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistoryRepository;
+import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetail;
+import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetailRepository;
 import org.mifosplatform.infrastructure.jobs.exception.JobNotFoundException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -23,13 +26,17 @@ public class SchedularWritePlatformServiceJpaRepositoryImpl implements Schedular
 
     private final ScheduledJobRunHistoryRepository scheduledJobRunHistoryRepository;
 
+    private final SchedulerDetailRepository schedulerDetailRepository;
+
     private final JobDetailDataValidator dataValidator;
 
     @Autowired
     public SchedularWritePlatformServiceJpaRepositoryImpl(final ScheduledJobDetailRepository scheduledJobDetailsRepository,
-            final ScheduledJobRunHistoryRepository scheduledJobRunHistoryRepository, final JobDetailDataValidator dataValidator) {
+            final ScheduledJobRunHistoryRepository scheduledJobRunHistoryRepository, final JobDetailDataValidator dataValidator,
+            final SchedulerDetailRepository schedulerDetailRepository) {
         this.scheduledJobDetailsRepository = scheduledJobDetailsRepository;
         this.scheduledJobRunHistoryRepository = scheduledJobRunHistoryRepository;
+        this.schedulerDetailRepository = schedulerDetailRepository;
         this.dataValidator = dataValidator;
     }
 
@@ -71,6 +78,22 @@ public class SchedularWritePlatformServiceJpaRepositoryImpl implements Schedular
         return this.scheduledJobDetailsRepository.findByJobId(jobId);
     }
 
+    @Override
+    @Transactional
+    public void updateSchedulerDetail(final SchedulerDetail schedulerDetail) {
+        this.schedulerDetailRepository.save(schedulerDetail);
+    }
+
+    @Override
+    public SchedulerDetail retriveSchedulerDetail() {
+        SchedulerDetail schedulerDetail = null;
+        List<SchedulerDetail> schedulerDetailList = schedulerDetailRepository.findAll();
+        if (schedulerDetailList != null) {
+            schedulerDetail = schedulerDetailList.get(0);
+        }
+        return schedulerDetail;
+    }
+
     @Transactional
     @Override
     public CommandProcessingResult updateJobDetail(final Long jobId, final JsonCommand command) {
@@ -89,4 +112,24 @@ public class SchedularWritePlatformServiceJpaRepositoryImpl implements Schedular
 
     }
 
+    @Transactional
+    @Override
+    public boolean processJobDetailForExecution(String jobKey, String triggerType) {
+        boolean isStopExecution = false;
+        final ScheduledJobDetail scheduledJobDetail = scheduledJobDetailsRepository.findByJobKeyWithLock(jobKey);
+        if (scheduledJobDetail.isCurrentlyRunning()
+                || (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON && (scheduledJobDetail.getNextRunTime().after(new Date())))) {
+            isStopExecution = true;
+        }
+        final SchedulerDetail schedulerDetail = retriveSchedulerDetail();
+        if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON && schedulerDetail.isSuspended()) {
+            scheduledJobDetail.updateTriggerMisfired(true);
+            isStopExecution = true;
+        } else if (!isStopExecution) {
+            scheduledJobDetail.updateCurrentlyRunningStatus(true);
+        }
+        scheduledJobDetailsRepository.save(scheduledJobDetail);
+        return isStopExecution;
+    }
+
 }

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java

@@ -2,12 +2,9 @@ package org.mifosplatform.infrastructure.jobs.service;
 
 import java.util.Date;
 
-import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant;
 import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail;
 import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory;
-import org.mifosplatform.infrastructure.jobs.exception.JobInProcessExecution;
-import org.mifosplatform.infrastructure.security.service.TenantDetailsService;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
 import org.quartz.JobKey;
@@ -31,12 +28,9 @@ public class SchedulerJobListener implements JobListener {
 
     private final SchedularWritePlatformService schedularService;
 
-    private final TenantDetailsService tenantDetailsService;
-
     @Autowired
-    public SchedulerJobListener(final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) {
+    public SchedulerJobListener(final SchedularWritePlatformService schedularService) {
         this.schedularService = schedularService;
-        this.tenantDetailsService = tenantDetailsService;
     }
 
     @Override
@@ -45,16 +39,7 @@ public class SchedulerJobListener implements JobListener {
     }
 
     @Override
-    public void jobToBeExecuted(final JobExecutionContext context) {
-        String tenantIdentifier = context.getTrigger().getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER);
-        MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier);
-        ThreadLocalContextUtil.setTenant(tenant);
-        JobKey key = context.getTrigger().getJobKey();
-        String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
-        final ScheduledJobDetail scheduledJobDetail = schedularService.findByJobKey(jobKey);
-        if (scheduledJobDetail.isCurrentlyRunning()) { throw new JobInProcessExecution(scheduledJobDetail.getJobName()); }
-        scheduledJobDetail.updateCurrentlyRunningStatus(true);
-        this.schedularService.saveOrUpdate(scheduledJobDetail);
+    public void jobToBeExecuted(@SuppressWarnings("unused") final JobExecutionContext context) {
 
     }
 
@@ -93,7 +78,8 @@ public class SchedulerJobListener implements JobListener {
         if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) {
             triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE);
         }
-        if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON) {
+        if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON && trigger.getNextFireTime() != null
+                && trigger.getNextFireTime().after(scheduledJobDetails.getNextRunTime())) {
             scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
         }
 

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerTriggerListener.java

@@ -0,0 +1,66 @@
+package org.mifosplatform.infrastructure.jobs.service;
+
+import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant;
+import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil;
+import org.mifosplatform.infrastructure.security.service.TenantDetailsService;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobKey;
+import org.quartz.Trigger;
+import org.quartz.Trigger.CompletedExecutionInstruction;
+import org.quartz.TriggerListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class SchedulerTriggerListener implements TriggerListener {
+
+    private final String name = "Global trigger Listner";
+
+    private final SchedularWritePlatformService schedularService;
+
+    private final TenantDetailsService tenantDetailsService;
+
+    @Autowired
+    public SchedulerTriggerListener(final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) {
+        this.schedularService = schedularService;
+        this.tenantDetailsService = tenantDetailsService;
+
+    }
+
+    @Override
+    public String getName() {
+        return this.name;
+    }
+
+    @Override
+    public void triggerFired(@SuppressWarnings("unused") Trigger trigger, @SuppressWarnings("unused") JobExecutionContext context) {
+
+    }
+
+    @Override
+    public boolean vetoJobExecution(Trigger trigger, JobExecutionContext context) {
+
+        String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER);
+        MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier);
+        ThreadLocalContextUtil.setTenant(tenant);
+        JobKey key = trigger.getJobKey();
+        String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
+        String triggerType = SchedulerServiceConstants.TRIGGER_TYPE_CRON;
+        if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) {
+            triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE);
+        }
+        return schedularService.processJobDetailForExecution(jobKey, triggerType);
+    }
+
+    @Override
+    public void triggerMisfired(@SuppressWarnings("unused") Trigger trigger) {
+
+    }
+
+    @Override
+    public void triggerComplete(@SuppressWarnings("unused") Trigger trigger, @SuppressWarnings("unused") JobExecutionContext context,
+            @SuppressWarnings("unused") CompletedExecutionInstruction triggerInstructionCode) {
+
+    }
+
+}