Add support for setting the serial consistency level for leader election

.github/workflows/maven.yml

@@ -0,0 +1,17 @@
+name: Maven CI
+
+on: push
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-java@v3
+        with:
+          java-version: '8'
+          distribution: 'temurin'
+          cache: maven
+      - name: Run Maven tests
+        run: mvn --batch-mode test
+

cassandra-migration-spring-boot-starter/pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <groupId>org.cognitor.cassandra</groupId>
         <artifactId>cassandra-migration-parent</artifactId>
-        <version>2.6.2_v4-SNAPSHOT</version>
+        <version>2.6.3_v4-SNAPSHOT</version>
     </parent>
 
     <artifactId>cassandra-migration-spring-boot-starter</artifactId>
-    <version>2.6.2_v4-SNAPSHOT</version>
+    <version>2.6.3_v4-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>Cassandra Migration Spring Boot Integration</name>
     <url>https://github.com/patka/cassandra-migration</url>

cassandra-migration/pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <groupId>org.cognitor.cassandra</groupId>
         <artifactId>cassandra-migration-parent</artifactId>
-        <version>2.6.2_v4-SNAPSHOT</version>
+        <version>2.6.3_v4-SNAPSHOT</version>
     </parent>
 
     <artifactId>cassandra-migration</artifactId>
-    <version>2.6.2_v4-SNAPSHOT</version>
+    <version>2.6.3_v4-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>Cassandra Migration</name>
     <url>https://github.com/patka/cassandra-migration</url>

cassandra-migration/src/main/java/org/cognitor/cassandra/migration/Database.java

@@ -98,7 +98,7 @@ public class Database implements Closeable {
     private final String keyspaceName;
     private final Keyspace keyspace;
     private final CqlSession session;
-    private final ConsistencyLevel consistencyLevel = ConsistencyLevel.QUORUM;
+    private ConsistencyLevel consistencyLevel = ConsistencyLevel.SERIAL;
     private final PreparedStatement logMigrationStatement;
     private final PreparedStatement takeMigrationLeadStatement;
     private final PreparedStatement releaseMigrationLeadStatement;
@@ -286,7 +286,8 @@ private void createSchemaTables() {
 
     /**
      * Attempts to acquire the lead on a migration through a LightWeight
-     * Transaction. For this statement the consistency level of <code>QUORUM</code> is used.
+     * Transaction. For this statement the consistency level of <code>SERIAL</code> is used by default,
+     * see {@link Database#setLeaderConsistencyLevel(ConsistencyLevel)} for configuring this.
      *
      * @param repositoryLatestVersion the latest version number in the migration repository
      * @return if taking the lead succeeded.
@@ -296,8 +297,8 @@ boolean takeLeadOnMigrations(int repositoryLatestVersion) {
             try {
                 LOGGER.debug("Trying to take lead on schema migrations");
                 BoundStatement boundStatement = takeMigrationLeadStatement.bind(getKeyspaceName(), this.instanceId,
-                        this.instanceAddress);
-                ResultSet lwtResult = executeStatement(boundStatement, this.consistencyLevel);
+                        this.instanceAddress).setSerialConsistencyLevel(this.consistencyLevel);
+                ResultSet lwtResult = executeStatement(boundStatement);
 
                 if (lwtResult.wasApplied()) {
                     LOGGER.debug("Took lead on schema migrations");
@@ -334,8 +335,8 @@ void removeLeadOnMigrations() {
         if (tookLead) {
             LOGGER.debug("Trying to release lead on schema migrations");
 
-            BoundStatement boundStatement = releaseMigrationLeadStatement.bind(getKeyspaceName(), this.instanceId);
-            ResultSet lwtResult = executeStatement(boundStatement, this.consistencyLevel);
+            BoundStatement boundStatement = releaseMigrationLeadStatement.bind(getKeyspaceName(), this.instanceId).setSerialConsistencyLevel(this.consistencyLevel);
+            ResultSet lwtResult = executeStatement(boundStatement);
 
             if (lwtResult.wasApplied()) {
                 LOGGER.debug("Released lead on schema migrations");
@@ -389,13 +390,11 @@ private void executeMigrationStatement(String statement, DbMigration migration)
     }
 
     private ResultSet executeStatement(String statement) throws DriverException {
-        return executeStatement(SimpleStatement.newInstance(statement), this.migrationConsistencyLevel);
+        return executeStatement(SimpleStatement.newInstance(statement).setConsistencyLevel(this.migrationConsistencyLevel));
     }
 
-    private ResultSet executeStatement(Statement<?> statement, ConsistencyLevel consistencyLevel) throws DriverException {
-        return session.execute(statement
-                .setExecutionProfileName(executionProfileName)
-                .setConsistencyLevel(consistencyLevel));
+    private ResultSet executeStatement(Statement<?> statement) throws DriverException {
+        return session.execute(statement.setExecutionProfileName(executionProfileName));
     }
 
     /**
@@ -406,8 +405,9 @@ private ResultSet executeStatement(Statement<?> statement, ConsistencyLevel cons
      */
     private void logMigration(DbMigration migration, boolean wasSuccessful) {
         BoundStatement boundStatement = logMigrationStatement.bind(wasSuccessful, migration.getVersion(),
-                migration.getScriptName(), migration.getMigrationScript(), Instant.now());
-        executeStatement(boundStatement, this.migrationConsistencyLevel);
+                migration.getScriptName(), migration.getMigrationScript(), Instant.now())
+            .setConsistencyLevel(this.migrationConsistencyLevel);
+        executeStatement(boundStatement);
     }
 
     /**
@@ -434,6 +434,18 @@ public Database setConsistencyLevel(ConsistencyLevel migrationConsistencyLevel)
         return this;
     }
 
+    /**
+     * Set the consistency level for the lightweight transaction used for taking the lead in migrations. Default is
+     * <code>ConsistencyLevel.SERIAL</code>.
+     *
+     * @param consistencyLevel The Cassandra serial consistency level to use for the lightweight transaction
+     * @return the current database instance
+     */
+    public Database setLeaderConsistencyLevel(ConsistencyLevel consistencyLevel) {
+        this.consistencyLevel = notNull(consistencyLevel, "consistencyLevel");
+        return this;
+    }
+
     /**
      * Set the execution profile name to be used for schema upgrades. This profile has to match an existing
      * profile defined in <code>application.conf</code> for the Cassandra driver. This should be set

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>org.cognitor.cassandra</groupId>
     <artifactId>cassandra-migration-parent</artifactId>
-    <version>2.6.2_v4-SNAPSHOT</version>
+    <version>2.6.3_v4-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>Cassandra Migration Parent</name>
     <url>https://github.com/patka/cassandra-migration</url>