Part 2 - Implementing Multi Threaded Batching Framework In Java

Batch.java

public class Batch
{

  private static final String ENFORCE_SINGLE_INSTANCE_PROPERTY_KEY = "enforceSingleInstance";

  private static final String ASSUME_JOB_TIMEOUT_MINUTES_KEY = "assumeJobTimeoutMinutes";

  private static final String GET_CONTROL_ENTRY_SQL = "SELECT ID, PROCESS_NAME, START_TMS, END_TMS, SYSTIMESTAMP as CURRENT_TIME, TOTAL_ROWS, FAILURE_COUNT FROM BATCHING_FRAMEWORK_CONTROL WHERE PROCESS_NAME = ? FOR UPDATE";

  private static final String ADD_CONTROL_ENTRY_SQL = "INSERT INTO BATCHING_FRAMEWORK_CONTROL (ID, PROCESS_NAME, START_TMS, END_TMS, TOTAL_ROWS) VALUES (BFC_ID_SEQ.NEXTVAL, ?, systimestamp, null, 0)";

  private static final String END_RUN_CONTROL_ENTRY_SQL = "UPDATE BATCHING_FRAMEWORK_CONTROL SET END_TMS=systimestamp"
      + ", TOTAL_ROWS=?, FAILURE_COUNT=? WHERE PROCESS_NAME=?";

  private static final String START_RUN_CONTROL_ENTRY_SQL = "UPDATE BATCHING_FRAMEWORK_CONTROL SET START_TMS=systimestamp, END_TMS=null"
      + ", TOTAL_ROWS=0, FAILURE_COUNT=0 WHERE PROCESS_NAME=?";

  /**
   * 6 hours timeout assumption
   */
  private static final String ASSUME_JOB_TIMEOUT_MINUTES_DEFAULT_VALUE = "360";

  private Constructor setConstructor;

  private Constructor processorConstructor;

  private String processorClassName;

  private int threads;

  private int metricDelay;

  private List<BatchProcessor> processors;

  private long startTime;

  private long total;

  private long count;

  private long failureCount;

  private boolean BatchSuccessful = true;

  private int numProcessors = -1;

  private boolean enforceSingleInstance;

  private int assumeJobTimeoutMilliseconds;

  /**
   * The JDBC connection from the datasource
   */
  private Connection connection;

  public Batch(Class setClass, Class processorClass, int threads, int metricDelay)
  {
    initialiseConfig(processorClass);

    try
    {
      this.threads = threads;
      this.metricDelay = metricDelay;
      processors = Collections.synchronizedList(new ArrayList<BatchProcessor>(this.threads));
      setConstructor = setClass.getConstructor(new Class[] { Batch.class, });
      processorConstructor = processorClass.getConstructor(new Class[] { Batch.class, BatchSet.class, });
      processorClassName = processorClass.getName();
    }
    catch (Exception exception)
    {
      throw new RuntimeException(errorMsg, exception);
    }
  }

  /**
   * Perform property configuration
   */
  private void initialiseConfig(Class processorClass)
  {

    ResourceHandler resourceHandler = ResourceHandler.getInstance();

    /**
     * Get property to see if we should enforce a single instance of this Batch
     * to run at a time
     */
    enforceSingleInstance = Boolean.valueOf(resourceHandler.getValue(Batch.class, ENFORCE_SINGLE_INSTANCE_PROPERTY_KEY, "true"));

    String subclassEnforceSingleInstance = resourceHandler.getValue(processorClass, ENFORCE_SINGLE_INSTANCE_PROPERTY_KEY, null);
    if (subclassEnforceSingleInstance != null)
    {
      // we've a class specific version of this property, so use this instead
      enforceSingleInstance = Boolean.valueOf(subclassEnforceSingleInstance);
    }

    /**
     * Get property for maximum duration a job should run for before we treat it
     * as
     */
    assumeJobTimeoutMilliseconds = Integer.valueOf(resourceHandler.getValue(Batch.class, ASSUME_JOB_TIMEOUT_MINUTES_KEY,
        ASSUME_JOB_TIMEOUT_MINUTES_DEFAULT_VALUE)) * 60000;

    String subclassAssumeJobTimeoutMinutes = resourceHandler.getValue(processorClass, ASSUME_JOB_TIMEOUT_MINUTES_KEY, null);
    if (subclassAssumeJobTimeoutMinutes != null)
    {
      // we've a class specific version of this property, so use this instead
      assumeJobTimeoutMilliseconds = Integer.valueOf(subclassAssumeJobTimeoutMinutes) * 60000;
    }

  }

  public synchronized void finished(final BatchProcessor processor)
  {
    if (processors.contains(processor))
    {
      processors.remove(processor);
    }

    if (processors.isEmpty())
    {
      notify();
    }
  }

