text: /**
   * Copyright (c) 2010 Yahoo! Inc. All rights reserved.
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   *  Unless required by applicable law or agreed to in writing, software
   *  distributed under the License is distributed on an "AS IS" BASIS,
   *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   *  See the License for the specific language governing permissions and
   *  limitations under the License. See accompanying LICENSE file.
   */
  package org.apache.oozie.service;
  
  import java.util.Date;
  import java.util.List;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.oozie.BundleJobBean;
  import org.apache.oozie.CoordinatorJobBean;
  import org.apache.oozie.command.bundle.BundlePauseXCommand;
  import org.apache.oozie.command.bundle.BundleStartXCommand;
  import org.apache.oozie.command.bundle.BundleUnpauseXCommand;
  import org.apache.oozie.command.coord.CoordPauseXCommand;
  import org.apache.oozie.command.coord.CoordUnpauseXCommand;
  import org.apache.oozie.executor.jpa.BundleJobsGetNeedStartJPAExecutor;
  import org.apache.oozie.executor.jpa.BundleJobsGetPausedJPAExecutor;
  import org.apache.oozie.executor.jpa.BundleJobsGetUnpausedJPAExecutor;
  import org.apache.oozie.executor.jpa.CoordJobsGetPausedJPAExecutor;
  import org.apache.oozie.executor.jpa.CoordJobsGetUnpausedJPAExecutor;
  import org.apache.oozie.service.SchedulerService;
  import org.apache.oozie.service.Service;
  import org.apache.oozie.service.Services;
  import org.apache.oozie.util.MemoryLocks;
  import org.apache.oozie.util.XLog;
  
  /**
   * PauseTransitService is the runnable which is scheduled to run at the configured interval, it checks all bundles
   * to see if they should be paused, un-paused or started.
   */
  public class PauseTransitService implements Service {
      public static final String CONF_PREFIX = Service.CONF_PREFIX + "PauseTransitService.";
      public static final String CONF_BUNDLE_PAUSE_START_INTERVAL = CONF_PREFIX + "PauseTransit.interval";
      private final static XLog LOG = XLog.getLog(PauseTransitService.class);
  
      /**
       * PauseTransitRunnable is the runnable which is scheduled to run at the configured interval, it checks all
       * bundles to see if they should be paused, un-paused or started.
       */
      static class PauseTransitRunnable implements Runnable {
          private JPAService jpaService = null;
          private MemoryLocks.LockToken lock;
  
          public PauseTransitRunnable() {
              jpaService = Services.get().get(JPAService.class);
              if (jpaService == null) {
                  LOG.error("Missing JPAService");
              }
          }
  
          public void run() {
              try {
                  // first check if there is some other running instance from the same service;
                  lock = Services.get().get(MemoryLocksService.class).getWriteLock(
                          PauseTransitService.class.getName(), lockTimeout);
                  if (lock == null) {
                      LOG.info("This PauseTransitService instance will"
                              + "not run since there is already an instance running");
                  }
                  else {
                      LOG.info("Acquired lock for [{0}]", PauseTransitService.class.getName());
  
                      updateBundle();
                      updateCoord();
                  }
              }
              catch (Exception ex) {
                  LOG.warn("Exception happened when pausing/unpausing/starting bundle/coord jobs", ex);
              }
              finally {
                  // release lock;
                  if (lock != null) {
                      lock.release();
                      LOG.info("Released lock for [{0}]", PauseTransitService.class.getName());
                  }
              }
          }
  
          private void updateBundle() {
              Date d = new Date(); // records the start time of this service run;
              List<BundleJobBean> jobList = null;
              // pause bundles as needed;
              try {
                  jobList = jpaService.execute(new BundleJobsGetUnpausedJPAExecutor(-1));
                  if (jobList != null) {
                      for (BundleJobBean bundleJob : jobList) {
                          if ((bundleJob.getPauseTime() != null) && !bundleJob.getPauseTime().after(d)) {
                              new BundlePauseXCommand(bundleJob).call();
                              LOG.debug("Calling BundlePauseXCommand for bundle job = " + bundleJob.getId());
                          }
                      }
                  }
              }
              catch (Exception ex) {
                  LOG.warn("Exception happened when pausing/unpausing/starting Bundle jobs", ex);
              }
              // unpause bundles as needed;
              try {
                  jobList = jpaService.execute(new BundleJobsGetPausedJPAExecutor(-1));
                  if (jobList != null) {
                      for (BundleJobBean bundleJob : jobList) {
                          if ((bundleJob.getPauseTime() == null || bundleJob.getPauseTime().after(d))) {
                              new BundleUnpauseXCommand(bundleJob).call();
                              LOG.debug("Calling BundleUnpauseXCommand for bundle job = " + bundleJob.getId());
                          }
                      }
                  }
              }
              catch (Exception ex) {
                  LOG.warn("Exception happened when pausing/unpausing/starting Bundle jobs", ex);
              }
              // start bundles as needed;
              try {
                  jobList = jpaService.execute(new BundleJobsGetNeedStartJPAExecutor(d));
                  if (jobList != null) {
                      for (BundleJobBean bundleJob : jobList) {
                          bundleJob.setKickoffTime(d);
                          new BundleStartXCommand(bundleJob.getId()).call();
                          LOG.debug("Calling BundleStartXCommand for bundle job = " + bundleJob.getId());
                      }
                  }
              }
              catch (Exception ex) {
                  LOG.warn("Exception happened when pausing/unpausing/starting Bundle jobs", ex);
              }
          }
  
          private void updateCoord() {
              Date d = new Date(); // records the start time of this service run;
              List<CoordinatorJobBean> jobList = null;
              Configuration conf = Services.get().getConf();
              boolean backwardSupportForCoordStatus = conf.getBoolean(StatusTransitService.CONF_BACKWARD_SUPPORT_FOR_COORD_STATUS, false);
  
              // pause coordinators as needed;
              try {
                  jobList = jpaService.execute(new CoordJobsGetUnpausedJPAExecutor(-1));
                  if (jobList != null) {
                      for (CoordinatorJobBean coordJob : jobList) {
                          // if namespace 0.1 is used and backward support is true, then ignore this coord job
                          if (backwardSupportForCoordStatus == true && coordJob.getAppNamespace() != null
                                  && coordJob.getAppNamespace().equals(SchemaService.COORDINATOR_NAMESPACE_URI_1)) {
                              continue;
                          }
                          if ((coordJob.getPauseTime() != null) && !coordJob.getPauseTime().after(d)) {
                              new CoordPauseXCommand(coordJob).call();
                              LOG.debug("Calling CoordPauseXCommand for coordinator job = " + coordJob.getId());
                          }
                      }
                  }
              }
              catch (Exception ex) {
                  LOG.warn("Exception happened when pausing/unpausing Coordinator jobs", ex);
              }
              // unpause coordinators as needed;
              try {
                  jobList = jpaService.execute(new CoordJobsGetPausedJPAExecutor(-1));
                  if (jobList != null) {
                      for (CoordinatorJobBean coordJob : jobList) {
                          // if namespace 0.1 is used and backward support is true, then ignore this coord job
                          if (backwardSupportForCoordStatus == true && coordJob.getAppNamespace() != null
                                  && coordJob.getAppNamespace().equals(SchemaService.COORDINATOR_NAMESPACE_URI_1)) {
                              continue;
                          }
                          if ((coordJob.getPauseTime() == null || coordJob.getPauseTime().after(d))) {
                              new CoordUnpauseXCommand(coordJob).call();
                              LOG.debug("Calling CoordUnpauseXCommand for coordinator job = " + coordJob.getId());
                          }
                      }
                  }
              }
              catch (Exception ex) {
                  LOG.warn("Exception happened when pausing/unpausing Coordinator jobs", ex);
              }
          }
      }
  
      /**
       * Initializes the {@link PauseTransitService}.
       *
       * @param services services instance.
       */
      @Override
      public void init(Services services) {
          Configuration conf = services.getConf();
          Runnable bundlePauseStartRunnable = new PauseTransitRunnable();
          services.get(SchedulerService.class).schedule(bundlePauseStartRunnable, 10,
                  conf.getInt(CONF_BUNDLE_PAUSE_START_INTERVAL, 60), SchedulerService.Unit.SEC);
      }
  
      /**
       * Destroy the StateTransit Jobs Service.
       */
      @Override
      public void destroy() {
      }
  
      /**
       * Return the public interface for the purge jobs service.
       *
       * @return {@link PauseTransitService}.
       */
      @Override
      public Class<? extends Service> getInterface() {
          return PauseTransitService.class;
      }
  }
  