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

Author: Jovica Andric

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

@@ -24,18 +24,16 @@ public class CassandraVersioner {
     private static final String DESCRIPTION = "description";
 
     private static final String CREATE_SCHEMA_VERSION_CQL = String.format("CREATE TABLE IF NOT EXISTS %s (",
-            SCHEMA_VERSION_CF)
-            + String.format("%s text,", TYPE)
-            + String.format("%s int,", VERSION)
-            + String.format("%s bigint,", TIMESTAMP)
-            + String.format("%s text,", DESCRIPTION)
+            SCHEMA_VERSION_CF) + String.format("%s text,", TYPE) + String.format("%s int,", VERSION)
+            + String.format("%s bigint,", TIMESTAMP) + String.format("%s text,", DESCRIPTION)
             + String.format("PRIMARY KEY (%s, %s)", TYPE, VERSION)
             + String.format(")  WITH CLUSTERING ORDER BY (%s DESC)", VERSION) + " AND COMMENT='Schema version';";
 
     private final Session session;
 
     /**
      * Create Cassandra versioner for active session.
+     * 
      * @param session Active Cassandra session
      */
     public CassandraVersioner(final Session session) {
@@ -50,16 +48,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,17 @@
 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 {
 
@@ -19,6 +22,7 @@ private MigrationEngine() {
 
     /**
      * Create migrator out of session fully prepared for doing migration of resources.
+     *
      * @param session Datastax driver session object
      * @return migrator instance with versioner and session which can migrate resources
      */
@@ -30,11 +34,13 @@ public static Migrator withSession(final Session session) {
      * Migrator handles migrations and errors.
      */
     public static class Migrator {
+
         private final Session session;
         private final CassandraVersioner versioner;
 
         /**
          * Create new Migrator with active Cassandra session.
+         *
          * @param session Active Cassandra session
          */
         public Migrator(final Session session) {
@@ -55,7 +61,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 +73,8 @@ public boolean migrate(final MigrationResources resources) {
                     continue;
                 }
 
+                checkIfAllReplicasAreUp(type);
+
                 migration.setSession(session);
 
                 final long start = System.currentTimeMillis();
@@ -94,6 +102,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) {