Updated method that checks current schema version with consistency level ONE instead of ALL

src/main/java/io/smartcat/migration/CassandraVersioner.java

@@ -50,16 +50,17 @@ private void createSchemaVersion() {
     }
 
     /**
-     * Get current database version for given migration type with ALL consistency. Select one row since
-     * migration history is saved ordered descending by timestamp. If there are no rows in the schema_version table,
-     * return 0 as default database version. Data version is changed by executing migrations.
+     * Get current database version for given migration type with ALL consistency. Select one row since migration
+     * history is saved ordered descending by timestamp. If there are no rows in the schema_version table, return 0 as
+     * default database version. Data version is changed by executing migrations.
      *
      * @param type Migration type
+     * @param consistencyLevel Desired consistency level
      * @return Database version for given type
      */
-    public int getCurrentVersion(final MigrationType type) {
+    public int getCurrentVersion(final MigrationType type, final ConsistencyLevel consistencyLevel) {
         final Statement select = QueryBuilder.select().all().from(SCHEMA_VERSION_CF)
-                .where(QueryBuilder.eq(TYPE, type.name())).limit(1).setConsistencyLevel(ConsistencyLevel.ALL);
+                .where(QueryBuilder.eq(TYPE, type.name())).limit(1).setConsistencyLevel(consistencyLevel);
         final ResultSet result = session.execute(select);
 
         final Row row = result.one();

src/main/java/io/smartcat/migration/MigrationEngine.java

@@ -1,14 +1,16 @@
 package io.smartcat.migration;
 
-import io.smartcat.migration.exceptions.MigrationException;
+import static com.datastax.driver.core.ConsistencyLevel.ALL;
+import static com.datastax.driver.core.ConsistencyLevel.ONE;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.datastax.driver.core.Session;
 
+import io.smartcat.migration.exceptions.MigrationException;
 /**
  * Migration engine wraps Migrator and provides DSL like API.
- *
  */
 public class MigrationEngine {
 
@@ -30,6 +32,7 @@ public static Migrator withSession(final Session session) {
      * Migrator handles migrations and errors.
      */
     public static class Migrator {
+
         private final Session session;
         private final CassandraVersioner versioner;
 
@@ -55,7 +58,7 @@ public boolean migrate(final MigrationResources resources) {
             for (final Migration migration : resources.getMigrations()) {
                 final MigrationType type = migration.getType();
                 final int migrationVersion = migration.getVersion();
-                final int version = versioner.getCurrentVersion(type);
+                final int version = versioner.getCurrentVersion(type, ONE);
 
                 LOGGER.info("Db is version {} for type {}.", version, type.name());
                 LOGGER.info("Compare {} migration version {} with description {}", type.name(), migrationVersion,
@@ -67,6 +70,8 @@ public boolean migrate(final MigrationResources resources) {
                     continue;
                 }
 
+                checkIfAllReplicasAreUp(type);
+
                 migration.setSession(session);
 
                 final long start = System.currentTimeMillis();
@@ -94,6 +99,17 @@ public boolean migrate(final MigrationResources resources) {
 
             return true;
         }
+
+        /**
+         * Method that checks if all replicas are up  by getting current migration type with consistency
+         * level ALL. This method will only execute if there are new migrations. Method will throw exception if any of
+         * replicas is down and migration process will break
+         * 
+         * @param type Migration type
+         */
+        private void checkIfAllReplicasAreUp(final MigrationType type) {
+            versioner.getCurrentVersion(type, ALL);
+        }
     }
 
 }

src/test/java/io/smartcat/migration/CassandraVersionerTest.java

@@ -1,5 +1,6 @@
 package io.smartcat.migration;
 
+import static com.datastax.driver.core.ConsistencyLevel.ALL;
 import static io.smartcat.migration.MigrationType.SCHEMA;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertThat;
@@ -34,7 +35,7 @@ public void setUp() throws Exception {
     public void whenSchemaVersionTableIsEmptyThenCurrentVersionShouldBe0() throws Exception {
         expectRetrieveEmptyCurrentVersion();
 
-        int currentVersion = versioner.getCurrentVersion(SCHEMA);
+        int currentVersion = versioner.getCurrentVersion(SCHEMA, ALL);
 
         assertThat(currentVersion, is(0));
     }
@@ -45,7 +46,7 @@ public void whenSchemaVersionTableIsNotEmptyThenCurrentVersionShouldBeRetrievedF
 
         expectRetrieveCurrentVersion(expectedVersion);
 
-        int currentVersion = versioner.getCurrentVersion(SCHEMA);
+        int currentVersion = versioner.getCurrentVersion(SCHEMA, ALL);
 
         assertThat(currentVersion, is(expectedVersion));
     }

src/test/java/io/smartcat/migration/MigratorTest.java

@@ -1,5 +1,6 @@
 package io.smartcat.migration;
 
+import static com.datastax.driver.core.ConsistencyLevel.ALL;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.junit.Assert.assertFalse;
@@ -24,6 +25,7 @@
 import io.smartcat.migration.migrations.schema.AddBookISBNFieldMigration;
 
 public class MigratorTest extends BaseTest {
+
     private static final Logger LOGGER = LoggerFactory.getLogger(MigratorTest.class);
 
     private static final String CONTACT_POINT = "localhost";
@@ -138,7 +140,7 @@ public void skipMigrationWithSameVersionThanCurrentSchemaVersion() throws Except
     }
 
     private int getCurrentVersion() {
-        return versioner.getCurrentVersion(MigrationType.SCHEMA);
+        return versioner.getCurrentVersion(MigrationType.SCHEMA, ALL);
     }
 
     private void assertTableDoesntContainsColumns(String table, String... columns) {