core/src/main/java/org/apache/oozie/service/StatusTransitService.java - oozie - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.
  */

 package org.apache.oozie.service;

 import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.oozie.BundleJobBean;
 import org.apache.oozie.CoordinatorJobBean;
 import org.apache.oozie.ErrorCode;
 import org.apache.oozie.command.CommandException;
 import org.apache.oozie.command.bundle.BundleStatusTransitXCommand;
 import org.apache.oozie.command.coord.CoordStatusTransitXCommand;
 import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
 import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery;
 import org.apache.oozie.executor.jpa.BundleJobsGetRunningOrPendingJPAExecutor;
 import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
 import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
 import org.apache.oozie.executor.jpa.CoordJobsGetPendingJPAExecutor;
 import org.apache.oozie.executor.jpa.JPAExecutorException;
 import org.apache.oozie.util.DateUtils;
 import org.apache.oozie.lock.LockToken;
 import org.apache.oozie.util.XLog;

 /**
  * StateTransitService is scheduled to run at the configured interval.
  * <p>
  * It is to update job's status according to its child actions' status. If all child actions' pending flag equals 0 (job
  * done), we reset the job's pending flag to 0. If all child actions are succeeded, we set the job's status to
  * SUCCEEDED.
  */
 public class StatusTransitService implements Service {
     private static final String CONF_PREFIX = Service.CONF_PREFIX + "StatusTransitService.";
     private static final String CONF_STATUSTRANSIT_INTERVAL = CONF_PREFIX + "statusTransit.interval";
     public static final String CONF_BACKWARD_SUPPORT_FOR_COORD_STATUS = CONF_PREFIX
             + "backward.support.for.coord.status";
     public static final String CONF_BACKWARD_SUPPORT_FOR_STATES_WITHOUT_ERROR = CONF_PREFIX
             + "backward.support.for.states.without.error";
     public static int limit = -1;
     public static Date lastInstanceStartTime = null;
     public final static XLog LOG = XLog.getLog(StatusTransitRunnable.class);

     /**
      * StateTransitRunnable is the runnable which is scheduled to run at the configured interval.
      * <p>
      * It is to update job's status according to its child actions' status. If all child actions' pending flag equals 0
      * (job done), we reset the job's pending flag to 0. If all child actions are succeeded, we set the job's status to
      * SUCCEEDED.
      */
     public static class StatusTransitRunnable implements Runnable {
         private JPAService jpaService = null;
         private LockToken lock;

         private Set<String> coordFailedIds = new HashSet<String>();
         private Set<String> bundleFailedIds = new HashSet<String>();

         public StatusTransitRunnable() {
             jpaService = Services.get().get(JPAService.class);
             if (jpaService == null) {
                 LOG.error("Missing JPAService");
             }
         }

         @Override
         public void run() {
             try {
                 final Date curDate = new Date(); // records the start time of this service run;

                 // first check if there is some other instance running;
                 lock = Services.get().get(MemoryLocksService.class)
                         .getWriteLock(StatusTransitService.class.getName(), lockTimeout);
                 if (lock == null) {
                     LOG.info("This StatusTransitService instance"
                             + " will not run since there is already an instance running");
                 }
                 else {
                     LOG.info("Acquired lock for [{0}]", StatusTransitService.class.getName());
                     coordTransit();
                     bundleTransit();
                     lastInstanceStartTime = curDate;
                 }
             }
             catch (Exception ex) {
                 LOG.warn("Exception happened during StatusTransitRunnable ", ex);
             }
             finally {
                 if (lock != null) {
                     lock.release();
                     LOG.info("Released lock for [{0}]", StatusTransitService.class.getName());
                 }
             }
         }

         /**
          * Aggregate bundle actions' status to bundle jobs
          *
          * @throws JPAExecutorException thrown if failed in db updates or retrievals
          * @throws CommandException thrown if failed to run commands
          */
         private void bundleTransit() throws JPAExecutorException, CommandException {
             List<BundleJobBean> pendingJobCheckList;
             final Set<String> bundleIds = new HashSet<String>();

             if (lastInstanceStartTime == null) {
                 LOG.info("Running bundle status service first instance");
                 // This is the first instance, we need to check for all pending or running jobs;
                 // TODO currently limit is = -1. Need to do actual batching
                 pendingJobCheckList = jpaService.execute(new BundleJobsGetRunningOrPendingJPAExecutor(limit));
             }
             else {
                 LOG.info("Running bundle status service from last instance time =  "
                         + DateUtils.formatDateOozieTZ(lastInstanceStartTime));
                 // this is not the first instance, we should only check jobs that have actions been
                 // updated >= start time of last service run;
                 pendingJobCheckList = BundleJobQueryExecutor.getInstance().getList(
                         BundleJobQuery.GET_BUNDLE_IDS_FOR_STATUS_TRANSIT, lastInstanceStartTime);
             }
             for (BundleJobBean job : pendingJobCheckList) {
                 bundleIds.add(job.getId());
             }
             bundleIds.addAll(bundleFailedIds);
             bundleFailedIds.clear();
             for (final String jobId : bundleIds) {
                 try {
                     new BundleStatusTransitXCommand(jobId).call();
                 }
                 catch (CommandException e) {
                     // Unable to acquire lock. Will try next time
                     if (e.getErrorCode() == ErrorCode.E0606) {
                         bundleFailedIds.add(jobId);
                         LOG.info("Unable to acquire lock for " + jobId + ". Will try next time");
                     }
                     else {
                         LOG.error("Error running BundleStatusTransitXCommand for job " + jobId, e);
                     }

                 }
             }
         }

         /**
          * Aggregate coordinator actions' status to coordinator jobs
          *
          * @throws JPAExecutorException thrown if failed in db updates or retrievals
          * @throws CommandException thrown if failed to run commands
          */
         private void coordTransit() throws JPAExecutorException, CommandException {
             List<CoordinatorJobBean> pendingJobCheckList = null;
             final Set<String> coordIds = new HashSet<String>();
             if (lastInstanceStartTime == null) {
                 LOG.info("Running coordinator status service first instance");
                 // this is the first instance, we need to check for all pending jobs;
                 pendingJobCheckList = jpaService.execute(new CoordJobsGetPendingJPAExecutor(limit));
             }
             else {
                 LOG.info("Running coordinator status service from last instance time =  "
                         + DateUtils.formatDateOozieTZ(lastInstanceStartTime));
                 // this is not the first instance, we should only check jobs.
                 // that have actions or jobs been updated >= start time of last service run;
                 pendingJobCheckList = CoordJobQueryExecutor.getInstance().getList(
                         CoordJobQuery.GET_COORD_IDS_FOR_STATUS_TRANSIT, lastInstanceStartTime);

                 pendingJobCheckList.addAll(CoordJobQueryExecutor.getInstance().getList(
                         CoordJobQuery.GET_COORD_JOBS_CHANGED, lastInstanceStartTime));
             }
             for (final CoordinatorJobBean job : pendingJobCheckList) {
                 coordIds.add(job.getId());
             }
             coordIds.addAll(coordFailedIds);
             coordFailedIds.clear();
             for (final String coordId : coordIds) {
                 try {
                     new CoordStatusTransitXCommand(coordId).call();
                 }
                 catch (CommandException e) {
                     // Unable to acquire lock. Will try next time
                     if (e.getErrorCode() == ErrorCode.E0606) {
                         coordFailedIds.add(coordId);
                         LOG.info("Unable to acquire lock for " + coordId + ". Will try next time");

                     }
                     else {
                         LOG.error("Error running CoordStatusTransitXCommand for job " + coordId, e);
                     }

                 }
             }
         }
     }

     /**
      * Initializes the {@link StatusTransitService}.
      *
      * @param services services instance.
      */
     @Override
     public void init(Services services) {
         final Configuration conf = services.getConf();
         Runnable stateTransitRunnable = new StatusTransitRunnable();
         services.get(SchedulerService.class).schedule(stateTransitRunnable, 10,
                 ConfigurationService.getInt(conf, CONF_STATUSTRANSIT_INTERVAL), SchedulerService.Unit.SEC);
     }

     /**
      * Destroy the StateTransit Jobs Service.
      */
     @Override
     public void destroy() {
     }

     /**
      * Return the public interface for the purge jobs service.
      *
      * @return {@link StatusTransitService}.
      */
     @Override
     public Class<? extends Service> getInterface() {
         return StatusTransitService.class;
     }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.
	*/

	package org.apache.oozie.service;

	import java.util.Date;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Set;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.oozie.BundleJobBean;
	import org.apache.oozie.CoordinatorJobBean;
	import org.apache.oozie.ErrorCode;
	import org.apache.oozie.command.CommandException;
	import org.apache.oozie.command.bundle.BundleStatusTransitXCommand;
	import org.apache.oozie.command.coord.CoordStatusTransitXCommand;
	import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
	import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery;
	import org.apache.oozie.executor.jpa.BundleJobsGetRunningOrPendingJPAExecutor;
	import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
	import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
	import org.apache.oozie.executor.jpa.CoordJobsGetPendingJPAExecutor;
	import org.apache.oozie.executor.jpa.JPAExecutorException;
	import org.apache.oozie.util.DateUtils;
	import org.apache.oozie.lock.LockToken;
	import org.apache.oozie.util.XLog;

	/**
	* StateTransitService is scheduled to run at the configured interval.
	* <p>
	* It is to update job's status according to its child actions' status. If all child actions' pending flag equals 0 (job
	* done), we reset the job's pending flag to 0. If all child actions are succeeded, we set the job's status to
	* SUCCEEDED.
	*/
	public class StatusTransitService implements Service {
	private static final String CONF_PREFIX = Service.CONF_PREFIX + "StatusTransitService.";
	private static final String CONF_STATUSTRANSIT_INTERVAL = CONF_PREFIX + "statusTransit.interval";
	public static final String CONF_BACKWARD_SUPPORT_FOR_COORD_STATUS = CONF_PREFIX
	+ "backward.support.for.coord.status";
	public static final String CONF_BACKWARD_SUPPORT_FOR_STATES_WITHOUT_ERROR = CONF_PREFIX
	+ "backward.support.for.states.without.error";
	public static int limit = -1;
	public static Date lastInstanceStartTime = null;
	public final static XLog LOG = XLog.getLog(StatusTransitRunnable.class);

	/**
	* StateTransitRunnable is the runnable which is scheduled to run at the configured interval.
	* <p>
	* It is to update job's status according to its child actions' status. If all child actions' pending flag equals 0
	* (job done), we reset the job's pending flag to 0. If all child actions are succeeded, we set the job's status to
	* SUCCEEDED.
	*/
	public static class StatusTransitRunnable implements Runnable {
	private JPAService jpaService = null;
	private LockToken lock;

	private Set<String> coordFailedIds = new HashSet<String>();
	private Set<String> bundleFailedIds = new HashSet<String>();

	public StatusTransitRunnable() {
	jpaService = Services.get().get(JPAService.class);
	if (jpaService == null) {
	LOG.error("Missing JPAService");
	}
	}

	@Override
	public void run() {
	try {
	final Date curDate = new Date(); // records the start time of this service run;

	// first check if there is some other instance running;
	lock = Services.get().get(MemoryLocksService.class)
	.getWriteLock(StatusTransitService.class.getName(), lockTimeout);
	if (lock == null) {
	LOG.info("This StatusTransitService instance"
	+ " will not run since there is already an instance running");
	}
	else {
	LOG.info("Acquired lock for [{0}]", StatusTransitService.class.getName());
	coordTransit();
	bundleTransit();
	lastInstanceStartTime = curDate;
	}
	}
	catch (Exception ex) {
	LOG.warn("Exception happened during StatusTransitRunnable ", ex);
	}
	finally {
	if (lock != null) {
	lock.release();
	LOG.info("Released lock for [{0}]", StatusTransitService.class.getName());
	}
	}
	}

	/**
	* Aggregate bundle actions' status to bundle jobs
	*
	* @throws JPAExecutorException thrown if failed in db updates or retrievals
	* @throws CommandException thrown if failed to run commands
	*/
	private void bundleTransit() throws JPAExecutorException, CommandException {
	List<BundleJobBean> pendingJobCheckList;
	final Set<String> bundleIds = new HashSet<String>();

	if (lastInstanceStartTime == null) {
	LOG.info("Running bundle status service first instance");
	// This is the first instance, we need to check for all pending or running jobs;
	// TODO currently limit is = -1. Need to do actual batching
	pendingJobCheckList = jpaService.execute(new BundleJobsGetRunningOrPendingJPAExecutor(limit));
	}
	else {
	LOG.info("Running bundle status service from last instance time = "
	+ DateUtils.formatDateOozieTZ(lastInstanceStartTime));
	// this is not the first instance, we should only check jobs that have actions been
	// updated >= start time of last service run;
	pendingJobCheckList = BundleJobQueryExecutor.getInstance().getList(
	BundleJobQuery.GET_BUNDLE_IDS_FOR_STATUS_TRANSIT, lastInstanceStartTime);
	}
	for (BundleJobBean job : pendingJobCheckList) {
	bundleIds.add(job.getId());
	}
	bundleIds.addAll(bundleFailedIds);
	bundleFailedIds.clear();
	for (final String jobId : bundleIds) {
	try {
	new BundleStatusTransitXCommand(jobId).call();
	}
	catch (CommandException e) {
	// Unable to acquire lock. Will try next time
	if (e.getErrorCode() == ErrorCode.E0606) {
	bundleFailedIds.add(jobId);
	LOG.info("Unable to acquire lock for " + jobId + ". Will try next time");
	}
	else {
	LOG.error("Error running BundleStatusTransitXCommand for job " + jobId, e);
	}

	}
	}
	}

	/**
	* Aggregate coordinator actions' status to coordinator jobs
	*
	* @throws JPAExecutorException thrown if failed in db updates or retrievals
	* @throws CommandException thrown if failed to run commands
	*/
	private void coordTransit() throws JPAExecutorException, CommandException {
	List<CoordinatorJobBean> pendingJobCheckList = null;
	final Set<String> coordIds = new HashSet<String>();
	if (lastInstanceStartTime == null) {
	LOG.info("Running coordinator status service first instance");
	// this is the first instance, we need to check for all pending jobs;
	pendingJobCheckList = jpaService.execute(new CoordJobsGetPendingJPAExecutor(limit));
	}
	else {
	LOG.info("Running coordinator status service from last instance time = "
	+ DateUtils.formatDateOozieTZ(lastInstanceStartTime));
	// this is not the first instance, we should only check jobs.
	// that have actions or jobs been updated >= start time of last service run;
	pendingJobCheckList = CoordJobQueryExecutor.getInstance().getList(
	CoordJobQuery.GET_COORD_IDS_FOR_STATUS_TRANSIT, lastInstanceStartTime);

	pendingJobCheckList.addAll(CoordJobQueryExecutor.getInstance().getList(
	CoordJobQuery.GET_COORD_JOBS_CHANGED, lastInstanceStartTime));
	}
	for (final CoordinatorJobBean job : pendingJobCheckList) {
	coordIds.add(job.getId());
	}
	coordIds.addAll(coordFailedIds);
	coordFailedIds.clear();
	for (final String coordId : coordIds) {
	try {
	new CoordStatusTransitXCommand(coordId).call();
	}
	catch (CommandException e) {
	// Unable to acquire lock. Will try next time
	if (e.getErrorCode() == ErrorCode.E0606) {
	coordFailedIds.add(coordId);
	LOG.info("Unable to acquire lock for " + coordId + ". Will try next time");

	}
	else {
	LOG.error("Error running CoordStatusTransitXCommand for job " + coordId, e);
	}

	}
	}
	}
	}

	/**
	* Initializes the {@link StatusTransitService}.
	*
	* @param services services instance.
	*/
	@Override
	public void init(Services services) {
	final Configuration conf = services.getConf();
	Runnable stateTransitRunnable = new StatusTransitRunnable();
	services.get(SchedulerService.class).schedule(stateTransitRunnable, 10,
	ConfigurationService.getInt(conf, CONF_STATUSTRANSIT_INTERVAL), SchedulerService.Unit.SEC);
	}

	/**
	* Destroy the StateTransit Jobs Service.
	*/
	@Override
	public void destroy() {
	}

	/**
	* Return the public interface for the purge jobs service.
	*
	* @return {@link StatusTransitService}.
	*/
	@Override
	public Class<? extends Service> getInterface() {
	return StatusTransitService.class;
	}

	}