src/org/objectweb/cjdbc/controller/recoverylog/JDBCRecoveryLog.java

説明を見る。

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 org.objectweb.cjdbc.common.i18n.Translate;
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;

public class JDBCRecoveryLog extends AbstractRecoveryLog
{
  private String            driverClassName;

  private String            driverName;

  private String            url;

  private String            login;

  private String            password;

  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;

  private String            checkpointTableName;
  private String            checkpointTableCreateStatement;

  private String            backendTableCreateStatement;
  private String            backendTableName;

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

  private Connection        connection;

  private PreparedStatement pstmt;

  private long              logTableId = 0;

  private int               timeout;

  private JDBCLoggerThread  loggerThread;

  public JDBCRecoveryLog(String driverName, String driverClassName, String url,
      String login, String password, int requestTimeout)
  {
    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
  }

  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();
  }

  public long getLastTransactionId() throws SQLException
  {
    Statement stmt = connection.createStatement();
    ResultSet rs = stmt.executeQuery("select max(transaction_id) from "
        + logTableName);
    if (rs.next())
      return rs.getInt(1);
    else
      return 0;
  }

  public void setLogTableCreateStatement(String tableName, String idType,
      String vloginType, String sqlType, String transactionIdType,
      String extraStatement)
  {
    this.logTableName = tableName;
    this.idType = idType;
    this.vloginType = vloginType;
    this.sqlType = sqlType;
    this.transactionIdType = transactionIdType;
    this.logextraStatement = extraStatement;
    String logTableCreateStatement = "CREATE TABLE " + tableName + " (id "
        + idType + ",vlogin " + vloginType + ",sql " + sqlType
        + ",transaction_id " + transactionIdType + extraStatement + ")";
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("recovery.jdbc.logtable.statement",
          logTableCreateStatement));

    this.logTableCreateStatement = logTableCreateStatement;
  }

  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;
  }

  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));
  }

  private synchronized long incrementLogTableId()
  {
    logTableId++;
    return logTableId;
  }

  private void intializeDatabase() throws SQLException
  {
    boolean createLogTable = true;
    boolean createCheckpointTable = true;
    boolean createBackendTable = true;
    // Check if tables exist
    try
    {
      // 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;
        }
      }
    }
    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()}));
      }
    }
    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()}));
      }
    }

  }

  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);
    }
  }

  public void logRequest(AbstractWriteRequest request) throws SQLException
  {
    loggerThread.log(incrementLogTableId(), request.getLogin(), request
        .getSQL(), request.getTransactionId(), request.getEscapeProcessing());
  }

  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());
    }
  }

  public void begin(TransactionMarkerMetaData tm) throws SQLException
  {
    // Store the begin in the database
    loggerThread.log(incrementLogTableId(), tm.getLogin(), "begin", tm
        .getTransactionId(), false);
  }

  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);
  }

  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.rollbackTransaction(transactionId);
      PreparedStatement stmt = null;
      try
      {
        stmt = connection.prepareStatement("DELETE FROM " + logTableName
            + " WHERE transaction_id=?");
        stmt.setLong(1, transactionId);
        stmt.executeUpdate();
        stmt.close();
      }
      catch (SQLException e)
      {
        try
        {
          if (stmt != null)
            stmt.close();
        }
        catch (Exception ignore)
        {
        }
        throw new SQLException(Translate.get(
            "recovery.jdbc.transaction.remove.failed", new String[]{
                String.valueOf(transactionId), e.getMessage()}));
      }
    }
  }

  public void storeCheckpoint(String checkpointName) throws SQLException
  {
    storeCheckpoint(checkpointName, logTableId);
  }

  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.
      return !rs.next();
    }
    catch (SQLException e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.check.failed", e));
    }
    finally
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (SQLException ignore)
      {
      }
    }
  }

  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
    {
      stmt = connection.prepareStatement("INSERT INTO " + checkpointTableName
          + " VALUES(?,?)");
      stmt.setString(1, checkpointName);
      stmt.setLong(2, requestId);
      stmt.executeUpdate();
      stmt.close();
    }
    catch (SQLException e)
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.store.failed", new String[]{checkpointName,
              e.getMessage()}));
    }
  }

  private void waitForTransactionsEnd(boolean forceRollback)
  {
    if (forceRollback)
    {
      loggerThread.rollbackTransactions();
    }
    else
    {
      synchronized (loggerThread)
      {
        while (!loggerThread.getLogQueueIsEmpty())
        {
          try
          {
            wait(100);
          }
          catch (Exception e)
          {
            logger.warn("Exception " + e
                + " while waiting for end of transactions");
          }
        }
      }
    }
  }

  public long getCheckpointRequestId(String checkpointName) throws SQLException
  {
    long requestId = -1;
    PreparedStatement stmt = null;
    try
    {
      ResultSet rs = null;
      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");
      }
      stmt.close();
    }
    catch (SQLException e)
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.not.found", new String[]{checkpointName,
              e.getMessage()}));
    }
    return requestId;
  }

  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;

      do
      {
        // Take a window of 2 requests if there is a small hole
        // induced by a removed begin/commit/rollback command.
        stmt = connection.prepareStatement("SELECT * FROM " + logTableName
            + " WHERE id>? AND id<=? ORDER BY id");
        stmt.setLong(1, previousRequestId);
        stmt.setLong(2, previousRequestId + 2);
        rs = stmt.executeQuery();
        previousRequestId += 2; // Shift the window
        emptyResult = !rs.next();
      }
      while (emptyResult && (previousRequestId <= logTableId));

      // No more request after this one
      if (emptyResult)
        return null;

      String sql = rs.getString("sql");
      String user = rs.getString("vlogin");
      int transactionId = rs.getInt("transaction_id");
      int id = rs.getInt("id");
      stmt.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)
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
      throw new SQLException(Translate.get("recovery.jdbc.recover.failed", e));
    }
    return task;
  }

  private void setDriverProcessedAndSkeleton(AbstractRequest request)
  {
    String sql = request.getSQL();
    boolean isDriverProcessed = sql
        .indexOf(org.objectweb.cjdbc.driver.Connection.endParamTag) == -1;
    request.setDriverProcessed(isDriverProcessed);
    if (isDriverProcessed)
      return; // No need to set the skeleton

    request.setSqlSkeleton(recreateSkeleton(sql));
  }

  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.startParamTag, start)) != -1)
    {
      skeleton.append(sql.substring(start, end)).append('?');
      start = sql.indexOf(org.objectweb.cjdbc.driver.Connection.endParamTag,
          end);
      if (start == -1)
        throw new RuntimeException("Malformed query in recovery log: " + sql);
      else
        start += org.objectweb.cjdbc.driver.Connection.endParamTag.length();
    }
    if (start < sql.length())
      skeleton.append(sql.substring(start));
    return skeleton.toString();
  }

  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 p = null;
    long transactionId = -1;
    try
    {
      // remove the rollbacked transaction from the database
      while (rs.next())
      {
        transactionId = rs.getLong("transaction_Id");
        p = connection.prepareStatement("DELETE FROM " + logTableName
            + " WHERE transaction_id=?");
        p.setLong(1, transactionId);
        p.executeUpdate();
        p.close();
      }
      stmt.close();
    }
    catch (SQLException e)
    {
      try
      {
        if (stmt != null)
          stmt.close();
        if (p != null)
          p.close();
      }
      catch (Exception ignore)
      {
      }
      throw new SQLException(Translate.get(
          "recovery.jdbc.transaction.remove.failed", new String[]{
              String.valueOf(transactionId), e.getMessage()}));
    }
  }

  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();
      if (!rs.next())
        throw new SQLException("Checkpoint " + checkpointName
            + " does not exist");

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

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

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

  }

  public ArrayList getCheckpointNames() throws SQLException
  {
    PreparedStatement stmt = null;

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

    try
    {
      stmt = connection.prepareStatement("SELECT name from "
          + checkpointTableName);
      ResultSet rs = stmt.executeQuery();
      ArrayList list = new ArrayList();
      while (rs.next())
      {
        list.add(rs.getString("name"));
      }
      return list;
    }
    catch (Exception e)
    {
      throw new SQLException(Translate.get(
          "recovery.jdbc.checkpoint.list.failed", e));
    }

  }

  public void storeBackendInfo(String databaseName, String backendName,
      String checkpoint, int state) throws SQLException
  {
    PreparedStatement stmt = null;
    PreparedStatement stmt2 = null;
    checkpoint = (checkpoint == null) ? "" : checkpoint;

    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())
      {
        stmt2 = connection.prepareStatement("INSERT INTO " + backendTableName
            + " values('" + databaseName + "','" + backendName + "' ," + state
            + ",'" + checkpoint + "')");
        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=" + state + ", checkpointName='" + checkpoint
            + "' where backendName='" + backendName + "' and databaseName='"
            + databaseName + "'");
        if (stmt2.executeUpdate() != 1)
          throw new SQLException(
              "Error while updating backend reference. Incorrect number of rows");
      }

      stmt.close();
      stmt2.close();
    }
    catch (SQLException e)
    {
      e.printStackTrace();
      try
      {
        if (stmt != null)
          stmt.close();
        if (stmt2 != null)
          stmt2.close();
      }
      catch (Exception ignore)
      {
      }
      throw new SQLException("Unable to update checkpoint '" + checkpoint
          + "' for backend:" + backendName);
    }
  }

  public BackendRecoveryInfo getBackendInfo(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");
      }
      stmt.close();
    }
    catch (SQLException e)
    {
      try
      {
        if (stmt != null)
          stmt.close();
      }
      catch (Exception ignore)
      {
      }
    }
    return new BackendRecoveryInfo(backendName, checkpoint, backendState,
        databaseName);
  }

  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();
  }

  public String getBackendTableName()
  {
    return backendTableName;
  }

  public String getCheckpointTableName()
  {
    return checkpointTableName;
  }

  public String getLogTableName()
  {
    return logTableName;
  }

  public String[][] getData()
  {
    try
    {
      ResultSet rs = connection.createStatement().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;
    }
  }
}

---