/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2000-2009 Sun Microsystems, Inc. All rights reserved. 
 *
 * The contents of this file are subject to the terms of either the GNU
 * General Public License Version 2 only ("GPL") or the Common Development
 * and Distribution License ("CDDL") (collectively, the "License").  You may
 * not use this file except in compliance with the License.  You can obtain
 * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
 * or mq/legal/LICENSE.txt.  See the License for the specific language
 * governing permissions and limitations under the License.
 * 
 * When distributing the software, include this License Header Notice in each
 * file and include the License file at mq/legal/LICENSE.txt.  Sun designates
 * this particular file as subject to the "Classpath" exception as provided by
 * Sun in the GPL Version 2 section of the License file that accompanied this
 * code.  If applicable, add the following below the License Header, with the
 * fields enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 * 
 * Contributor(s):
 * 
 * If you wish your version of this file to be governed by only the CDDL or
 * only the GPL Version 2, indicate your decision by adding "[Contributor]
 * elects to include this software in this distribution under the [CDDL or GPL
 * Version 2] license."  If you don't indicate a single choice of license, a
 * recipient has the option to distribute your version of this file under
 * either the CDDL, the GPL Version 2 or  to extend the choice of license to
 * its licensees as provided above.  However, if you add GPL Version 2 code
 * and therefore, elected the GPL Version 2 license, then the option applies
 * only if the new code is made subject to such option by the copyright holder. 
 */

/*
 * @(#)BrokerDAOImpl.java	1.30 06/29/07
 */ 

package com.sun.messaging.jmq.jmsserver.persist.jdbc;

import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.jmsserver.util.*;
import com.sun.messaging.jmq.jmsserver.cluster.BrokerState;
import com.sun.messaging.jmq.jmsserver.persist.HABrokerInfo;
import com.sun.messaging.jmq.jmsserver.persist.TakeoverLockException;
import com.sun.messaging.jmq.jmsserver.resources.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.io.Status;

import java.util.*;
import java.sql.*;

/**
 * BrokerDAOImpl defines/implements the generic DAO API for the Broker table.
 */
class BrokerDAOImpl extends BaseDAOImpl implements BrokerDAO {

    public static final String STORE_SESSION_ID_COLUMN = "STORE_SESSION_ID";

    protected String tableName;

    // SQLs
    protected String insertSQL;
    protected String updateSQL;
    protected String updateHeartbeatSQL;
    protected String updateHeartbeatAndCheckStateSQL;
    protected String updateStateSQL;
    protected String takeoverSQL;
    protected String deleteSQL;
    protected String selectSQL;
    protected String selectAllSQL;
    protected String selectAllByStateSQL;
    protected String selectHeartbeatSQL;
    protected String selectAllHeartbeatsSQL;
    protected String selectStateSQL;
    protected String selectAllStatesSQL;
    protected String selectIsBeingTakenOverSQL;

