job-scheduler/src/main/java/org/hawkular/metrics/scheduler/impl/SchedulerImpl.java
private Single<JobExecutionState> reschedule(JobExecutionState executionState) {
        Trigger nextTrigger = executionState.currentDetails.getTrigger().nextTrigger();
        if (nextTrigger == null) {
            logger.debugf("No more scheduled executions for %s", executionState.currentDetails);
            return Single.just(executionState);
        }

        JobDetailsImpl details = executionState.currentDetails;
        JobDetailsImpl newDetails = new JobDetailsImpl(details, nextTrigger);

        if (nextTrigger.getTriggerTime() <= now.get().getMillis()) {
            logger.infof("%s missed its next execution at %d. It will be rescheduled for immediate execution.",
                    details, nextTrigger.getTriggerTime());
            AtomicLong nextTimeSlice = new AtomicLong(currentMinute().getMillis());
            Observable<Lock> scheduled = Observable.defer(() ->
                    lockManager.acquireLock(QUEUE_LOCK_PREFIX + nextTimeSlice.addAndGet(60000L), SCHEDULING_LOCK,
                            SCHEDULING_LOCK_TIMEOUT_IN_SEC, false))
                    .map(lock -> {
                        if (!lock.isLocked()) {
                            throw new RuntimeException();
                        }
                        return lock;
                    })
                    .retry();
            return scheduled
                    .map(lock -> new JobExecutionState(executionState.currentDetails, executionState.timeSlice,
                            newDetails, new Date(nextTimeSlice.get()), executionState.activeJobs))
                    .flatMap(state -> jobsService.insert(state.nextTimeSlice, state.nextDetails)
                            .map(updated -> state))
                    .doOnNext(state -> logger.debugf("Rescheduled %s to execute in time slice %s with trigger time " +
                            "of %s", state.nextDetails.getJobName(), state.nextTimeSlice, new Date(state.nextDetails
                            .getTrigger().getTriggerTime())))
                    .toSingle();
        }

        logger.debugf("Scheduling %s for next execution at %s", newDetails, new Date(nextTrigger.getTriggerTime()));

        JobExecutionState newState = new JobExecutionState(details, executionState.timeSlice, newDetails,
                new Date(nextTrigger.getTriggerTime()), executionState.activeJobs);
        return jobsService.insert(newState.nextTimeSlice, newState.nextDetails)
                .map(updated -> newState)
                .toSingle();
    }
private Single<JobExecutionState> reschedule(JobExecutionState executionState) {
        Trigger nextTrigger = executionState.currentDetails.getTrigger().nextTrigger();
        if (nextTrigger == null) {
            logger.debugf("No more scheduled executions for %s", executionState.currentDetails);
            return Single.just(executionState);
        }

        JobDetailsImpl details = executionState.currentDetails;
        JobDetailsImpl newDetails = new JobDetailsImpl(details, nextTrigger);

        if (nextTrigger.getTriggerTime() <= now.get().getMillis()) {
            logger.infof("%s missed its next execution at %d. It will be rescheduled for immediate execution.",
                    details, nextTrigger.getTriggerTime());
            AtomicLong nextTimeSlice = new AtomicLong(currentMinute().getMillis());
            Observable<Lock> scheduled = Observable.defer(() ->
                    lockManager.acquireLock(QUEUE_LOCK_PREFIX + nextTimeSlice.addAndGet(60000L), SCHEDULING_LOCK,
                            SCHEDULING_LOCK_TIMEOUT_IN_SEC, false))
                    .map(lock -> {
                        if (!lock.isLocked()) {
                            throw new RuntimeException();
                        }
                        return lock;
                    })
                    .retry();
            return scheduled
                    .map(lock -> new JobExecutionState(executionState.currentDetails, executionState.timeSlice,
                            newDetails, new Date(nextTimeSlice.get()), executionState.activeJobs))
                    .flatMap(state -> jobsService.insert(state.nextTimeSlice, state.nextDetails)
                            .map(updated -> state))
                    .doOnNext(state -> logger.debugf("Rescheduled %s to execute in time slice %s with trigger time " +
                            "of %s", state.nextDetails.getJobName(), state.nextTimeSlice, new Date(state.nextDetails
                            .getTrigger().getTriggerTime())))
                    .toSingle();
        }

        logger.debugf("Scheduling %s for next execution at %s", newDetails, new Date(nextTrigger.getTriggerTime()));

        JobExecutionState newState = new JobExecutionState(details, executionState.timeSlice, newDetails,
                new Date(nextTrigger.getTriggerTime()), executionState.activeJobs);
        return jobsService.insert(newState.nextTimeSlice, newState.nextDetails)
                .map(updated -> newState)
                .toSingle();
    }