  public synchronized boolean done()
  {
   
    if (processors.isEmpty()) return true;
    try
    {
      wait(metricDelay);
    }
    catch (InterruptedException interruptedException)
    {
    }
    return false;
  }

  public void execute()
  {

    ControlTO controlTO = markJobAsRunning(processorClassName);
    if (controlTO.isJobIsRunning())
    {
      //  // Do something meaningful like record this into error table etc
      return;
    }

    RuntimeException firstException = null;

    try
    {
      runProcesses();
    }
    catch (RuntimeException re)
    {
      firstException = re;
    }
    finally
    {
      try
      {
        completeControlEntry(processorClassName);
      }
      catch (RuntimeException re)
      {
        if (firstException != null)
        {
          firstException = re;
        }
      }
    }

    if (firstException != null)
    {
      throw firstException;
    }

  }

  /**
   * This performs the main functionality of this class, that is, get the set of
   * data to work on an then dish out this work to each thread. This code will
   * only run if this is the only instance of this process running
   * 
   */
  private void runProcesses()
  {
    BatchSet set;
    set = null;
    try
    {
      try
      {
        set = (BatchSet) setConstructor.newInstance(new Object[] { this, });
        set.open();
        total = set.total();
        for (int i = 0; i < threads; i++)
        {
          final BatchProcessor processor = (BatchProcessor) processorConstructor.newInstance(new Object[] { this, set, });
          processors.add(processor);
        }
      }
      catch (Exception e)
      {
        throw new RuntimeException("Error constructing Batch", e);
      }

      startTime = System.currentTimeMillis();
      List<BatchProcessor> toStart = new ArrayList<BatchProcessor>(processors.size());
      toStart.addAll(processors);

      numProcessors = processors.size();

      for (BatchProcessor processor : toStart)
      {
        (new Thread(processor)).start();
      }

      do
      {
        if (done())
        {
          break;
        }

        count = set.count();
      }
      while (true);

      // output final stats
    }
    catch (RuntimeException re)
    {
      // ensure we catch this and log it as we DO NOT want any missing messages
      logger.logMessage(Logger.LEVEL_ERROR, "An exception occurred during the Batch run:", re);

      // this Batch hasn't worked correctly...
      setBatchSuccessful(false);

      // throw to client - this may cause another version of the message
      throw re;
    }
    finally
    {
      if (set != null)
      {
        set.close();
      }
    }

    /**
     * One or more rows being processed have failed so throw a RuntimeException
     * to inform the user of this.
     */
    if (!isBatchSuccessful())
    {
      throw new RuntimeException("An error occurred when processing 1 or more of the rows in the set");
    }
  }

  /**
   * Has the Batch ran successfully?
   * 
   * @return
   */
  public boolean isBatchSuccessful()
  {
    return BatchSuccessful;
  }

  /**
   * Set whether the Batch ran successfully or not
   * 
   * @param BatchSuccessful
   */
  public synchronized void setBatchSuccessful(boolean BatchSuccessful)
  {
    this.BatchSuccessful = BatchSuccessful;
    
    if (!BatchSuccessful)
    {
      // increment the number of failures we've had
      failureCount++;
    }
  }

  /**
   * Get the number of failures that have occurred.
   * 
   * @return
   */
  public long getFailureCount()
  {
    return failureCount;
  }

    private ControlTO markJobAsRunning(String processName)
  {

    Connection conn = null;
    RuntimeException firstException = null;

    ControlTO controlTO = null;

    if (!enforceSingleInstance)
    {
      return new ControlTO();
    }

    try
    {
      conn = getConnection();

      controlTO = getControlEntry(conn, processName);

      if (controlTO == null)
      {
        // this is the 1st time we've encountered this, so add new entry in the
        // control table
        Long maxBatchingId=getMaxBatchingControlID(conn);
        controlTO = newControlEntry(conn,maxBatchingId,processName);
      }
      else
      {
        // calculate whether we think this job has been assumed to have timed
        // out
        if (controlTO.getElapsedRuntime() > assumeJobTimeoutMilliseconds)
        {
          String duration = DurationFormatUtils.formatDurationWords(controlTO.getElapsedRuntime(), true, true);
          String overrideDuration = DurationFormatUtils.formatDurationWords(assumeJobTimeoutMilliseconds, true, true);

          String warningMesssage = "markJobAsRunning(" + controlTO.getProcessName()
              + ") - job has timed out. Running anyway. Job has been running for " + duration + " and timeout period is " + overrideDuration;
          logger.logMessage(Logger.LEVEL_WARN, warningMesssage);

          RuntimeException dummyException = new RuntimeException("BatchingFramework Warning: " + controlTO.getProcessName() + " " + warningMesssage);

        }
        else
        {
          if (controlTO.getEndTimestamp() == null)
          {
            if (logger.isDebugEnabled())
            {
              logger.logMessage(Logger.LEVEL_DEBUG, "markJobAsRunning() returned " + true);
            }
            conn.rollback();

            controlTO.setJobIsRunning(true);

            return controlTO;
          }
        }

        updateControlEntry(conn, controlTO, false);
      }

      conn.commit();
    }
    catch (Exception e)
    {
      firstException = new RuntimeException(e);
    }
    finally
    {
      try
      {
        closeConnection();
      }
    }
    return controlTO;
  }