    /**
     * Constructor
     * @throws BrokerException
     */
    BrokerDAOImpl() throws BrokerException {

        // Initialize all SQLs
        DBManager dbMgr = DBManager.getDBManager();

        tableName = dbMgr.getTableName( TABLE_NAME_PREFIX );

        insertSQL = new StringBuffer(128)
            .append( "INSERT INTO " ).append( tableName )
            .append( " ( " )
            .append( ID_COLUMN ).append( ", " )
            .append( URL_COLUMN ).append( ", " )
            .append( VERSION_COLUMN ).append( ", " )
            .append( STATE_COLUMN ).append( ", " )
            .append( TAKEOVER_BROKER_COLUMN ).append( ", " )
            .append( HEARTBEAT_TS_COLUMN )
            .append( ") VALUES ( ?, ?, ?, ?, ?, ? )" )
            .toString();

        updateSQL = new StringBuffer(128)
            .append( "UPDATE " ).append( tableName )
            .append( " SET " )
            .append( URL_COLUMN ).append( " = ?, " )
            .append( VERSION_COLUMN ).append( " = ?, " )
            .append( STATE_COLUMN ).append( " = ?, " )
            .append( TAKEOVER_BROKER_COLUMN ).append( " = ?" )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .toString();

        updateHeartbeatSQL = new StringBuffer(128)
            .append( "UPDATE " ).append( tableName )
            .append( " SET " )
            .append( HEARTBEAT_TS_COLUMN ).append( " = ?" )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .toString();

        updateHeartbeatAndCheckStateSQL = new StringBuffer( updateHeartbeatSQL )
            .append( " AND " )
            .append( HEARTBEAT_TS_COLUMN ).append( " = ?" )
            .append( " AND " )
            .append( STATE_COLUMN ).append( " NOT IN (" )
            .append( BrokerState.I_FAILOVER_PENDING ).append( ", " )
            .append( BrokerState.I_FAILOVER_STARTED ).append( ", " )
            .append( BrokerState.I_FAILOVER_COMPLETE ).append( ", " )
            .append( BrokerState.I_FAILOVER_FAILED ).append( ")" )
            .toString();

        updateStateSQL = new StringBuffer(128)
            .append( "UPDATE " ).append( tableName )
            .append( " SET " )
            .append( STATE_COLUMN ).append( " = ?" )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .append( " AND " )
            .append( STATE_COLUMN ).append( " = ?" )
            .toString();

        takeoverSQL = new StringBuffer(128)
            .append( "UPDATE " ).append( tableName )
            .append( " SET " )
            .append( TAKEOVER_BROKER_COLUMN ).append( " = ?, " )
            .append( STATE_COLUMN ).append( " = ?, " )
            .append( HEARTBEAT_TS_COLUMN ).append( " = ?" )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .append( " AND " )
            .append( STATE_COLUMN ).append( " = ?" )
            .append( " AND " )
            .append( HEARTBEAT_TS_COLUMN ).append( " = ?" )
            .append( " AND " )
            .append( TAKEOVER_BROKER_COLUMN ).append( " is NULL" )
            .toString();

        deleteSQL = new StringBuffer(128)
            .append( "DELETE FROM " ).append( tableName )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .toString();

        /*
         * All the supported store sessions for a broker is kept in the session
         * table but for convenience we will load the current store session
         * when load the broker info. If a broker doesn't have a current store
         * session, i.e. taken over by another broker, then its value will be 0.
         */

        selectAllSQL = new StringBuffer(128)
            .append( "SELECT bTbl." )
            .append( ID_COLUMN ).append( ", " )
            .append( URL_COLUMN ).append( ", " )
            .append( VERSION_COLUMN ).append( ", " )
            .append( STATE_COLUMN ).append( ", " )
            .append( TAKEOVER_BROKER_COLUMN ).append( ", " )
            .append( HEARTBEAT_TS_COLUMN ).append( ", sTbl." )
            .append( StoreSessionDAO.ID_COLUMN ).append( " AS " )
            .append( STORE_SESSION_ID_COLUMN )
            .append( " FROM " ).append( tableName )
            .append(   " bTbl LEFT JOIN " )
            .append(   dbMgr.getTableName( StoreSessionDAO.TABLE_NAME_PREFIX ) )
            .append(   " sTbl ON bTbl." ).append( ID_COLUMN ).append( " = sTbl." )
            .append(   StoreSessionDAO.BROKER_ID_COLUMN ).append( " AND sTbl." )
            .append(   StoreSessionDAO.IS_CURRENT_COLUMN ).append( " = 1" )
            .toString();

        selectAllByStateSQL = new StringBuffer(128)
            .append( selectAllSQL )
            .append( " WHERE bTbl." )
            .append( STATE_COLUMN ).append( " = ?" )
            .toString();

        selectSQL = new StringBuffer(128)
            .append( selectAllSQL )
            .append( " WHERE bTbl." )
            .append( ID_COLUMN ).append( " = ?" )
            .toString();

        selectHeartbeatSQL = new StringBuffer(128)
            .append( "SELECT " )
            .append( HEARTBEAT_TS_COLUMN )
            .append( " FROM " ).append( tableName )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .toString();

        selectAllHeartbeatsSQL = new StringBuffer(128)
            .append( "SELECT " )
            .append( ID_COLUMN ).append( ", " )
            .append( HEARTBEAT_TS_COLUMN )
            .append( " FROM " ).append( tableName )
            .toString();

        selectStateSQL = new StringBuffer(128)
            .append( "SELECT " )
            .append( STATE_COLUMN )
            .append( " FROM " ).append( tableName )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ?" )
            .toString();

        selectAllStatesSQL = new StringBuffer(128)
            .append( "SELECT " )
            .append( ID_COLUMN ).append( ", " )
            .append( STATE_COLUMN )
            .append( " FROM " ).append( tableName )
            .toString();

        // SQL that can be embedded in EXISTS clause to check if the specified
        // broker is being takenover (uses by other DAOs)
        selectIsBeingTakenOverSQL = new StringBuffer(128)
            .append( "SELECT 1 FROM " ).append( tableName )
            .append( " WHERE " )
            .append( ID_COLUMN ).append( " = ? AND " )
            .append( STATE_COLUMN ).append( " IN (" )
            .append( BrokerState.I_FAILOVER_PENDING ).append( ", " )
            .append( BrokerState.I_FAILOVER_STARTED ).append( ", " )
            .append( BrokerState.I_FAILOVER_COMPLETE ).append( ", " )
            .append( BrokerState.I_FAILOVER_FAILED ).append( ")" )
            .toString();
    }

