DBLockManager

DbLockManager.java

package org.talend.example;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DbLockManager {
    private static Logger LOG = LoggerFactory.getLogger(DbLockManager.class);
    private DataSource dataSource;
    private int sleepTime = 1000;
    private String lockTableName = "lockTable";

    private boolean active;

    private Thread thread;
    private boolean shouldStop;
    private boolean reconnect;
    private String lastExceptionMessage;

    public void start() {
        shouldStop = false;
        this.thread = new Thread(new Runnable() {

            @Override
            public void run() {
                manageLock();
            }
        }, "LockManager");
        thread.start();
    }

    public void stop() {
        shouldStop = true;
    }

    /**
     * Keep Try to get or refresh the lock on the table
     * If an exception happens or we shut down we exit from the loop
     * 
     * Basically we would not need to exit the loop if only the transaction
     * times out while getting the lock. As this case is difficult to
     * detect with all different drivers we also exit in that case.
     * 
     * @param stmt
     * @throws SQLException When either the transaction times out getting the log
     *         or another problem happens 
     * @throws InterruptedException when the sleep is interrupted
     */
    private void manageLockUsingStatement(Statement stmt) throws SQLException, InterruptedException {
        do {
            this.active = false;
            stmt.execute(String.format("select * from %s for update", lockTableName));
            this.active = true;
            Thread.sleep(sleepTime);
        } while (!shouldStop);
    }
    
    /**
     * Check and log the exception and determine if a reconnect is needed
     * 
     * Override this if your database returns something else on transaction timeout
     * 
     * @param t Throwable that was thrown
     * @return true if reconnect is needed
     */
    protected boolean handleException(Throwable t) {
        if (t.getMessage().startsWith("Lock wait timeout exceeded")) {
            lastExceptionMessage = null;
            LOG.debug("Failed to acquire lock. " + t.getMessage());
            return false;
        } else {
            if (lastExceptionMessage != null && lastExceptionMessage.equals(t.getMessage())) {
                LOG.debug(t.getMessage(), t);
            } else {
                LOG.error(t.getMessage(), t);
                lastExceptionMessage = t.getMessage();
            }
            return true;
        }
    }

    /**
     * Create a connection and statement and try to get the lock
     * 
     * In case of an error we reconnect so we can also handle if the server is down
     */
    private void manageLock() {
        do {
            Connection con = null;
            Statement stmt = null;
            try {
                if (con == null) {
                    con = dataSource.getConnection();
                    con.setAutoCommit(false);
                }
                stmt = con.createStatement();
                manageLockUsingStatement(stmt);
            } catch (Throwable t) {
                reconnect = handleException(t);
            } finally {
                try {
                    Thread.sleep(sleepTime);
                } catch (InterruptedException e) {
                    return;
                }
                if (stmt != null) {
                    try {
                        stmt.close();
                    } catch (SQLException e1) {
                    }
                }
                if (con != null && this.reconnect) {
                    try {
                        con.close();
                    } catch (SQLException e1) {
                    } finally {
                        con = null;
                        this.reconnect = false;
                    }
                }
            }
        } while (!shouldStop);

    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    public void setSleepTime(int sleepTime) {
        this.sleepTime = sleepTime;
    }

    public void setLockTableName(String lockTableName) {
        this.lockTableName = lockTableName;
    }

    public boolean isActive() {
        return active;
    }
}