    /**
   * Mark the job as complete
   */
  private void completeControlEntry(String processName)
  {

    if (!enforceSingleInstance)
    {
      return;
    }

    Connection conn = null;
    RuntimeException firstException = null;

    try
    {
      conn = getConnection();

      ControlTO controlTO = getControlEntry(conn, processName);
      controlTO.setTotalRows(total);
      controlTO.setFailureCount(failureCount);

      updateControlEntry(conn, controlTO, true);

      conn.commit();
    }
    catch (Exception e)
    {
      
    }
    finally
    {
      try
      {
        closeConnection();
      }
      catch (Exception e)
      {
      
      }
    }
  }

  /**
   * Get the control data for a process
   * 
   * @param conn
   * @param processName
   * @return
   */
  private ControlTO getControlEntry(Connection conn, String processName)
  {

    Exception firstException = null;

    ControlTO controlTO = null;

    PreparedStatement ps = null;
    ResultSet rs = null;

    ps = conn.prepareStatement(GET_CONTROL_ENTRY_SQL);
    ps.setString(1, processName);

    //  TODO continue writing code and execute the prepared statement and store all the values in ControlTO

    return controlTO;
  }
  
  private Long getMaxBatchingControlID(Connection conn)
  {

    Exception firstException = null;

    Long maxId=null;

    PreparedStatement ps = null;
    ResultSet rs = null;
    ps = conn.prepareStatement(GET_MAX_ID_SQL);

//  TODO continue writing code and execute the prepared statement

    return maxId;
  }

  /**
   * Add a new control entry for a process
   */
  private ControlTO newControlEntry(Connection conn, Long maxBatchingId,String processName)
  {

    RuntimeException firstException = null;

    PreparedStatement ps = null;

    ps = conn.prepareStatement(ADD_CONTROL_ENTRY_SQL);

    //TODO continue writing code and execute the prepared statement
    
    // now read it back
    ControlTO controlTO = getControlEntry(conn, processName);

    return controlTO;
  }

  /**
   * Update new control entry for a process
   */
  private ControlTO updateControlEntry(Connection conn, ControlTO controlTO, boolean endJob)
  {
    RuntimeException firstException = null;

    PreparedStatement ps = null;
    try
    {
      if (endJob)
      {
        // we're ending job, so mark end_tms with timestamp and add a few
        // metrics
        
        Calendar cal = Calendar.getInstance();
        java.sql.Date jsqlD = new java.sql.Date( cal.getTime().getTime() );
        
        ps = conn.prepareStatement(END_RUN_CONTROL_ENTRY_SQL);
        ps.setDate(1,jsqlD);
        ps.setLong(2, controlTO.getTotalRows());
        ps.setLong(3, controlTO.getFailureCount());
        ps.setString(4, controlTO.getProcessName());
      }
      else
      {
        
        Calendar cal = Calendar.getInstance();
        java.sql.Date jsqlD = new java.sql.Date( cal.getTime().getTime() );

        // we're starting a job, so set new start time and mark end time as null
        // to indicate it's running + reset metrics
        ps = conn.prepareStatement(START_RUN_CONTROL_ENTRY_SQL);
        ps.setDate(1,jsqlD);
        ps.setString(2, controlTO.getProcessName());
      }

      ps.execute();
    }
    catch (SQLException e)
    {
    }
    finally
    {
      try
      {
        ps.close();
      }
      catch (SQLException e)
      {
      }
    }

    return controlTO;
  }

  /**
   * Get a database connection from the datasource
   */
  private Connection getConnection()
  {
   // Write code to return java.sql.Connection 
    return connection;
  }

  /**
   * Return the connection to the datasource
   */
  private void closeConnection()
  {
    // Write code to close Connection
  }

  private static class ControlTO
  {
    private Long id;

    private String processName;

    private Date startTimestamp;

    private Date endTimestamp;

    private Date currentTimestamp;

    private Long totalRows;

    private Long failureCount;

    private boolean jobIsRunning;

   // TODO Add get set methods for all fields 

    public long getElapsedRuntime()
    {
      if (getEndTimestamp() != null)
      {
        // this is a job which has finished
        return getEndTimestamp().getTime() - getStartTimestamp().getTime();
      }

      // this is a job which is still running
      return getCurrentTimestamp().getTime() - getStartTimestamp().getTime();
    }
  }
}

.. TO Be Continued In next article