JDBCRecoveryLog.java

/**
 * C-JDBC: Clustered JDBC.
 * Copyright (C) 2002-2005 French National Institute For Research In Computer
 * Science And Control (INRIA).
 * Contact: c-jdbc@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by the
 * Free Software Foundation; either version 2.1 of the License, or any later
 * version.
 * 
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 *
 * Initial developer(s): Emmanuel Cecchet.
 * Contributor(s): Julie Marguerite, Greg Ward, Jess Sightler.
 */

package org.objectweb.cjdbc.controller.recoverylog;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;

import javax.management.NotCompliantMBeanException;

import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.jmx.mbeans.AbstractRecoveryLogMBean;
import org.objectweb.cjdbc.common.shared.BackendState;
import org.objectweb.cjdbc.common.sql.AbstractRequest;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.AlterRequest;
import org.objectweb.cjdbc.common.sql.CreateRequest;
import org.objectweb.cjdbc.common.sql.DeleteRequest;
import org.objectweb.cjdbc.common.sql.DropRequest;
import org.objectweb.cjdbc.common.sql.InsertRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.sql.UpdateRequest;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.controller.connection.DriverManager;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.BeginTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.CommitTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.ReadStoredProcedureTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.RollbackTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.WriteRequestTask;
import org.objectweb.cjdbc.controller.loadbalancer.tasks.WriteStoredProcedureTask;
import org.objectweb.cjdbc.controller.requestmanager.TransactionMarkerMetaData;

/**
 * Recovery Log using a database accessed through JDBC.
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @author <a href="mailto:Julie.Marguerite@inria.fr">Julie Marguerite </a>
 * @author <a href="mailto:Nicolas.Modrzyk@inrialpes.fr">Nicolas Modrzyk </a>*
 * @version 1.0
 */