    /**
     * Get the prefix name of the table.
     * @return table name
     */
    public final String getTableNamePrefix() {
        return TABLE_NAME_PREFIX;
    }

    /**
     * Get the name of the table.
     * @return table name
     */
    public final String getTableName() {
        return tableName;
    }

    /**
     * Insert a new entry.
     * @param conn database connection
     * @param id Broker ID
     * @param takeoverID Broker ID taken over the store
     * @param url the broker's URL
     * @param version the broker's version
     * @param state the broker's state
     * @param sessionID the broker's session ID
     * @param heartbeat the broker's heartbeat timestamp
     * @throws BrokerException
     */
    public void insert( Connection conn, String id, String takeoverID, String url,
        int version, int state, long sessionID, long heartbeat )
        throws BrokerException {

        boolean myConn = false;
        PreparedStatement pstmt = null;
        try {
            // Get a connection
            DBManager dbMgr = DBManager.getDBManager();
            if ( conn == null ) {
                conn = dbMgr.getConnection( false );
                myConn = true;
            }

            // First insert broker info
            pstmt = conn.prepareStatement( insertSQL );
            pstmt.setString( 1, id );
            pstmt.setString( 2, url );
            pstmt.setInt( 3, version );
            pstmt.setInt( 4, state );
            Util.setString( pstmt, 5, takeoverID );
            Util.setLong( pstmt, 6, heartbeat );

            pstmt.executeUpdate();
            pstmt.close();

            // Now insert store session
            if ( sessionID > 0 ) {
                dbMgr.getDAOFactory().getStoreSessionDAO().insert( conn, id, sessionID );
            }

            if ( myConn ) {
                conn.commit();
            }
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + insertSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_PERSIST_BROKERINFO_FAILED,
                    id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( null, pstmt, conn );
            } else {
                Util.close( null, pstmt, null );
            }
        }
    }

    /**
     * Update an existing entry.
     * @param conn database connection
     * @param id Broker ID
     * @param takeoverID Broker ID taken over the store
     * @param url the broker's URL
     * @param version the broker's version
     * @param state the broker's state
     * @param sessionID the broker's session ID
     * @throws BrokerException
     */
    public void update( Connection conn, String id, String takeoverID,
        String url, int version, int state, long sessionID )
        throws BrokerException {

        boolean myConn = false;
        PreparedStatement pstmt = null;
        try {
            // Get a connection
            DBManager dbMgr = DBManager.getDBManager();
            if ( conn == null ) {
                conn = dbMgr.getConnection( false );
                myConn = true;
            }

            pstmt = conn.prepareStatement( updateSQL );
            pstmt.setString( 1, url );
            pstmt.setInt( 2, version );
            pstmt.setInt( 3, state );
            Util.setString( pstmt, 4, takeoverID );
            pstmt.setString( 5, id );
            pstmt.executeUpdate();
            pstmt.close();

            // Will insert a new entry and mark as current ONLY IF
            // it does not exists!
            dbMgr.getDAOFactory().getStoreSessionDAO().insert( conn, id, sessionID );

            if ( myConn ) {
                conn.commit();
            }
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + updateSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_PERSIST_BROKERINFO_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( null, pstmt, conn );
            } else {
                Util.close( null, pstmt, null );
            }
        }
    }

    /**
     * Update the broker heartbeat timestamp to the current time.
     * @param conn database connection
     * @param id Broker ID
     * @param heartbeat timestamp
     * @return true if heartbeat timestamp is successfully updated.
     * @throws BrokerException
     */
    public boolean updateHeartbeat( Connection conn, String id, long heartbeat )
        throws BrokerException {

        boolean updated = false;
        boolean myConn = false;
        PreparedStatement pstmt = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( updateHeartbeatSQL );
            pstmt.setLong( 1, heartbeat );
            pstmt.setString( 2, id );
            if ( pstmt.executeUpdate() == 1 ) {
                updated = true;
            }
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + updateHeartbeatSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_UPDATE_HEARTBEAT_TS_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( null, pstmt, conn );
            } else {
                Util.close( null, pstmt, null );
            }
        }

        return updated;
    }

    /**
     * Update the broker heartbeat timestamp only if the specified lastHeartbeat
     * value match the value store in the DB.
     * @param conn database connection
     * @param id Broker ID
     * @param heartbeat timestamp
     * @param lastHeartbeat broker's last heartbeat
     * @return true if heartbeat timestamp is successfully updated.
     * @throws BrokerException
     */
    public boolean updateHeartbeat( Connection conn, String id, long heartbeat,
        long lastHeartbeat ) throws BrokerException {

        boolean updated = false;
        boolean myConn = false;
        PreparedStatement pstmt = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( updateHeartbeatAndCheckStateSQL );
            pstmt.setLong( 1, heartbeat );
            pstmt.setString( 2, id );
            pstmt.setLong( 3, lastHeartbeat );
            if ( pstmt.executeUpdate() == 0 ) {
                HABrokerInfo bkrInfo = getBrokerInfo( conn, id );
                if ( bkrInfo == null ) {
                    String errorMsg = br.getKString(
                        BrokerResources.E_BROKERINFO_NOT_FOUND_IN_STORE, id );
                    throw new BrokerException( br.getKString(
                        BrokerResources.X_UPDATE_HEARTBEAT_TS_2_FAILED, id, errorMsg ) );
                }

                // Verify if persistent store is being taken over
                String takeoverBroker = bkrInfo.getTakeoverBrokerID();
                int state = bkrInfo.getState();
                if ( takeoverBroker != null && takeoverBroker.length() > 0 &&
                     ( state == BrokerState.I_FAILOVER_PENDING ||
                       state == BrokerState.I_FAILOVER_STARTED ||
                       state == BrokerState.I_FAILOVER_COMPLETE ||
                       state == BrokerState.I_FAILOVER_FAILED ) ) {
                    BrokerException be = new StoreBeingTakenOverException(
                        br.getKString( BrokerResources.E_STORE_BEING_TAKEN_OVER ) );
                    logger.log( Logger.ERROR, br.getKString(
                        BrokerResources.X_UPDATE_HEARTBEAT_TS_2_FAILED, id,
                        bkrInfo.toString() ), be );
                    throw be;
                }
            } else {
                updated = true;
            }
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + updateHeartbeatAndCheckStateSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            String arg = "Expected last heartbeat " + lastHeartbeat;
            throw new BrokerException(
                br.getKString( BrokerResources.X_UPDATE_HEARTBEAT_TS_2_FAILED, id, arg ), ex );
        } finally {
            if ( myConn ) {
                Util.close( null, pstmt, conn );
            } else {
                Util.close( null, pstmt, null );
            }
        }

        return updated;
    }

    /**
     * Update the state of a broker only if the current state matches the
     * expected state.
     * @param conn database connection
     * @param id Broker ID
     * @param newState the new state
     * @param expectedState the expected state
     * @return true if state is successfully updated.
     * @throws BrokerException
     */
    public boolean updateState( Connection conn, String id,
        BrokerState newState, BrokerState expectedState) throws BrokerException {

        boolean updated = false;
        boolean myConn = false;
        PreparedStatement pstmt = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( updateStateSQL );
            pstmt.setInt( 1, newState.intValue() );
            pstmt.setString( 2, id );
            pstmt.setInt( 3, expectedState.intValue() );
            if ( pstmt.executeUpdate() == 1 ) {
                updated = true;
            }
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + updateStateSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_PERSIST_BROKERINFO_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( null, pstmt, conn );
            } else {
                Util.close( null, pstmt, null );
            }
        }

        return updated;
    }

    /**
     * Update the state and other relevant attributes of a broker to signify
     * the store is being taken over by another broker. If the operation is
     * successful then this means that the current broker was able to acquire
     * the lock and it is now responsible for taken over the store of the
     * target broker.
     * @param conn database connection
     * @param id the current or local broker ID
     * @param targetBrokerID the broker ID of the store being taken over
     * @param lastHeartbeat broker's last heartbeat
     * @param expectedState the expected state
     * @param newHeartbeat the new timestamp
     * @param newState the new state
     * @throws TakeoverLockException if the current broker is unable to acquire
     *      the takeover lock
     * @throws BrokerException
     * @return previous broker's info associated with the broker
     */
    public HABrokerInfo takeover( Connection conn, String id,
        String targetBrokerID, long lastHeartbeat, BrokerState expectedState,
        long newHeartbeat, BrokerState newState) throws BrokerException {

        HABrokerInfo bkrInfo = null;

        PreparedStatement pstmt = null;
        try {
            // Save the broker's state before updating
            bkrInfo = getBrokerInfo( conn, targetBrokerID );
            if ( bkrInfo == null ) {
                String errorMsg = br.getKString(
                    BrokerResources.E_BROKERINFO_NOT_FOUND_IN_STORE, targetBrokerID );
                throw new BrokerException(
                    br.getKString( BrokerResources.E_INTERNAL_BROKER_ERROR, errorMsg ) );
            }

            pstmt = conn.prepareStatement( takeoverSQL );
            pstmt.setString( 1, id );
            pstmt.setInt( 2, newState.intValue() );
            pstmt.setLong( 3, newHeartbeat );
            pstmt.setString( 4, targetBrokerID );
            pstmt.setInt( 5, expectedState.intValue() );
            pstmt.setLong( 6, lastHeartbeat );

            if ( pstmt.executeUpdate() != 1 ) {
                String errorMsg = br.getKString(
                    BrokerResources.E_UNABLE_TO_ACQUIRE_TAKEOVER_LOCK, targetBrokerID );
                TakeoverLockException ex = new TakeoverLockException( errorMsg );
                ex.setBrokerInfo( bkrInfo ); // Store broker info
                throw ex;
            }
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + takeoverSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.E_UNABLE_TO_TAKEOVER_BROKER, targetBrokerID ), ex );
        } finally {
            Util.close( null, pstmt, null );
        }

        return bkrInfo;
    }

    /**
     * Delete an entry.
     * @param conn database connection
     * @param id Broker ID
     * @throws BrokerException
     */
    public void delete( Connection conn, String id )
        throws BrokerException {

        boolean myConn = false;
        PreparedStatement pstmt = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( deleteSQL );
            pstmt.setString( 1, id );
            pstmt.executeUpdate();
        } catch ( Exception e ) {
            try {
                if ( (conn != null) && !conn.getAutoCommit() ) {
                    conn.rollback();
                }
            } catch ( SQLException rbe ) {
                logger.log( Logger.ERROR, BrokerResources.X_DB_ROLLBACK_FAILED, rbe );
            }

            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + deleteSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_REMOVE_BROKERINFO_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( null, pstmt, conn );
            } else {
                Util.close( null, pstmt, null );
            }
        }
    }

    /**
     * Delete all entries.
     * @param conn database connection
     * @throws BrokerException
     */
    public void deleteAll( Connection conn )
        throws BrokerException {

        if ( Globals.getHAEnabled() ) {
            return; // // Broker table cannot be reset
        } else {
            super.deleteAll( conn );
        }
    }

    /**
     * Get the heartbeat timestamp for the specified brokerID.
     * @param conn database connection
     * @param id Broker ID
     * @return heartbeat timestamp
     * @throws BrokerException
     */
    public long getHeartbeat( Connection conn, String id )
        throws BrokerException {

        long heartBeat = -1;

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectHeartbeatSQL );
            pstmt.setString( 1, id );
            rs = pstmt.executeQuery();
            if ( rs.next() ) {
                heartBeat = rs.getLong( 1 );
            }
        } catch ( Exception e ) {
            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectHeartbeatSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_BROKERINFO_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return heartBeat;
    }

    /**
     * Get the heartbeat timestamp for all brokers in an HA cluster.
     * @param conn database connection
     * @return a HashMap object where the key is the broker ID and the entry
     * value is the broker's heartbeat timestamps
     * @throws BrokerException
     */
    public HashMap getAllHeartbeats( Connection conn )
        throws BrokerException {

        HashMap data = new HashMap();

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectAllHeartbeatsSQL );
            rs = pstmt.executeQuery();
            while ( rs.next() ) {
                String id = rs.getString( 1 );
                long timestamp = rs.getLong( 2 );
                data.put( id, new Long( timestamp ) );
            }
        } catch ( Exception e ) {
            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectAllHeartbeatsSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_ALL_BROKERINFO_FAILED ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return data;
    }

    /**
     * Get the state for the specified brokerID.
     * @param conn database connection
     * @param id Broker ID
     * @return state of the broker
     * @throws BrokerException
     */
    public BrokerState getState( Connection conn, String id )
        throws BrokerException {

        BrokerState state = null;

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectStateSQL );
            pstmt.setString( 1, id );
            rs = pstmt.executeQuery();
            if ( rs.next() ) {
                state = BrokerState.getState( rs.getInt( 1 ) );
            } else {
                throw new BrokerException(
                    br.getKString( BrokerResources.E_BROKERINFO_NOT_FOUND_IN_STORE, id ),
                    Status.NOT_FOUND );
            }
        } catch ( Exception e ) {
            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectStateSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_BROKERINFO_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return state;
    }

    /**
     * Get the state for all brokers in an HA cluster.
     * @param conn database connection
     * @return an array of Object whose 1st element contains an ArrayList
     * of broker IDs and the 2nd element contains an ArrayList of BrokerState
     * @throws BrokerException
     */
    public Object[] getAllStates( Connection conn )
        throws BrokerException {

        ArrayList ids = new ArrayList();
        ArrayList states = new ArrayList();
        Object[] data = new Object[2];
        data[0] = ids;
        data[1] = states;

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            if ( conn == null ) {
                conn = DBManager.getDBManager().getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectAllStatesSQL );
            rs = pstmt.executeQuery();
            while ( rs.next() ) {
                String id = rs.getString( 1 );
                int state = rs.getInt( 2 );
                ids.add( id );
                states.add( BrokerState.getState( state ) );
            }
        } catch ( Exception e ) {
            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectAllStatesSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_ALL_BROKERINFO_FAILED ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return data;
    }

    /**
     * Get broker information.
     * @param conn database connection
     * @param id the broker ID.
     * @return a HABrokerInfo object
     * @throws BrokerException
     */
    public HABrokerInfo getBrokerInfo( Connection conn, String id )
        throws BrokerException {

        HABrokerInfo bkrInfo = null;

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            DBManager dbMgr = DBManager.getDBManager();
            if ( conn == null ) {
                conn = dbMgr.getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectSQL );
            pstmt.setString( 1, id );
            rs = pstmt.executeQuery();
            if ( rs.next() ) {
                bkrInfo = loadData( rs );
            }
        } catch ( Exception e ) {
            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_BROKERINFO_FAILED, id ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return bkrInfo;
    }

    /**
     * Get broker information for all brokers.
     * @param conn database connection
     * @param loadSession specify if store sessions should be loaded
     * @return a HashMap object containing HABrokerInfo for all brokers
     * @throws BrokerException
     */
    public HashMap getAllBrokerInfos( Connection conn, boolean loadSession )
        throws BrokerException {

        HashMap data = new HashMap();

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            DBManager dbMgr = DBManager.getDBManager();
            if ( conn == null ) {
                conn = dbMgr.getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectAllSQL );
            rs = pstmt.executeQuery();
            while ( rs.next() ) {
                HABrokerInfo bkrInfo = loadData( rs );
                data.put( bkrInfo.getId(), bkrInfo );
            }

            if ( loadSession ) {
                rs.close();
                pstmt.close();
                
                Map sessionMap =
                    dbMgr.getDAOFactory().getStoreSessionDAO().getAllStoreSessions( conn );
                Iterator itr = sessionMap.entrySet().iterator();
                while ( itr.hasNext() ) {
                    Map.Entry entry = (Map.Entry)itr.next();
                    String brokerID = (String)entry.getKey();
                    HABrokerInfo bkrInfo = (HABrokerInfo)data.get( brokerID );
                    if ( bkrInfo != null ) {
                        bkrInfo.setSessionList( (List)entry.getValue() );
                    }
                }
            }
        } catch ( Exception e ) {
        	Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectAllSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_ALL_BROKERINFO_FAILED ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return data;
    }

    /**
     * Get broker information for all brokers in an HA cluster by state.
     * @param conn database connection
     * @param state the state of the broker
     * @return a HashMap object containing HABrokerInfo for all brokers
     * @throws BrokerException
     */
    public HashMap getAllBrokerInfosByState( Connection conn, BrokerState state )
        throws BrokerException {

        HashMap data = new HashMap();

        boolean myConn = false;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            // Get a connection
            DBManager dbMgr = DBManager.getDBManager();
            if ( conn == null ) {
                conn = dbMgr.getConnection( true );
                myConn = true;
            }

            pstmt = conn.prepareStatement( selectAllByStateSQL );
            pstmt.setInt( 1, state.intValue() );
            rs = pstmt.executeQuery();
            while ( rs.next() ) {
                HABrokerInfo bkrInfo = loadData( rs );
                data.put( bkrInfo.getId(), bkrInfo );
            }
        } catch ( Exception e ) {
            Exception ex;
            if ( e instanceof BrokerException ) {
                throw (BrokerException)e;
            } else if ( e instanceof SQLException ) {
                ex = DBManager.wrapSQLException("[" + selectAllByStateSQL + "]", (SQLException)e);
            } else {
                ex = e;
            }

            throw new BrokerException(
                br.getKString( BrokerResources.X_LOAD_ALL_BROKERINFO_FAILED ), ex );
        } finally {
            if ( myConn ) {
                Util.close( rs, pstmt, conn );
            } else {
                Util.close( rs, pstmt, null );
            }
        }

        return data;
    }

    /**
     * Get debug information about the store.
     * @param conn database connection
     * @return a HashMap of name value pair of information
     */
    public HashMap getDebugInfo( Connection conn ) {

        HashMap map = new HashMap();
        StringBuffer strBuf = new StringBuffer( 512 );

        try {
            // Get info for all brokers in the cluster
            Collection data = getAllBrokerInfos( conn, false ).values();
            Iterator itr = data.iterator();
            while ( itr.hasNext() ) {
                Object obj = itr.next();
                strBuf.append( obj.toString() ).append( BrokerResources.NL );
            }

        } catch ( Exception e ) {}

        map.put( "Broker(" + tableName + ")", strBuf.toString() );
        return map;
    }

    /**
     * Check whether the specified broker is being taken over by another broker.
     * @param conn database connection
     * @param id Broker ID
     * @return true if the specified broker is being taken over
     * @throws BrokerException
     */
    public boolean isBeingTakenOver( Connection conn, String id )
        throws BrokerException {

        BrokerState brokerState = getState( conn, id );
        int state = brokerState.intValue();
        return ( state == BrokerState.I_FAILOVER_PENDING ||
                 state == BrokerState.I_FAILOVER_STARTED ||
                 state == BrokerState.I_FAILOVER_COMPLETE ||
                 state == BrokerState.I_FAILOVER_FAILED );
    }

    /**
     * Load the broker info to a value object.
     */
    protected HABrokerInfo loadData( ResultSet rs )
        throws SQLException {

        HABrokerInfo brokerInfo = new HABrokerInfo(
            rs.getString( ID_COLUMN ),
            rs.getString( TAKEOVER_BROKER_COLUMN ),
            rs.getString( URL_COLUMN ),
            rs.getInt( VERSION_COLUMN ),
            rs.getInt( STATE_COLUMN ),
            rs.getLong( STORE_SESSION_ID_COLUMN ),
            rs.getLong( HEARTBEAT_TS_COLUMN )
        );

        return brokerInfo;
    }
}