private Single < JobExecutionState > reschedule ( JobExecutionState executionState ) { Trigger nextTrigger = executionState . currentDetails . getTrigger ( ) . nextTrigger ( ) ; if ( nextTrigger == null ) { logger . debugf ( "No more scheduled executions for %s" , executionState . currentDetails ) ; return Single . just ( executionState ) ; } JobDetailsImpl details = executionState . currentDetails ; JobDetailsImpl newDetails = new JobDetailsImpl ( details , nextTrigger ) ; if ( nextTrigger . getTriggerTime ( ) <= now . get ( ) . getMillis ( ) ) { logger . infof ( "%s missed its next execution at %d. It will be rescheduled for immediate execution." , details , nextTrigger . getTriggerTime ( ) ) ; AtomicLong nextTimeSlice = new AtomicLong ( currentMinute ( ) . getMillis ( ) ) ; Observable < Lock > scheduled = Observable . defer ( ( ) -> lockManager . acquireLock ( QUEUE_LOCK_PREFIX + nextTimeSlice . addAndGet ( 60000L ) , SCHEDULING_LOCK , SCHEDULING_LOCK_TIMEOUT_IN_SEC , false ) ) . map ( lock -> { if ( ! lock . isLocked ( ) ) { throw new RuntimeException ( ) ; } return lock ; } ) . retry ( ) ; return scheduled . map ( lock -> new JobExecutionState ( executionState . currentDetails , executionState . timeSlice , newDetails , new Date ( nextTimeSlice . get ( ) ) , executionState . activeJobs ) ) . flatMap ( state -> jobsService . insert ( state . nextTimeSlice , state . nextDetails ) . map ( updated -> state ) ) . doOnNext ( state -> logger . debugf ( "Rescheduled %s to execute in time slice %s with trigger time " + "of %s" , state . nextDetails . getJobName ( ) , state . nextTimeSlice , new Date ( state . nextDetails . getTrigger ( ) . getTriggerTime ( ) ) ) ) . toSingle ( ) ; } logger . debugf ( "Scheduling %s for next execution at %s" , newDetails , new Date ( nextTrigger . getTriggerTime ( ) ) ) ; JobExecutionState newState = new JobExecutionState ( details , executionState . timeSlice , newDetails , new Date ( nextTrigger . getTriggerTime ( ) ) , executionState . activeJobs ) ; return jobsService . insert ( newState . nextTimeSlice , newState . nextDetails ) . map ( updated -> newState ) . toSingle ( ) ; }
If a job is single execution then this is a no-op; otherwise, the scheduled jobs index and jobs tables are
updated with the next execution time/details. If the job has missed its next execution, it will get scheduled
in the next available time slice. The job scheduler will actually execute the job immediately when the job falls
behind, but we still want to persist the scheduling update for durability.
If a job is single execution then this is a no - op ; otherwise the scheduled jobs index and jobs tables are updated with the next execution time / details . If the job has missed its next execution it will get scheduled in the next available time slice . The job scheduler will actually execute the job immediately when the job falls behind but we still want to persist the scheduling update for durability .