// Path: src/main/java/uk/sky/cqlmigrate/exception/CannotAcquireLockException.java
// public class CannotAcquireLockException extends LockException {
// 
//     public CannotAcquireLockException(String message) {
//         super(message);
//     }
// 
//     public CannotAcquireLockException(String message, Throwable cause) {
//         super(message, cause);
//     }
// }
// 
// Path: src/main/java/uk/sky/cqlmigrate/exception/CannotReleaseLockException.java
// public class CannotReleaseLockException extends LockException {
// 
//     public CannotReleaseLockException(String message, Throwable cause) {
//         super(message, cause);
//     }
// 
//     public CannotReleaseLockException(String message) {
//         super(message);
//     }
// }


import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.servererrors.WriteTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import uk.sky.cqlmigrate.exception.CannotAcquireLockException;
import uk.sky.cqlmigrate.exception.CannotReleaseLockException;
import static java.lang.String.format;
package uk.sky.cqlmigrate;



class CassandraLockingMechanism extends LockingMechanism {

    private static final Logger log = LoggerFactory.getLogger(CassandraLockingMechanism.class);

    private final CqlSession session;
    private final ConsistencyLevel consistencyLevel;
    private final String lockKeyspace;

    private PreparedStatement selectLockQuery;
    private PreparedStatement insertLockQuery;
    private PreparedStatement deleteLockQuery;
    private boolean isRetryAfterWriteTimeout;

    public CassandraLockingMechanism(CqlSession session, String keyspace, ConsistencyLevel consistencyLevel, String lockKeyspace) {
        super(keyspace + ".schema_migration");
        this.session = session;
        this.consistencyLevel = consistencyLevel;
        this.lockKeyspace = lockKeyspace;
    }

    /**
     * Prepares queries for acquiring and releasing lock.
     *
     * @throws CannotAcquireLockException if any DriverException thrown while executing queries.
     */
    @Override
    public void init() throws CannotAcquireLockException {
        super.init();

        try {
            String selectQuery = format("SELECT name,client FROM %s.locks LIMIT 1", lockKeyspace);
            String insertQuery = format("INSERT INTO %s.locks (name, client) VALUES (?, ?) IF NOT EXISTS", lockKeyspace);
            String deleteQuery = format("DELETE FROM %s.locks WHERE name = ? IF client = ?", lockKeyspace);

            selectLockQuery = session.prepare(SimpleStatement.newInstance(selectQuery).setConsistencyLevel(consistencyLevel));
            insertLockQuery = session.prepare(SimpleStatement.newInstance(insertQuery).setConsistencyLevel(consistencyLevel));
            deleteLockQuery = session.prepare(SimpleStatement.newInstance(deleteQuery).setConsistencyLevel(consistencyLevel));
        } catch (DriverException e) {
// Path: src/main/java/uk/sky/cqlmigrate/exception/CannotAcquireLockException.java
// public class CannotAcquireLockException extends LockException {
// 
//     public CannotAcquireLockException(String message) {
//         super(message);
//     }
// 
//     public CannotAcquireLockException(String message, Throwable cause) {
//         super(message, cause);
//     }
// }
// 
// Path: src/main/java/uk/sky/cqlmigrate/exception/CannotReleaseLockException.java
// public class CannotReleaseLockException extends LockException {
// 
//     public CannotReleaseLockException(String message, Throwable cause) {
//         super(message, cause);
//     }
// 
//     public CannotReleaseLockException(String message) {
//         super(message);
//     }
// }


// Path: src/main/java/uk/sky/cqlmigrate/CassandraLockingMechanism.java
import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.DriverException;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.servererrors.WriteTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import uk.sky.cqlmigrate.exception.CannotAcquireLockException;
import uk.sky.cqlmigrate.exception.CannotReleaseLockException;
import static java.lang.String.format;

package uk.sky.cqlmigrate;



class CassandraLockingMechanism extends LockingMechanism {

    private static final Logger log = LoggerFactory.getLogger(CassandraLockingMechanism.class);

    private final CqlSession session;
    private final ConsistencyLevel consistencyLevel;
    private final String lockKeyspace;

    private PreparedStatement selectLockQuery;
    private PreparedStatement insertLockQuery;
    private PreparedStatement deleteLockQuery;
    private boolean isRetryAfterWriteTimeout;

    public CassandraLockingMechanism(CqlSession session, String keyspace, ConsistencyLevel consistencyLevel, String lockKeyspace) {
        super(keyspace + ".schema_migration");
        this.session = session;
        this.consistencyLevel = consistencyLevel;
        this.lockKeyspace = lockKeyspace;
    }

    /**
     * Prepares queries for acquiring and releasing lock.
     *
     * @throws CannotAcquireLockException if any DriverException thrown while executing queries.
     */
    @Override
    public void init() throws CannotAcquireLockException {
        super.init();

        try {
            String selectQuery = format("SELECT name,client FROM %s.locks LIMIT 1", lockKeyspace);
            String insertQuery = format("INSERT INTO %s.locks (name, client) VALUES (?, ?) IF NOT EXISTS", lockKeyspace);
            String deleteQuery = format("DELETE FROM %s.locks WHERE name = ? IF client = ?", lockKeyspace);

            selectLockQuery = session.prepare(SimpleStatement.newInstance(selectQuery).setConsistencyLevel(consistencyLevel));
            insertLockQuery = session.prepare(SimpleStatement.newInstance(insertQuery).setConsistencyLevel(consistencyLevel));
            deleteLockQuery = session.prepare(SimpleStatement.newInstance(deleteQuery).setConsistencyLevel(consistencyLevel));
        } catch (DriverException e) {
            throw new CannotAcquireLockException("Query to prepare locks queries failed", e);