public class JDBCRecoveryLog extends AbstractRecoveryLog
    implements
      AbstractRecoveryLogMBean
{
  /** Driver class name. */
  private String            driverClassName;

  /** Driver name. */
  private String            driverName;

  /** Driver URL. */
  private String            url;

  /** User's login. */
  private String            login;

  /** User's password. */
  private String            password;

  /** Name of the log table. */
  private String            logTableName;
  private String            logTableCreateStatement;
  // keep record of values
  private String            idType;
  private String            vloginType;
  private String            sqlType;
  private String            transactionIdType;
  private String            logextraStatement;

  /** Checkpoint table name. */
  private String            checkpointTableName;
  private String            checkpointTableCreateStatement;

  /**
   * Backend table
   * 
   * @see #setBackendTableCreateStatement(String, String, String, String,
   *      String, String)
   */
  private String            backendTableCreateStatement;
  private String            backendTableName;

  // Keep record of the values
  private String            nameType;
  private String            requestIdType;
  private String            checkextraStatement;

  /** Connection to the database. */
  private Connection        connection;

  /** <code>PreparedStatement</code> used to log requests. */
  private PreparedStatement pstmt;

  /** Current maximum value of the primary key in logTableName. */
  private long              logTableId = 0;

  /** Timeout for SQL requests. */
  private int               timeout;

  private JDBCLoggerThread  loggerThread;

  private String            sqlColumnName;

  /**
   * Creates a new <code>JDBCRecoveryLog</code> instance.
   * 
   * @param driverName the driverClassName name.
   * @param driverClassName the driverClassName class name.
   * @param url the JDBC URL.
   * @param login the login to use to connect to the database.
   * @param password the password to connect to the database.
   * @param requestTimeout timeout in seconds for update queries.
   * @throws NotCompliantMBeanException if the MBean is not JMX compliant
   */
  public JDBCRecoveryLog(String driverName, String driverClassName, String url,
      String login, String password, int requestTimeout)
      throws NotCompliantMBeanException
  {
    super(AbstractRecoveryLogMBean.class);
    this.driverName = driverName;
    this.driverClassName = driverClassName;
    this.url = url;
    this.login = login;
    this.password = password;
    this.timeout = requestTimeout;

    // Connect to the database
    try
    {
      connectToDatabase();
    }
    catch (SQLException e)
    {
      throw new RuntimeException("Unable to connect to the database: " + e);
    }

    // Logger thread will be created in checkRecoveryLogTables()
    // after database has been initialized
  }

  /**
   * Checks if the tables (log and checkpoint) already exist, and create them if
   * needed.
   */
  private void intializeDatabase() throws SQLException
  {
    boolean createLogTable = true;
    boolean createCheckpointTable = true;
    boolean createBackendTable = true;
    // Check if tables exist
    try
    {
      connection.setAutoCommit(false);
      // Get DatabaseMetaData
      DatabaseMetaData metaData = connection.getMetaData();

      // Get a description of tables matching the catalog, schema, table name
      // and type.
      // Sending in null for catalog and schema drop them from the selection
      // criteria. Replace the last argument in the getTables method with
      // types below to obtain only database tables. (Sending in null
      // retrieves all types).
      String[] types = {"TABLE", "VIEW"};
      ResultSet rs = metaData.getTables(null, null, "%", types);

      // Get tables metadata
      String tableName;
      while (rs.next())
      {
        // 1 is table catalog, 2 is table schema, 3 is table name, 4 is type
        tableName = rs.getString(3);
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("recovery.jdbc.table.found", tableName));
        if (tableName.equalsIgnoreCase(logTableName))
        {
          if (tableName.compareTo(logTableName) != 0)
            logger.warn(Translate.get("recovery.jdbc.logtable.case.mismatch",
                new String[]{logTableName, tableName}));
          createLogTable = false;
          // initialize logTableId
          PreparedStatement p = null;
          try
          {
            ResultSet result = null;
            p = connection.prepareStatement("SELECT MAX(id) AS max_id FROM "
                + logTableName);
            result = p.executeQuery();
            if (result.next())
              logTableId = result.getInt("max_id");
            else
              logTableId = 0;
            p.close();
          }
          catch (SQLException e)
          {
            try
            {
              if (p != null)
                p.close();
            }
            catch (Exception ignore)
            {
            }
            throw new RuntimeException(Translate.get(
                "recovery.jdbc.logtable.getvalue.failed", e));
          }

        }
        if (tableName.equalsIgnoreCase(checkpointTableName))
        {
          if (tableName.compareTo(checkpointTableName) != 0)
            logger.warn(Translate.get(
                "recovery.jdbc.checkpointtable.case.mismatch", new String[]{
                    checkpointTableName, tableName}));
          createCheckpointTable = false;
        }
        if (tableName.equalsIgnoreCase(backendTableName))
        {
          if (tableName.compareTo(backendTableName) != 0)
            logger.warn(Translate.get(
                "recovery.jdbc.backendtable.case.mismatch", new String[]{
                    backendTableName, tableName}));
          createBackendTable = false;
        }
      }
      try
      {
        connection.commit();
        connection.setAutoCommit(true);
      }
      catch (Exception ignore)
      {
        // Read-only transaction we don't care
      }
    }
    catch (SQLException e)
    {
      logger.error(Translate.get("recovery.jdbc.table.no.description"), e);
      throw e;
    }

    // Create the missing tables
    Statement stmt = null;
    if (createLogTable)
    {
      if (logger.isInfoEnabled())
        logger.info(Translate
            .get("recovery.jdbc.logtable.create", logTableName));
      try
      {
        stmt = connection.createStatement();
        stmt.executeUpdate(logTableCreateStatement);
        stmt.close();
      }
      catch (SQLException e)
      {
        throw new SQLException(Translate.get(
            "recovery.jdbc.logtable.create.failed", new String[]{logTableName,
                e.getMessage()}));
      }
    }
    if (createCheckpointTable)
    {
      if (logger.isInfoEnabled())
        logger.info(Translate.get("recovery.jdbc.checkpointtable.create",
            checkpointTableName));
      try
      {
        stmt = connection.createStatement();
        stmt.executeUpdate(checkpointTableCreateStatement);
        stmt.close();
      }
      catch (SQLException e)
      {
        throw new SQLException(Translate.get(
            "recovery.jdbc.checkpointtable.create.failed", new String[]{
                logTableName, e.getMessage()}));
      }

      // Add an initial checkpoint in the table
      String checkpointName = "Initial_empty_recovery_log";
      try
      {
        if (logger.isDebugEnabled())
          logger.debug("Storing checkpoint " + checkpointName
              + " at request id " + logTableId);
        pstmt = connection.prepareStatement("INSERT INTO "
            + checkpointTableName + " VALUES(?,?)");
        pstmt.setString(1, checkpointName);
        pstmt.setLong(2, logTableId);
        pstmt.executeUpdate();
        pstmt.close();
      }
      catch (SQLException e)
      {
        try
        {
          if (pstmt != null)
            pstmt.close();
        }
        catch (Exception ignore)
        {
        }
        throw new SQLException(Translate.get(
            "recovery.jdbc.checkpoint.store.failed", new String[]{
                checkpointName, e.getMessage()}));
      }
    }
    if (createBackendTable)
    {
      if (logger.isInfoEnabled())
        logger.info(Translate.get("recovery.jdbc.backendtable.create",
            backendTableName));
      try
      {
        stmt = connection.createStatement();
        stmt.executeUpdate(backendTableCreateStatement);
        stmt.close();
      }
      catch (SQLException e)
      {
        throw new SQLException(Translate.get(
            "recovery.jdbc.backendtable.create.failed", new String[]{
                logTableName, e.getMessage()}));
      }
    }

  }

  /**
   * Gets a connection to the database.
   * 
   * @exception SQLException if an error occurs.
   */
  private void connectToDatabase() throws SQLException
  {
    // Get a connection
    try
    {
      if (logger.isDebugEnabled())
        logger.debug(Translate.get("recovery.jdbc.connect", new String[]{url,
            login}));
      connection = DriverManager.getConnection(url, login, password,
          driverName, driverClassName);
    }
    catch (RuntimeException e)
    {
      String msg = Translate.get("recovery.jdbc.connect.failed", e);
      logger.debug(msg, e);
      throw new SQLException(msg);
    }
    catch (SQLException e)
    {
      String msg = Translate.get("recovery.jdbc.connect.failed", e);
      logger.debug(msg, e);
      throw new SQLException(msg);
    }
  }

  //
  //
  // Logging related methods
  //
  //

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#logRequest(AbstractWriteRequest)
   */
  public void logRequest(AbstractWriteRequest request) throws SQLException
  {
    loggerThread.log(incrementLogTableId(), request.getLogin(), request
        .getSQL(), request.getTransactionId(), request.getEscapeProcessing());
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#logRequest(org.objectweb.cjdbc.common.sql.StoredProcedure,boolean)
   */
  public void logRequest(StoredProcedure proc, boolean isRead)
      throws SQLException
  {
    if (isRead)
      loggerThread.log(incrementLogTableId(), proc.getLogin(), proc.getSQL(),
          proc.getTransactionId(), proc.getEscapeProcessing());
    else
    { // Reverse the first bracket so that we can identify a write call
      StringBuffer writeCall = new StringBuffer(proc.getSQL());
      writeCall.setCharAt(0, '}');
      loggerThread.log(incrementLogTableId(), proc.getLogin(), writeCall
          .toString(), proc.getTransactionId(), proc.getEscapeProcessing());
    }
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#getLastTransactionId()
   */
  public long getLastTransactionId() throws SQLException
  {
    Statement stmt = null;
    ResultSet rs = null;
    try
    {
      stmt = connection.createStatement();
      rs = stmt.executeQuery("select max(transaction_id) from " + logTableName);
      if (rs.next())
        return rs.getInt(1);
      else
        return 0;
    }
    catch (SQLException e)
    {
      throw e;
    }
    finally
    {
      try
      {
        if (rs != null)
          rs.close();
      }
      catch (Exception ignore)
      {
      }
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
    }
  }

  /**
   * Increments the value of logTableId.
   */
  private synchronized long incrementLogTableId()
  {
    logTableId++;
    return logTableId;
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#begin(TransactionMarkerMetaData)
   */
  public void begin(TransactionMarkerMetaData tm) throws SQLException
  {
    // Store the begin in the database
    loggerThread.log(incrementLogTableId(), tm.getLogin(), "begin", tm
        .getTransactionId(), false);
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#abort(org.objectweb.cjdbc.controller.requestmanager.TransactionMarkerMetaData)
   */
  public void abort(TransactionMarkerMetaData tm) throws SQLException
  {
    // We have to perform exactly the same job as a rollback
    rollback(tm);
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#commit(TransactionMarkerMetaData)
   */
  public void commit(TransactionMarkerMetaData tm) throws SQLException
  {
    // Some backends started a recovery process, log the commit
    loggerThread.log(incrementLogTableId(), tm.getLogin(), "commit", tm
        .getTransactionId(), false);
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#rollback(TransactionMarkerMetaData)
   */
  public void rollback(TransactionMarkerMetaData tm) throws SQLException
  {
    long transactionId = tm.getTransactionId();
    if (isRecovering())
    {
      // Some backends started a recovery process, log the rollback
      loggerThread.log(incrementLogTableId(), tm.getLogin(), "rollback",
          transactionId, false);
    }
    else
    {
      // The transaction failed
      // Remove the requests with this transactionId from the database
      loggerThread.removeQueriesOfTransactionFromQueue(transactionId);
      PreparedStatement stmt = null;
      try
      {
        stmt = connection.prepareStatement("DELETE FROM " + logTableName
            + " WHERE transaction_id=?");
        stmt.setLong(1, transactionId);
        stmt.executeUpdate();
      }
      catch (SQLException e)
      {
        throw new SQLException(Translate.get(
            "recovery.jdbc.transaction.remove.failed", new String[]{
                String.valueOf(transactionId), e.getMessage()}));
      }
      finally
      {
        try
        {
          if (stmt != null)
            stmt.close();
        }
        catch (Exception ignore)
        {
        }
      }
    }
  }

  //
  //
  // Recovery related functions
  //
  //

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#getCheckpointRequestId(String)
   */
  public long getCheckpointRequestId(String checkpointName) throws SQLException
  {
    long requestId = -1;
    PreparedStatement stmt = null;
    ResultSet rs = null;
    try
    {
      stmt = connection.prepareStatement("SELECT request_id FROM "
          + checkpointTableName + " WHERE name LIKE ?");
      stmt.setString(1, checkpointName);
      rs = stmt.executeQuery();

      if (rs.next())
        requestId = rs.getInt("request_id");
      else
      {
        String msg = Translate.get("recovery.jdbc.checkpoint.not.found",
            checkpointName);
        logger.info(msg);
        throw new SQLException(msg);
      }
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.not.found.error", new String[]{
              checkpointName, e.getMessage()}));
    }
    finally
    {
      try
      {
        if (rs != null)
          rs.close();
      }
      catch (Exception ignore)
      {
      }
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
    }
    return requestId;
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#recoverNextRequest(long)
   */
  public RecoveryTask recoverNextRequest(long previousRequestId)
      throws SQLException
  {
    RecoveryTask task = null;

    // Get the request with the id after previousRequestId.
    PreparedStatement stmt = null;
    try
    {
      ResultSet rs = null;
      boolean emptyResult;

      stmt = connection.prepareStatement("SELECT * FROM " + logTableName
          + " WHERE id>? AND id<=? ORDER BY id");
      // Note that the statement is closed in the finally block
      do
      {
        // Take a window of 2 requests if there is a small hole
        // induced by a removed begin/commit/rollback command.
        stmt.setLong(1, previousRequestId);
        stmt.setLong(2, previousRequestId + 2);
        // Close ResultSet of previous loop step to free resources.
        if (rs != null)
          rs.close();
        rs = stmt.executeQuery();
        previousRequestId += 2; // Shift the window
        emptyResult = !rs.next();
      }
      while (emptyResult && (previousRequestId <= logTableId));

      // No more request after this one
      if (emptyResult)
      {
        rs.close();
        return null;
      }

      String sql = rs.getString(sqlColumnName);
      String user = rs.getString("vlogin");
      int transactionId = rs.getInt("transaction_id");
      int id = rs.getInt("id");
      rs.close();

      // Construct the request object according to its type
      boolean escapeProcessing = true;
      sql = sql.trim();
      // Check that the command starts with (only 2 letters are needed)
      // in[sert]/up[date]/de[lete]/cr[eate]/dr[op]/be[gin]/co[mmit]/ro[llback]/{c[all]/}c[all]
      // (write stored procedure call)
      String lower = sql.substring(0, 2).toLowerCase();
      if (lower.equals("in"))
      { // insert
        AbstractWriteRequest wr = new InsertRequest(sql, escapeProcessing,
            timeout, "\n");
        wr.setLogin(user);
        if (logger.isDebugEnabled())
          logger.debug("insert request: " + sql);
        setDriverProcessedAndSkeleton(wr);
        if (transactionId != 0)
        {
          wr.setIsAutoCommit(false);
          wr.setTransactionId(transactionId);
        }
        else
          wr.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new WriteRequestTask(1, 1,
            wr));
      }
      else if (lower.equals("up"))
      { // update
        AbstractWriteRequest wr = new UpdateRequest(sql, escapeProcessing,
            timeout, "\n");
        wr.setLogin(user);
        setDriverProcessedAndSkeleton(wr);
        if (logger.isDebugEnabled())
          logger.debug("update request: " + sql);
        if (transactionId != 0)
        {
          wr.setIsAutoCommit(false);
          wr.setTransactionId(transactionId);
        }
        else
          wr.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new WriteRequestTask(1, 1,
            wr));
      }
      else if (lower.equals("de"))
      { // delete
        AbstractWriteRequest wr = new DeleteRequest(sql, escapeProcessing,
            timeout, "\n");
        wr.setLogin(user);
        setDriverProcessedAndSkeleton(wr);
        if (logger.isDebugEnabled())
          logger.debug("delete request: " + sql);
        if (transactionId != 0)
        {
          wr.setIsAutoCommit(false);
          wr.setTransactionId(transactionId);
        }
        else
          wr.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new WriteRequestTask(1, 1,
            wr));
      }
      else if (lower.equals("cr"))
      { // create
        AbstractWriteRequest wr = new CreateRequest(sql, escapeProcessing,
            timeout, "\n");
        wr.setLogin(user);
        setDriverProcessedAndSkeleton(wr);
        if (logger.isDebugEnabled())
          logger.debug("create request: " + sql);
        if (transactionId != 0)
        {
          wr.setIsAutoCommit(false);
          wr.setTransactionId(transactionId);
        }
        else
          wr.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new WriteRequestTask(1, 1,
            wr));
      }
      else if (lower.equals("al"))
      { // alter
        AbstractWriteRequest wr = new AlterRequest(sql, escapeProcessing,
            timeout, "\n");
        wr.setLogin(user);
        setDriverProcessedAndSkeleton(wr);
        if (logger.isDebugEnabled())
          logger.debug("alter request: " + sql);
        if (transactionId != 0)
        {
          wr.setIsAutoCommit(false);
          wr.setTransactionId(transactionId);
        }
        else
          wr.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new WriteRequestTask(1, 1,
            wr));
      }
      else if (lower.equals("dr"))
      { // drop
        AbstractWriteRequest wr = new DropRequest(sql, escapeProcessing,
            timeout, "\n");
        wr.setLogin(user);
        setDriverProcessedAndSkeleton(wr);
        if (logger.isDebugEnabled())
          logger.debug("drop request: " + sql);
        if (transactionId != 0)
        {
          wr.setIsAutoCommit(false);
          wr.setTransactionId(transactionId);
        }
        else
          wr.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new WriteRequestTask(1, 1,
            wr));
      }
      else if (lower.equals("be"))
      { // begin
        task = new RecoveryTask(transactionId, id, new BeginTask(1, 1,
            (long) timeout * 1000, user, transactionId));
        if (logger.isDebugEnabled())
          logger.debug("begin transaction: " + transactionId);
      }
      else if (lower.equals("co"))
      { // commit
        task = new RecoveryTask(transactionId, id, new CommitTask(1, 1,
            (long) timeout * 1000, user, transactionId));
        if (logger.isDebugEnabled())
          logger.debug("commit transaction: " + transactionId);
      }
      else if (lower.equals("ro"))
      { // rollback
        task = new RecoveryTask(transactionId, id, new RollbackTask(1, 1,
            (long) timeout * 1000, user, transactionId));
        if (logger.isDebugEnabled())
          logger.debug("rollback transaction: " + transactionId);
      }
      else if (lower.equals("{c"))
      { // read stored procedure call "{call ...}"
        StoredProcedure proc = new StoredProcedure(sql, escapeProcessing,
            timeout, "\n");
        proc.setLogin(user);
        setDriverProcessedAndSkeleton(proc);
        if (logger.isDebugEnabled())
          logger.debug("read stored procedure call: " + sql);
        if (transactionId != 0)
        {
          proc.setIsAutoCommit(false);
          proc.setTransactionId(transactionId);
        }
        else
          proc.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id, new ReadStoredProcedureTask(
            1, 1, proc, null));
      }
      else if (lower.equals("}c"))
      { // write stored procedure call,
        // we must replace "}call ...}" with "{call ...}"
        StringBuffer writeCall = new StringBuffer(sql);
        writeCall.setCharAt(0, '{');
        StoredProcedure proc = new StoredProcedure(writeCall.toString(),
            escapeProcessing, timeout, "\n");
        proc.setLogin(user);
        setDriverProcessedAndSkeleton(proc);
        if (logger.isDebugEnabled())
          logger.debug("write stored procedure call: " + sql);
        if (transactionId != 0)
        {
          proc.setIsAutoCommit(false);
          proc.setTransactionId(transactionId);
        }
        else
          proc.setIsAutoCommit(true);
        task = new RecoveryTask(transactionId, id,
            new WriteStoredProcedureTask(1, 1, proc));
      }
      else
        throw new SQLException(Translate.get("recovery.jdbc.sql.unkwown", sql));
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get("recovery.jdbc.recover.failed", e));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
    }
    return task;
  }

  /**
   * Set the driverProcessed flag of the given request according to its SQL
   * content and rebuild the SQL skeleton if necessary.
   * 
   * @param request Request to process
   */
  private void setDriverProcessedAndSkeleton(AbstractRequest request)
  {
    String sql = request.getSQL();
    boolean isDriverProcessed = sql
        .indexOf(org.objectweb.cjdbc.driver.Connection.END_PARAM_TAG) == -1;
    request.setDriverProcessed(isDriverProcessed);
    if (isDriverProcessed)
      return; // No need to set the skeleton

    request.setSqlSkeleton(recreateSkeleton(sql));
  }

  /**
   * Recreate the skeleton of the query from the query itself with the param tag
   * 
   * @param sql the sql query as it is recorded in the recovery log
   * @return the sql skelton with '?' instead of the <?...|...?>tags
   */
  private String recreateSkeleton(String sql)
  {
    // Here we have to rebuild the query skeleton to be able to call setXXX on
    // the PreparedStatement
    StringBuffer skeleton = new StringBuffer();
    int start = 0;
    int end;
    while ((end = sql.indexOf(
        org.objectweb.cjdbc.driver.Connection.START_PARAM_TAG, start)) != -1)
    {
      skeleton.append(sql.substring(start, end)).append('?');
      start = sql.indexOf(org.objectweb.cjdbc.driver.Connection.END_PARAM_TAG,
          end);
      if (start == -1)
        throw new RuntimeException("Malformed query in recovery log: " + sql);
      else
        start += org.objectweb.cjdbc.driver.Connection.END_PARAM_TAG.length();
    }
    if (start < sql.length())
      skeleton.append(sql.substring(start));
    return skeleton.toString();
  }

  /**
   * Removes all rollbacked transaction from the recovery log.
   * 
   * @exception SQLException if an error occurs.
   */
  public void cleanRecoveryLog() throws SQLException
  {
    PreparedStatement stmt = null;

    // Remove the rollback statements and associated requests from the database
    ResultSet rs = null;
    try
    {
      // Get the list of transaction ids on which a rollback occurred
      stmt = connection.prepareStatement("SELECT transaction_id FROM "
          + logTableName + " WHERE sql LIKE ?");
      stmt.setString(1, "rollback");
      rs = stmt.executeQuery();
    }
    catch (SQLException e)
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
      throw new SQLException("Unable get rollback statements : " + e);
    }
    PreparedStatement pstmt = null;
    long transactionId = -1;
    try
    {
      // remove the rollbacked transaction from the database
      while (rs.next())
      {
        transactionId = rs.getLong("transaction_Id");
        pstmt = connection.prepareStatement("DELETE FROM " + logTableName
            + " WHERE transaction_id=?");
        pstmt.setLong(1, transactionId);
        pstmt.executeUpdate();
        pstmt.close();
      }
      rs.close();
      stmt.close();
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.transaction.remove.failed", new String[]{
              String.valueOf(transactionId), e.getMessage()}));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
      try
      {
        if (pstmt != null)
          pstmt.close();
      }
      catch (Exception ignore)
      {
      }

    }
  }

  //
  //
  // Checkpoints management
  //
  //

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#getCheckpointNames()
   */
  public ArrayList getCheckpointNames() throws SQLException
  {
    PreparedStatement stmt = null;

    // Wait for transaction to finish
    waitForTransactionsEnd(true);

    try
    {
      if (logger.isDebugEnabled())
        logger.debug("Retrieving checkpoint names list");
      stmt = connection.prepareStatement("SELECT name from "
          + checkpointTableName);
      ResultSet rs = stmt.executeQuery();
      ArrayList list = new ArrayList();
      while (rs.next())
      {
        list.add(rs.getString(1));
      }
      rs.close();
      return list;
    }
    catch (Exception e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.list.failed", e));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (SQLException ignore)
      {
      }
    }
  }

  /**
   * Checks if a checkpoint with the name checkpointName is already stored in
   * the database.
   * 
   * @param checkpointName name of the checkpoint.
   * @return ResultSet empty if no checkpoint was found.
   */
  private boolean validCheckpointName(String checkpointName)
      throws SQLException
  {
    PreparedStatement stmt = null;
    ResultSet rs = null;
    try
    {
      stmt = connection.prepareStatement("SELECT * FROM " + checkpointTableName
          + " WHERE name LIKE ?");
      stmt.setString(1, checkpointName);
      rs = stmt.executeQuery();

      // If the query returned any rows, the checkpoint name is already
      // in use and therefore invalid.
      boolean checkpointExists = rs.next();
      rs.close();
      return !checkpointExists;
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.check.failed", e));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (SQLException ignore)
      {
      }
    }
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#storeCheckpoint(String)
   */
  public void storeCheckpoint(String checkpointName) throws SQLException
  {
    storeCheckpoint(checkpointName, logTableId);
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#storeCheckpoint(String,
   *      long)
   */
  public void storeCheckpoint(String checkpointName, long requestId)
      throws SQLException
  {
    PreparedStatement stmt = null;

    // Check if a checkpoint with the name checkpointName already exists
    if (!validCheckpointName(checkpointName))
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.duplicate", checkpointName));
    }

    // Wait for transaction to finish
    waitForTransactionsEnd(true);

    try
    {
      if (logger.isDebugEnabled())
        logger.debug("Storing checkpoint " + checkpointName + " at request id "
            + requestId);
      stmt = connection.prepareStatement("INSERT INTO " + checkpointTableName
          + " VALUES(?,?)");
      stmt.setString(1, checkpointName);
      stmt.setLong(2, requestId);
      stmt.executeUpdate();
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.store.failed", new String[]{checkpointName,
              e.getMessage()}));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
    }
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#removeCheckpoint(java.lang.String)
   */
  public void removeCheckpoint(String checkpointName) throws SQLException
  {
    PreparedStatement stmt = null;

    // Wait for transaction to finish
    waitForTransactionsEnd(true);

    try
    {
      // 1. Get the reference point to delete
      stmt = connection.prepareStatement("SELECT * FROM " + checkpointTableName
          + " WHERE name LIKE ?");
      stmt.setString(1, checkpointName);
      ResultSet rs = stmt.executeQuery();
      boolean checkpointExists = rs.next();
      if (!checkpointExists)
      {
        rs.close();
        stmt.close();
        throw new SQLException("Checkpoint " + checkpointName
            + " does not exist");
      }

      int requestId = rs.getInt("request_id");
      rs.close();
      stmt.close();

      // Delete all entries below
      stmt = connection.prepareStatement("DELETE FROM " + logTableName
          + " WHERE ID <= ?");
      stmt.setInt(1, requestId);
      stmt.executeUpdate();
      stmt.close();

      // Delete checkpoint name
      stmt = connection.prepareStatement("DELETE FROM " + checkpointTableName
          + " WHERE name like ?");
      stmt.setString(1, checkpointName);
      stmt.executeUpdate();
      stmt.close();
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.remove.failed", new String[]{
              checkpointName, e.getMessage()}));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }

    }

  }

  /**
   * Wait for current transactions to complete. Current transactions can be
   * forced to be rollbacked if necessary.
   * 
   * @param forceRollback true if currently open transactions must be
   *          rollbacked.
   */
  private void waitForTransactionsEnd(boolean forceRollback)
  {
    if (forceRollback)
    {
      loggerThread.rollbackTransactions();
    }
    else
    {
      synchronized (loggerThread)
      {
        while (!loggerThread.getLogQueueIsEmpty())
        {
          try
          {
            wait();
          }
          catch (Exception e)
          {
            logger.warn("Exception " + e
                + " while waiting for end of transactions");
          }
        }
      }
    }
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#getBackendRecoveryInfo(java.lang.String,
   *      java.lang.String)
   */
  public BackendRecoveryInfo getBackendRecoveryInfo(String databaseName,
      String backendName)
  {
    PreparedStatement stmt = null;
    String checkpoint = null;
    int backendState = BackendState.UNKNOWN;
    try
    {
      // 1. Get the reference point to delete
      stmt = connection.prepareStatement("SELECT * FROM " + backendTableName
          + " WHERE backendName LIKE ? AND databaseName LIKE ?");
      stmt.setString(1, backendName);
      stmt.setString(2, databaseName);
      ResultSet rs = stmt.executeQuery();

      if (rs.next())
      {
        checkpoint = rs.getString("checkpointName");
        backendState = rs.getInt("backendState");
      }
      rs.close();
    }
    catch (SQLException e)
    {
      logger.info(
          "An error occured while retrieving backend recovery information", e);
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
    }
    return new BackendRecoveryInfo(backendName, checkpoint, backendState,
        databaseName);
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#storeBackendRecoveryInfo(String,
   *      BackendRecoveryInfo)
   */
  public void storeBackendRecoveryInfo(String databaseName,
      BackendRecoveryInfo backendRecoveryInfo) throws SQLException
  {
    PreparedStatement stmt = null;
    PreparedStatement stmt2 = null;
    if ((backendRecoveryInfo.getCheckpoint() == null)
        || ((backendRecoveryInfo.getBackendState() != BackendState.DISABLED) && (backendRecoveryInfo
            .getBackendState() != BackendState.UNKNOWN)))
      backendRecoveryInfo.setCheckpoint(""); // No checkpoint
    else
    { // Check checkpoint name validity
      getCheckpointRequestId(backendRecoveryInfo.getCheckpoint());
    }

    try
    {
      // 1. Get the reference point to delete
      stmt = connection.prepareStatement("SELECT * FROM " + backendTableName
          + " WHERE backendName LIKE ? and databaseName LIKE ?");
      stmt.setString(1, backendRecoveryInfo.getBackendName());
      stmt.setString(2, databaseName);
      ResultSet rs = stmt.executeQuery();
      boolean mustUpdate = rs.next();
      rs.close();
      if (!mustUpdate)
      {
        stmt2 = connection.prepareStatement("INSERT INTO " + backendTableName
            + " values(?,?,?,?)");
        stmt2.setString(1, databaseName);
        stmt2.setString(2, backendRecoveryInfo.getBackendName());
        stmt2.setInt(3, backendRecoveryInfo.getBackendState());
        stmt2.setString(4, backendRecoveryInfo.getCheckpoint());
        if (stmt2.executeUpdate() != 1)
          throw new SQLException(
              "Error while inserting new backend reference. Incorrect number of rows");
      }
      else
      {
        stmt2 = connection
            .prepareStatement("UPDATE "
                + backendTableName
                + " set backendState=?,checkpointName=? where backendName=? and databaseName=?");
        stmt2.setInt(1, backendRecoveryInfo.getBackendState());
        stmt2.setString(2, backendRecoveryInfo.getCheckpoint());
        stmt2.setString(3, backendRecoveryInfo.getBackendName());
        stmt2.setString(4, databaseName);
        if (stmt2.executeUpdate() != 1)
          throw new SQLException(
              "Error while updating backend reference. Incorrect number of rows");
      }
    }
    catch (SQLException e)
    {
      throw new SQLException("Unable to update checkpoint '"
          + backendRecoveryInfo.getCheckpoint() + "' for backend:"
          + backendRecoveryInfo.getBackendName());
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
      try
      {
        if (stmt2 != null)
          stmt2.close();
      }
      catch (Exception ignore)
      {
      }
    }
  }

  //
  // 
  // Recovery log database tables management
  //
  //

  /**
   * Checks if the recovery log and checkpoint tables exist, and create them if
   * they do not exist. This method also starts the logger thread.
   */
  public void checkRecoveryLogTables()
  {
    try
    {
      intializeDatabase();
      pstmt = connection.prepareStatement("INSERT INTO " + logTableName
          + " VALUES(?,?,?,?)");
    }
    catch (SQLException e)
    {
      throw new RuntimeException("Unable to initialize the database: " + e);
    }

    // Start the logger thread
    loggerThread = new JDBCLoggerThread(pstmt, logger);
    loggerThread.start();
  }

  /**
   * Returns the backendTableName value.
   * 
   * @return Returns the backendTableName.
   */
  public String getBackendTableName()
  {
    return backendTableName;
  }

  /**
   * Returns the checkpointTableName value.
   * 
   * @return Returns the checkpointTableName.
   */
  public String getCheckpointTableName()
  {
    return checkpointTableName;
  }

  /**
   * Returns the logTableName value.
   * 
   * @return Returns the logTableName.
   */
  public String getLogTableName()
  {
    return logTableName;
  }

  /**
   * Sets the backend table create statement
   * 
   * @param tableName the backend table name
   * @param checkpointNameType type for the checkpointName column
   * @param backendNameType type for the backendName column
   * @param backendStateType type for the backendState column
   * @param databaseNameType type for the databaseName column
   * @param extraStatement like primary keys
   */
  public void setBackendTableCreateStatement(String tableName,
      String checkpointNameType, String backendNameType,
      String backendStateType, String databaseNameType, String extraStatement)
  {
    this.backendTableName = tableName;
    this.backendTableCreateStatement = "CREATE TABLE " + backendTableName
        + " (databaseName " + databaseNameType + ", backendName "
        + backendNameType + ",backendState " + backendStateType
        + ", checkpointName " + checkpointNameType + " " + extraStatement + ")";

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("recovery.jdbc.backendtable.statement",
          backendTableCreateStatement));
  }

  /**
   * Sets the checkpoint table name and create statement.
   * 
   * @param tableName name of the checkpoint table.
   * @param nameType type for the name column
   * @param requestIdType type for the requestId column
   * @param extraStatement like primary keys
   */
  public void setCheckpointTableCreateStatement(String tableName,
      String nameType, String requestIdType, String extraStatement)
  {
    this.checkpointTableName = tableName;
    this.nameType = nameType;
    this.requestIdType = requestIdType;
    this.checkextraStatement = extraStatement;
    // CREATE TABLE tableName (
    //   name checkpointNameColumnType,
    //   request_id requestIdColumnType,
    //   extraStatement)

    String checkpointTableCreateStatement = "CREATE TABLE " + tableName
        + " (name " + nameType + ",request_id " + requestIdType
        + extraStatement + ")";
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("recovery.jdbc.checkpointtable.statement",
          checkpointTableCreateStatement));

    this.checkpointTableCreateStatement = checkpointTableCreateStatement;
  }

  /**
   * Sets the log table name and create statement.
   * 
   * @param tableName of the log table
   * @param idType of the id column
   * @param vloginType of the login column
   * @param sqlName name of the sql statement column
   * @param sqlType of the sql column
   * @param transactionIdType of the transaction column
   * @param extraStatement like primary keys ...
   */
  public void setLogTableCreateStatement(String tableName, String idType,
      String vloginType, String sqlName, String sqlType,
      String transactionIdType, String extraStatement)
  {
    this.logTableName = tableName;
    this.idType = idType;
    this.vloginType = vloginType;
    this.sqlColumnName = sqlName;
    this.sqlType = sqlType;
    this.transactionIdType = transactionIdType;
    this.logextraStatement = extraStatement;
    String logTableCreateStatement = "CREATE TABLE " + tableName + " (id "
        + idType + ",vlogin " + vloginType + "," + sqlColumnName + " "
        + sqlType + ",transaction_id " + transactionIdType + extraStatement
        + ")";
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("recovery.jdbc.logtable.statement",
          logTableCreateStatement));

    this.logTableCreateStatement = logTableCreateStatement;
  }

  //
  //
  // Info/Monitoring/Debug related functions
  //
  //

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#getData()
   */
  public String[][] getData()
  {
    Statement stmt = null;
    ResultSet rs = null;
    try
    {
      stmt = connection.createStatement();
      rs = stmt.executeQuery("select * from " + logTableName);
      ArrayList list = new ArrayList();
      while (rs.next())
      {
        // 3: Query 2: User 1: ID 4: TID
        list.add(new String[]{rs.getString(3), rs.getString(2),
            rs.getString(1), rs.getString(4)});
      }
      String[][] result = new String[list.size()][4];
      for (int i = 0; i < list.size(); i++)
        result[i] = (String[]) list.get(i);
      return result;
    }
    catch (SQLException e)
    {
      return null;
    }
    finally
    {
      try
      {
        rs.close();
      }
      catch (SQLException ignore)
      {
      }
      try
      {
        stmt.close();
      }
      catch (SQLException ignore)
      {
      }
    }
  }

  /**
   * @see org.objectweb.cjdbc.controller.jmx.AbstractStandardMBean#getAssociatedString()
   */
  public String getAssociatedString()
  {
    return "jdbcrecoverylog";
  }

  /**
   * @see org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog#getXmlImpl()
   */
  public String getXmlImpl()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_JDBCRecoveryLog + " "
        + DatabasesXmlTags.ATT_driver + "=\"" + driverClassName + "\" "
        + DatabasesXmlTags.ATT_url + "=\"" + url + "\" ");
    if (driverName != null)
    {
      info.append(DatabasesXmlTags.ATT_driverPath + "=\"" + driverName + "\" ");
    }
    info.append(DatabasesXmlTags.ATT_login + "=\"" + login + "\" "
        + DatabasesXmlTags.ATT_password + "=\"" + password + "\" "
        + DatabasesXmlTags.ATT_requestTimeout + "=\"" + (timeout / 1000)
        + "\">");
    // Recovery Log table
    info.append("<" + DatabasesXmlTags.ELT_RecoveryLogTable + " "
        + DatabasesXmlTags.ATT_tableName + "=\"" + logTableName + "\"" + " "
        + DatabasesXmlTags.ATT_idColumnType + "=\"" + idType + "\"" + " "
        + DatabasesXmlTags.ATT_vloginColumnType + "=\"" + vloginType + "\""
        + " " + DatabasesXmlTags.ATT_sqlColumnType + "=\"" + sqlType + "\""
        + " " + DatabasesXmlTags.ATT_transactionIdColumnType + "=\""
        + transactionIdType + "\"" + " "
        + DatabasesXmlTags.ATT_extraStatementDefinition + "=\""
        + logextraStatement + "\"/>");
    // Checkpoint table
    info.append("<" + DatabasesXmlTags.ELT_CheckpointTable + " "
        + DatabasesXmlTags.ATT_tableName + "=\"" + checkpointTableName + "\""
        + " " + DatabasesXmlTags.ATT_checkpointNameColumnType + "=\""
        + nameType + "\"" + " " + DatabasesXmlTags.ATT_requestIdColumnType
        + "=\"" + requestIdType + "\"" + " "
        + DatabasesXmlTags.ATT_extraStatementDefinition + "=\""
        + checkextraStatement + "\"" + "/>");
    // BackendLog table
    info.append("<" + DatabasesXmlTags.ELT_BackendTable + " "
        + DatabasesXmlTags.ATT_tableName + "=\"" + backendTableName + "\""
        + "/>");

    info.append("</" + DatabasesXmlTags.ELT_JDBCRecoveryLog + ">");

    return info.toString();
  }

}

---