diff --git a/src/main/java/io/smartcat/migration/CassandraVersioner.java b/src/main/java/io/smartcat/migration/CassandraVersioner.java index 9806f77..fd1ad4e 100644 --- a/src/main/java/io/smartcat/migration/CassandraVersioner.java +++ b/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(); diff --git a/src/main/java/io/smartcat/migration/MigrationEngine.java b/src/main/java/io/smartcat/migration/MigrationEngine.java index 57c7340..91c558b 100644 --- a/src/main/java/io/smartcat/migration/MigrationEngine.java +++ b/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); + } } } diff --git a/src/test/java/io/smartcat/migration/CassandraVersionerTest.java b/src/test/java/io/smartcat/migration/CassandraVersionerTest.java index ca8ee39..3c10baf 100644 --- a/src/test/java/io/smartcat/migration/CassandraVersionerTest.java +++ b/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)); } diff --git a/src/test/java/io/smartcat/migration/MigratorTest.java b/src/test/java/io/smartcat/migration/MigratorTest.java index 2eed788..78cebce 100644 --- a/src/test/java/io/smartcat/migration/MigratorTest.java +++ b/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) {