ambari-server/src/main/java/org/apache/ambari/server/scheduler/ExecutionSchedulerImpl.java

/*
 * 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.ambari.server.scheduler;

import java.util.List;
import java.util.Properties;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.configuration.Configuration.DatabaseType;
import org.apache.ambari.server.state.scheduler.GuiceJobFactory;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;
import com.google.inject.Injector;
import com.google.inject.Singleton;

@Singleton
public class ExecutionSchedulerImpl implements ExecutionScheduler {
  @Inject
  private Configuration configuration;
  @Inject
  GuiceJobFactory guiceJobFactory;

  private static final Logger LOG = LoggerFactory.getLogger(ExecutionSchedulerImpl.class);
  protected static final String DEFAULT_SCHEDULER_NAME = "ExecutionScheduler";
  protected Scheduler scheduler;
  protected static volatile boolean isInitialized = false;

  @Inject
  public ExecutionSchedulerImpl(Injector injector) {
    injector.injectMembers(this);
  }

  /**
   * Constructor used for unit testing
   * @param configuration
   */
  protected ExecutionSchedulerImpl(Configuration configuration) {
    this.configuration = configuration;
  }

  protected synchronized void initializeScheduler() {
    StdSchedulerFactory sf = new StdSchedulerFactory();
    Properties properties = getQuartzSchedulerProperties();
    try {
      sf.initialize(properties);
    } catch (SchedulerException e) {
      LOG.warn("Failed to initialize Request Execution Scheduler properties !");
      LOG.debug("Scheduler properties: \n{}", properties);
      e.printStackTrace();
      return;
    }
    try {
      scheduler = sf.getScheduler();
      scheduler.setJobFactory(guiceJobFactory);
      isInitialized = true;
    } catch (SchedulerException e) {
      LOG.warn("Failed to create Request Execution scheduler !");
      e.printStackTrace();
    }
  }

  protected Properties getQuartzSchedulerProperties() {
    Properties properties = new Properties();
    properties.setProperty("org.quartz.scheduler.instanceName", DEFAULT_SCHEDULER_NAME);
    properties.setProperty("org.quartz.scheduler.instanceId", "AUTO");
    properties.setProperty("org.quartz.threadPool.class",
      "org.quartz.simpl.SimpleThreadPool");
    properties.setProperty("org.quartz.threadPool.threadCount",
      configuration.getExecutionSchedulerThreads());

    // Job Store Configuration
    properties.setProperty("org.quartz.jobStore.class",
      "org.quartz.impl.jdbcjobstore.JobStoreTX");
    properties.setProperty("org.quartz.jobStore.isClustered",
      configuration.isExecutionSchedulerClusterd());

    String[] subProps = getQuartzDbDelegateClassAndValidationQuery();

    properties.setProperty("org.quartz.jobStore.driverDelegateClass",
      subProps[0]);
    // Allow only strings in the jobDataMap which is serialized
    properties.setProperty("org.quartz.jobStore.useProperties", "false");

    // Data store configuration
    properties.setProperty("org.quartz.jobStore.dataSource", "myDS");
    properties.setProperty("org.quartz.dataSource.myDS.driver",
      configuration.getDatabaseDriver());
    properties.setProperty("org.quartz.dataSource.myDS.URL",
      configuration.getDatabaseUrl());
    properties.setProperty("org.quartz.dataSource.myDS.user",
      configuration.getDatabaseUser());
    properties.setProperty("org.quartz.dataSource.myDS.password",
      configuration.getDatabasePassword());
    properties.setProperty("org.quartz.dataSource.myDS.maxConnections",
      configuration.getExecutionSchedulerConnections());
    properties.setProperty("org.quartz.dataSource.myDS.maxCachedStatementsPerConnection",
      configuration.getExecutionSchedulerMaxStatementsPerConnection());
    properties.setProperty("org.quartz.dataSource.myDS.validationQuery",
      subProps[1]);

    // Skip update check
    properties.setProperty("org.quartz.scheduler.skipUpdateCheck", "true");

    LOG.debug("Using quartz properties: {}", properties);
    return properties;
  }

  protected synchronized boolean isInitialized() {
    return isInitialized;
  }


  protected String[] getQuartzDbDelegateClassAndValidationQuery() {
    String dbDelegate = "org.quartz.impl.jdbcjobstore.StdJDBCDelegate";
    String dbValidate = "select 0";

    DatabaseType databaseType = configuration.getDatabaseType();
    if (databaseType == DatabaseType.POSTGRES) {
      dbDelegate = "org.quartz.impl.jdbcjobstore.PostgreSQLDelegate";
    } else if (databaseType == DatabaseType.ORACLE) {
      dbDelegate = "org.quartz.impl.jdbcjobstore.oracle.OracleDelegate";
      dbValidate = "select 0 from dual";
    }

    return new String[] { dbDelegate, dbValidate };
  }

  @Override
  public synchronized void startScheduler(Integer delay) throws AmbariException {
    try {
      if (!isInitialized) {
        initializeScheduler();
        isInitialized = true;
      }
    } catch (Exception e) {
      String msg = "Unable to initialize Request Execution scheduler !";
      LOG.warn(msg);
      e.printStackTrace();
      throw new AmbariException(msg);
    }
    try {
      if (!scheduler.isStarted()) {
        // To avoid issue created due to change in server clock in between the
        // scheduler initialization and scheduler start,
        // start immediately if no delay provided.
        if (delay != null) {
          scheduler.startDelayed(delay);
        } else {
          scheduler.start();
        }

      } else {
        LOG.info("Scheduler " + scheduler.getSchedulerInstanceId() +
          " already started. Skipping start.");
      }
    } catch (SchedulerException e) {
      LOG.error("Failed to start scheduler", e);
      throw new AmbariException(e.getMessage());
    }
  }

  @Override
  public synchronized void stopScheduler() throws AmbariException {
    if (scheduler == null) {
      throw new AmbariException("Scheduler not instantiated !");
    }
    try {
      scheduler.shutdown();
    } catch (SchedulerException e) {
      LOG.error("Failed to stop scheduler", e);
      throw new AmbariException(e.getMessage());
    }
  }

  @Override
  public void scheduleJob(Trigger trigger) throws SchedulerException {
    scheduler.scheduleJob(trigger);
  }

  @Override
  public void addJob(JobDetail jobDetail) throws SchedulerException {
    scheduler.addJob(jobDetail, true);
  }

  @Override
  public void deleteJob(JobKey jobKey) throws SchedulerException {
    scheduler.deleteJob(jobKey);
  }

  @Override
  public JobDetail getJobDetail(JobKey jobKey) throws SchedulerException {
    return scheduler.getJobDetail(jobKey);
  }

  @Override
  public List<? extends Trigger> getTriggersForJob(JobKey jobKey) throws SchedulerException {
    return scheduler.getTriggersOfJob(jobKey);
  }

  @Override
  public boolean isSchedulerStarted() throws SchedulerException {
    return scheduler.isStarted();
  }

}