From e8bb261e12e2b531e61eac49ad9716f6daa154fa Mon Sep 17 00:00:00 2001 From: Jeff Blaisdell Date: Fri, 27 Dec 2019 12:41:30 -0600 Subject: [PATCH] Upgrade Cassandra Driver to 4.3.1 --- CHANGELOG.md | 4 + build.gradle | 20 ++--- gradle/wrapper/gradle-wrapper.properties | 2 +- .../cassandra/CassandraConnection.java | 63 +++++++++++----- .../smartthings/cassandra/CassandraLock.java | 74 ++++++++++++++----- .../migration/MigrationParameters.java | 49 ++++++++---- .../migration/MigrationRunner.java | 12 +-- src/main/java/smartthings/util/Util.java | 6 +- .../cassandra/CassandraConnectionSpec.groovy | 62 ++++++++++++---- .../migration/MigrationRunnerSpec.groovy | 20 ++--- src/test/resources/test-cassandra.yaml | 2 + version.txt | 2 +- 12 files changed, 216 insertions(+), 100 deletions(-) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..975c5df --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,4 @@ +| Version | Change | +|---------|--------------------------------------------------| +| 0.1.6 | Dependent on Cassandra Driver 3.2.0 and Guava 19 | +| 0.2.0 | Dependent on Cassandra Driver 4.3.1 | diff --git a/build.gradle b/build.gradle index f28f673..1ec7dc3 100644 --- a/build.gradle +++ b/build.gradle @@ -4,8 +4,8 @@ buildscript { } dependencies { - classpath 'com.jfrog.bintray.gradle:gradle-bintray-plugin:1.7.1' - classpath 'com.github.ben-manes:gradle-versions-plugin:0.13.0' + classpath 'com.jfrog.bintray.gradle:gradle-bintray-plugin:1.8.4' + classpath 'com.github.ben-manes:gradle-versions-plugin:0.27.0' } } @@ -24,20 +24,22 @@ targetCompatibility = "1.8" repositories { + mavenLocal() + maven { + url = "https://oss.sonatype.org/content/repositories/snapshots/" + } mavenCentral() } dependencies { - compile 'com.datastax.cassandra:cassandra-driver-core:3.2.0' - compile 'com.google.guava:guava:19.0' + compile 'com.datastax.oss:java-driver-core:4.3.1' testCompile 'org.spockframework:spock-core:1.0-groovy-2.4' testCompile 'org.codehaus.groovy:groovy-all:2.4.7' - testCompile 'org.cassandraunit:cassandra-unit:3.0.0.1' - testCompile 'cglib:cglib-nodep:3.2.4' - testCompile 'org.objenesis:objenesis:2.4' - testRuntime 'io.netty:netty-all:4.1.4.Final' - testRuntime 'ch.qos.logback:logback-classic:1.1.7' + testCompile 'org.cassandraunit:cassandra-unit:4.2.2.0-SNAPSHOT' + testCompile 'cglib:cglib-nodep:3.3.0' + testCompile 'org.objenesis:objenesis:3.1' + testRuntime 'ch.qos.logback:logback-classic:1.2.3' } jacocoTestReport { diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index a95009c..9492014 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,5 +1,5 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-4.9-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-6.0.1-bin.zip zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/src/main/java/smartthings/cassandra/CassandraConnection.java b/src/main/java/smartthings/cassandra/CassandraConnection.java index 904bffe..562b209 100644 --- a/src/main/java/smartthings/cassandra/CassandraConnection.java +++ b/src/main/java/smartthings/cassandra/CassandraConnection.java @@ -1,8 +1,14 @@ package smartthings.cassandra; -import com.datastax.driver.core.*; -import com.google.common.base.Charsets; -import com.google.common.io.Files; +import com.datastax.oss.driver.api.core.*; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +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.ssl.ProgrammaticSslEngineFactory; +import com.datastax.oss.driver.shaded.guava.common.base.Charsets; +import com.datastax.oss.driver.shaded.guava.common.io.Files; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import smartthings.migration.CassandraMigrationException; @@ -14,8 +20,10 @@ import java.io.File; import java.io.FileInputStream; import java.io.IOException; +import java.net.InetSocketAddress; import java.security.KeyStore; import java.security.SecureRandom; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -30,11 +38,12 @@ public class CassandraConnection implements AutoCloseable { private String truststorePassword; private String keystorePath; private String keystorePassword; - private Cluster cluster; - private Session session; + private CqlSession session; + private boolean mySession = false; private String keyspace; private String host; private int port; + private String localDatacenter; private String username; private String password; @@ -53,38 +62,44 @@ public CassandraConnection(MigrationParameters parameters, String ownerName) { if (session == null) { this.host = parameters.getHost(); this.port = parameters.getPort(); + this.localDatacenter = parameters.getLocalDatacenter(); this.username = parameters.getUsername(); this.password = parameters.getPassword(); this.truststorePassword = parameters.getTruststorePassword(); this.truststorePath = parameters.getTruststorePath(); this.keystorePassword = parameters.getKeystorePassword(); this.keystorePath = parameters.getKeystorePath(); + this.mySession = true; } this.keyspace = parameters.getKeyspace(); - } public void connect() throws Exception { if (session == null) { logger.debug("Connecting to Cassandra at " + host + ":" + port); - QueryOptions queryOptions = new QueryOptions().setConsistencyLevel(ConsistencyLevel.QUORUM); - - Cluster.Builder builder = Cluster.builder().addContactPoint(host).withPort(port).withMaxSchemaAgreementWaitSeconds(20).withQueryOptions(queryOptions); + CqlSessionBuilder builder = CqlSession.builder() + .addContactPoint(new InetSocketAddress(host, port)) + .withLocalDatacenter(localDatacenter) + .withConfigLoader( + DriverConfigLoader.programmaticBuilder() + .withDuration(DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_TIMEOUT, Duration.ofSeconds(20)) + .withString(DefaultDriverOption.REQUEST_CONSISTENCY, ConsistencyLevel.QUORUM.toString()) + .build() + ); if (all(truststorePath, truststorePassword, keystorePath, keystorePassword)) { logger.debug("Using SSL for the connection"); SSLContext sslContext = getSSLContext(truststorePath, truststorePassword, keystorePath, keystorePassword); - builder.withSSL(JdkSSLOptions.builder().withSSLContext(sslContext).withCipherSuites(cipherSuites).build()); + builder.withSslEngineFactory(new ProgrammaticSslEngineFactory(sslContext, cipherSuites)); } if (username != null && password != null) { logger.debug("Using withCredentials for the connection"); - builder.withCredentials(username, password); + builder.withAuthCredentials(username, password); } - cluster = builder.build(); - session = cluster.connect(); + session = builder.build(); } if (keyspace != null) { @@ -103,9 +118,9 @@ public void close() { lock.unlock(); } - if (cluster != null) { + if (session != null && isMySession()) { //We don't close the connection if we were given a session - cluster.close(); + session.close(); } } @@ -135,7 +150,7 @@ public void setKeyspace(String keyspace) { } public ResultSet execute(String query, Object... params) { - return session.execute(query, params); + return session.execute(SimpleStatement.newInstance(query, params)); } @@ -174,7 +189,7 @@ public void backfillMigrations() { } public void setupMigration() { - if (!session.getCluster().getMetadata().checkSchemaAgreement()) { + if (!session.checkSchemaAgreement()) { throw new CassandraMigrationException("Migration table setup precheck: schema not in agreement"); } if (!tableExists("migrations")) { @@ -291,18 +306,18 @@ public void keepLockAlive() { public String getMigrationMd5(String fileName) { File file = new File(fileName); ResultSet result = executeWithLock("SELECT sha FROM migrations WHERE name=?", file.getName()); - if (result.isExhausted()) { + if (result.isFullyFetched()) { return null; } return result.one().getString("sha"); } - public Session getSession() { + public CqlSession getSession() { return session; } - public void setSession(Session session) { + public void setSession(CqlSession session) { this.session = session; } @@ -377,4 +392,12 @@ public void setKeystorePassword(String keystorePassword) { public String getOwnerName() { return ownerName; } + + /** + * Returns `true` if CassandraConnection created it's own CqlSession. `false` if connection was parameterized by caller. + * @return + */ + public boolean isMySession() { + return mySession; + } } diff --git a/src/main/java/smartthings/cassandra/CassandraLock.java b/src/main/java/smartthings/cassandra/CassandraLock.java index e71d91e..7c26b0d 100644 --- a/src/main/java/smartthings/cassandra/CassandraLock.java +++ b/src/main/java/smartthings/cassandra/CassandraLock.java @@ -1,6 +1,8 @@ package smartthings.cassandra; -import com.datastax.driver.core.*; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; +import com.datastax.oss.driver.api.core.cql.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import smartthings.migration.CassandraMigrationException; @@ -13,7 +15,7 @@ public class CassandraLock implements AutoCloseable { private final int ttl; private final CassandraConnection cassandraConnection; private final String owner; - private final Session session; + private final CqlSession session; private final PreparedStatement insertLock; private final PreparedStatement deleteLock; private final PreparedStatement selectLock; @@ -31,22 +33,35 @@ public CassandraLock(CassandraConnection cassandraConnection, int ttl) { setupTables(); - insertLock = session.prepare("INSERT INTO databasechangelock(id, lockedby) VALUES (:lockId, :owner) IF NOT EXISTS USING TTL :ttl"); - insertLock.setConsistencyLevel(ConsistencyLevel.QUORUM); - - deleteLock = session.prepare("DELETE FROM databasechangelock WHERE id = :lockId IF lockedby = :owner"); - deleteLock.setConsistencyLevel(ConsistencyLevel.QUORUM); - - selectLock = session.prepare("SELECT lockedby, TTL(lockedby) AS ttl FROM databasechangelock WHERE id = :lockId"); - selectLock.setConsistencyLevel(ConsistencyLevel.SERIAL); - - updateLock = session.prepare("UPDATE databasechangelock USING TTL :ttl SET lockedby = :owner WHERE id = :lockId IF lockedby = :owner"); - deleteLock.setConsistencyLevel(ConsistencyLevel.QUORUM); + insertLock = session.prepare( + SimpleStatement.newInstance("INSERT INTO databasechangelock(id, lockedby) VALUES (:lockId, :owner) IF NOT EXISTS USING TTL :ttl") + .setConsistencyLevel(DefaultConsistencyLevel.QUORUM) + ); + + deleteLock = session.prepare( + SimpleStatement.newInstance("DELETE FROM databasechangelock WHERE id = :lockId IF lockedby = :owner") + .setConsistencyLevel(DefaultConsistencyLevel.QUORUM) + ); + + selectLock = session.prepare( + SimpleStatement.newInstance("SELECT lockedby, TTL(lockedby) AS ttl FROM databasechangelock WHERE id = :lockId") + .setConsistencyLevel(DefaultConsistencyLevel.SERIAL) + ); + + updateLock = session.prepare( + SimpleStatement.newInstance("UPDATE databasechangelock USING TTL :ttl SET lockedby = :owner WHERE id = :lockId IF lockedby = :ifowner") + .setConsistencyLevel(DefaultConsistencyLevel.QUORUM) + ); } public boolean tryLock() { - ResultSet rs = session.execute(insertLock.bind().setInt("lockId", lockId) - .setInt("ttl", ttl).setString("owner", owner)); + ResultSet rs = session.execute( + insertLock.boundStatementBuilder() + .setInt("lockId", lockId) + .setInt("ttl", ttl) + .setString("owner", owner) + .build() + ); if (rs.wasApplied()) { return true; } else { @@ -58,7 +73,12 @@ public boolean tryLock() { public void unlock() { // Only try to release lock if its mine if (isMine()) { - ResultSet rs = session.execute(deleteLock.bind().setInt("lockId", lockId).setString("owner", owner)); + ResultSet rs = session.execute( + deleteLock.boundStatementBuilder() + .setInt("lockId", lockId) + .setString("owner", owner) + .build() + ); if (!rs.wasApplied()) { // if ownership was lost, should be fine, since we are relinquishing ownership if (isMine()) { @@ -69,15 +89,25 @@ public void unlock() { } public void keepAlive() { - ResultSet rs = session.execute(updateLock.bind().setInt("lockId", lockId) - .setInt("ttl", ttl).setString("owner", owner)); + ResultSet rs = session.execute( + updateLock.boundStatementBuilder() + .setInt("lockId", lockId) + .setInt("ttl", ttl) + .setString("owner", owner) + .setString("ifowner", owner) + .build() + ); if (!rs.wasApplied()) { throw new CassandraLockException("unable to keep alive lock"); } } public String getOwner() { - Row row = session.execute(selectLock.bind().setInt("lockId", lockId)).one(); + Row row = session.execute( + selectLock.boundStatementBuilder() + .setInt("lockId", lockId) + .build() + ).one(); if (row != null) { return row.getString("lockedby"); } @@ -86,7 +116,11 @@ public String getOwner() { } public int getTtl() { - Row row = session.execute(selectLock.bind().setInt("lockId", lockId)).one(); + Row row = session.execute( + selectLock.boundStatementBuilder() + .setInt("lockId", lockId) + .build() + ).one(); if (row != null) { return row.getInt("ttl"); } diff --git a/src/main/java/smartthings/migration/MigrationParameters.java b/src/main/java/smartthings/migration/MigrationParameters.java index 8a04b3b..70529e1 100644 --- a/src/main/java/smartthings/migration/MigrationParameters.java +++ b/src/main/java/smartthings/migration/MigrationParameters.java @@ -1,13 +1,13 @@ package smartthings.migration; -import com.datastax.driver.core.Session; - +import com.datastax.oss.driver.api.core.CqlSession; import java.io.File; -import java.util.Optional; import java.util.UUID; public class MigrationParameters { + private final static String DEFAULT_LOCAL_DC = "datacenter1"; + private Boolean override; private HandlerClass handlerClass; private File migrationFile; @@ -25,11 +25,15 @@ public class MigrationParameters { private String keystorePassword; private String leaderId = UUID.randomUUID().toString(); - private Session session; + // @See https://docs.datastax.com/en/developer/java-driver/4.2/manual/core/load_balancing/ + private String localDatacenter = "datacenter1"; + + private CqlSession session; public MigrationParameters() { host = System.getProperty("host", "localhost"); port = Integer.parseInt(System.getProperty("port", "9042")); + localDatacenter = System.getProperty("localDatacenter", "datacenter1"); keyspace = System.getProperty("keyspace", "test"); migrationsPath = System.getProperty("migrationPath", "../migrations"); @@ -48,11 +52,11 @@ public MigrationParameters() { override = new Boolean(System.getProperty("override")); } - public MigrationParameters(Boolean override, HandlerClass handlerClass, File migrationFile, String host, String keyspace, String location, String migrationsPath, String password, String username, int port, String truststorePassword, String truststorePath, String keystorePassword, String keystorePath, String migrationsLogFile) { - this(override, handlerClass, migrationFile, host, keyspace, location, migrationsPath, password, username, port, truststorePassword, truststorePath, keystorePassword, keystorePath, migrationsLogFile, null); + public MigrationParameters(Boolean override, HandlerClass handlerClass, File migrationFile, String host, String keyspace, String location, String migrationsPath, String password, String username, int port, String truststorePassword, String truststorePath, String keystorePassword, String keystorePath, String migrationsLogFile, String localDatacenter) { + this(override, handlerClass, migrationFile, host, keyspace, location, migrationsPath, password, username, port, truststorePassword, truststorePath, keystorePassword, keystorePath, migrationsLogFile, localDatacenter, null); } - public MigrationParameters(Boolean override, HandlerClass handlerClass, File migrationFile, String host, String keyspace, String location, String migrationsPath, String password, String username, int port, String truststorePassword, String truststorePath, String keystorePassword, String keystorePath, String migrationsLogFile, Session session) { + public MigrationParameters(Boolean override, HandlerClass handlerClass, File migrationFile, String host, String keyspace, String location, String migrationsPath, String password, String username, int port, String truststorePassword, String truststorePath, String keystorePassword, String keystorePath, String migrationsLogFile, String localDatacenter, CqlSession session) { this.override = override; this.handlerClass = handlerClass; this.migrationFile = migrationFile; @@ -68,16 +72,16 @@ public MigrationParameters(Boolean override, HandlerClass handlerClass, File mig this.truststorePassword = truststorePassword; this.truststorePath = truststorePath; this.migrationsLogFile = migrationsLogFile; + this.localDatacenter = localDatacenter; this.session = session; } - public MigrationParameters(String migrationsLogFile, String keyspace, Session session) { + public MigrationParameters(String migrationsLogFile, String keyspace, CqlSession session) { this.override = false; this.migrationsLogFile = migrationsLogFile; this.keyspace = keyspace; this.session = session; this.handlerClass = HandlerClass.MigrationHandler; - } public String toString() { @@ -209,11 +213,11 @@ public void setMigrationsLogFile(String migrationsLogFile) { this.migrationsLogFile = migrationsLogFile; } - public Session getSession() { + public CqlSession getSession() { return session; } - public void setSession(Session session) { + public void setSession(CqlSession session) { this.session = session; } @@ -221,6 +225,16 @@ public void setSession(Session session) { public void setLeaderId(String leaderId) { this.leaderId = leaderId; } + public String getLocalDatacenter() { + if (localDatacenter == null || localDatacenter.isEmpty()) { + return DEFAULT_LOCAL_DC; + } + return localDatacenter; + } + + public void setLocalDatacenter(String localDatacenter) { + this.localDatacenter = localDatacenter; + } public static class Builder { private File migrationFile; @@ -236,8 +250,9 @@ public static class Builder { private String truststorePassword; private String keystorePath; private String keystorePassword; + private String localDatacenter; - private Session session; + private CqlSession session; public Builder() {} @@ -301,15 +316,19 @@ public Builder setMigrationsLogFile(String migrationsLogFile) { return this; } - public Builder setSession(Session session) { + public Builder setLocalDatacenter(String localDatacenter) { + this.localDatacenter = localDatacenter; + return this; + } + + public Builder setSession(CqlSession session) { this.session = session; return this; } public MigrationParameters build() { - if (session == null) { - return new MigrationParameters(false, HandlerClass.MigrationHandler, migrationFile, host, keyspace, null, migrationsPath, password, username, port, truststorePassword, truststorePath, keystorePassword, keystorePath, migrationsLogFile); + return new MigrationParameters(false, HandlerClass.MigrationHandler, migrationFile, host, keyspace, null, migrationsPath, password, username, port, truststorePassword, truststorePath, keystorePassword, keystorePath, migrationsLogFile, localDatacenter); } else { return new MigrationParameters(migrationsLogFile, keyspace, session); } diff --git a/src/main/java/smartthings/migration/MigrationRunner.java b/src/main/java/smartthings/migration/MigrationRunner.java index 283773f..c4f5e8d 100644 --- a/src/main/java/smartthings/migration/MigrationRunner.java +++ b/src/main/java/smartthings/migration/MigrationRunner.java @@ -1,9 +1,9 @@ package smartthings.migration; -import com.google.common.base.Charsets; -import com.google.common.io.CharSource; -import com.google.common.io.Files; -import com.google.common.io.Resources; +import com.datastax.oss.driver.shaded.guava.common.base.Charsets; +import com.datastax.oss.driver.shaded.guava.common.io.CharSource; +import com.datastax.oss.driver.shaded.guava.common.io.Files; +import com.datastax.oss.driver.shaded.guava.common.io.Resources; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import smartthings.cassandra.CassandraConnection; @@ -99,7 +99,7 @@ private void doMigration(CassandraConnection connection, MigrationParameters mig } } else if (migrationParameters.getMigrationFile() != null) { File f = migrationParameters.getMigrationFile(); - handler.handle(f.getName(), Files.toString(f, Charsets.UTF_8)); + handler.handle(f.getName(), Files.asCharSource(f, Charsets.UTF_8).read()); } else { File migrationsDir = migrationParameters.getMigrationsDir(); logger.info("Using migrations Directory " + migrationsDir); @@ -107,7 +107,7 @@ private void doMigration(CassandraConnection connection, MigrationParameters mig File[] files = migrationsDir.listFiles(); if (files != null) { for (File file : files) { - handler.handle(file.getName(), Files.toString(file, Charsets.UTF_8)); + handler.handle(file.getName(), Files.asCharSource(file, Charsets.UTF_8).read()); } } else { logger.warn("No files found in migrations directory."); diff --git a/src/main/java/smartthings/util/Util.java b/src/main/java/smartthings/util/Util.java index 5fc0800..0f73bc3 100644 --- a/src/main/java/smartthings/util/Util.java +++ b/src/main/java/smartthings/util/Util.java @@ -1,8 +1,8 @@ package smartthings.util; -import com.google.common.base.Charsets; -import com.google.common.hash.Hashing; -import com.google.common.io.Files; +import com.datastax.oss.driver.shaded.guava.common.base.Charsets; +import com.datastax.oss.driver.shaded.guava.common.hash.Hashing; +import com.datastax.oss.driver.shaded.guava.common.io.Files; import java.io.File; import java.io.IOException; diff --git a/src/test/groovy/smartthings/cassandra/CassandraConnectionSpec.groovy b/src/test/groovy/smartthings/cassandra/CassandraConnectionSpec.groovy index 0df52ce..eef321a 100644 --- a/src/test/groovy/smartthings/cassandra/CassandraConnectionSpec.groovy +++ b/src/test/groovy/smartthings/cassandra/CassandraConnectionSpec.groovy @@ -1,6 +1,11 @@ package smartthings.cassandra -import com.datastax.driver.core.* +import com.datastax.oss.driver.api.core.ConsistencyLevel +import com.datastax.oss.driver.api.core.CqlSession +import com.datastax.oss.driver.api.core.cql.ExecutionInfo +import com.datastax.oss.driver.api.core.cql.ResultSet +import com.datastax.oss.driver.api.core.cql.Row +import com.datastax.oss.driver.internal.core.cql.DefaultSimpleStatement import smartthings.migration.CassandraMigrationException import smartthings.migration.MigrationParameters import spock.lang.Specification @@ -9,8 +14,8 @@ class CassandraConnectionSpec extends Specification { CassandraConnection cassandraConnection - Session session = Mock() - CassandraLock lock = Mock() + CqlSession session = Mock(CqlSession) + CassandraLock lock = Mock(CassandraLock) void setup() { MigrationParameters parameters = new MigrationParameters.Builder().setSession(session).build() @@ -21,18 +26,24 @@ class CassandraConnectionSpec extends Specification { def "Migration is successful"() { setup: String migrationFileName = 'make-a-table.cql' - ResultSet migrationsResultSet = Mock() - ResultSet createResultSet = Mock() - ExecutionInfo createExecutionInfo = Mock() + ResultSet migrationsResultSet = Mock(ResultSet) + ResultSet createResultSet = Mock(ResultSet) + ExecutionInfo createExecutionInfo = Mock(ExecutionInfo) when: cassandraConnection.runMigration(migrationFileName, 'CREATE TABLE;', 'SHA1', false) then: - 1 * session.execute('INSERT INTO migrations (name, sha) VALUES (?, ?) IF NOT EXISTS;', [migrationFileName, 'SHA1']) >> migrationsResultSet + 1 * session.execute({ statement -> + statement.query == 'INSERT INTO migrations (name, sha) VALUES (?, ?) IF NOT EXISTS;' && + statement.positionalValues.get(0) == migrationFileName && + statement.positionalValues.get(1) == 'SHA1' + }) >> migrationsResultSet 1 * migrationsResultSet.wasApplied() >> true - 1 * session.execute('CREATE TABLE;') >> createResultSet + 1 * session.execute({ statement -> + statement.query == 'CREATE TABLE;' + }) >> createResultSet 1 * createResultSet.getExecutionInfo() >> createExecutionInfo 1 * createExecutionInfo.isSchemaInAgreement() >> true _ * lock.isMine() >> true @@ -49,7 +60,11 @@ class CassandraConnectionSpec extends Specification { cassandraConnection.runMigration(migrationFileName, 'CREATE TABLE;', 'SHA1', false) then: - 1 * session.execute('INSERT INTO migrations (name, sha) VALUES (?, ?) IF NOT EXISTS;', [migrationFileName, 'SHA1']) >> migrationsResultSet + 1 * session.execute({ statement -> + statement.query == 'INSERT INTO migrations (name, sha) VALUES (?, ?) IF NOT EXISTS;' && + statement.positionalValues.get(0) == migrationFileName && + statement.positionalValues.get(1) == 'SHA1' + }) >> migrationsResultSet 1 * migrationsResultSet.wasApplied() >> false _ * lock.isMine() >> true _ * lock.keepAlive() @@ -68,12 +83,21 @@ class CassandraConnectionSpec extends Specification { cassandraConnection.runMigration(migrationFileName, 'CREATE TABLE;', 'SHA1', false) then: - 1 * session.execute('INSERT INTO migrations (name, sha) VALUES (?, ?) IF NOT EXISTS;', [migrationFileName, 'SHA1']) >> migrationsResultSet + 1 * session.execute({ statement -> + statement.query == 'INSERT INTO migrations (name, sha) VALUES (?, ?) IF NOT EXISTS;' && + statement.positionalValues.get(0) == migrationFileName && + statement.positionalValues.get(1) == 'SHA1' + }) >> migrationsResultSet 1 * migrationsResultSet.wasApplied() >> true - 1 * session.execute('CREATE TABLE;') >> createResultSet + 1 * session.execute({ statement -> + statement.query == 'CREATE TABLE;' + }) >> createResultSet 1 * createResultSet.getExecutionInfo() >> createExecutionInfo 1 * createExecutionInfo.isSchemaInAgreement() >> false - 1 * session.execute('DELETE FROM migrations WHERE name = ? IF EXISTS', [migrationFileName]) >> removeResultSet + 1 * session.execute({ statement -> + statement.query == 'DELETE FROM migrations WHERE name = ? IF EXISTS' && + statement.positionalValues.get(0) == migrationFileName + }) >> removeResultSet 1 * removeResultSet.wasApplied() >> true _ * lock.isMine() >> true _ * lock.keepAlive() @@ -90,8 +114,11 @@ class CassandraConnectionSpec extends Specification { String result = cassandraConnection.getMigrationMd5('/tmp/add-column.cql') then: - 1 * session.execute('SELECT sha FROM migrations WHERE name=?', ['add-column.cql']) >> resultSet - 1 * resultSet.isExhausted() >> false + 1 * session.execute({ statement -> + statement.query == 'SELECT sha FROM migrations WHERE name=?' && + statement.positionalValues.get(0) == 'add-column.cql' + }) >> resultSet + 1 * resultSet.isFullyFetched() >> false 1 * resultSet.one() >> row 1 * row.getString('sha') >> '1234567890' _ * lock.isMine() >> true @@ -108,8 +135,11 @@ class CassandraConnectionSpec extends Specification { String result = cassandraConnection.getMigrationMd5('/tmp/add-column.cql') then: - 1 * session.execute('SELECT sha FROM migrations WHERE name=?', ['add-column.cql']) >> resultSet - 1 * resultSet.isExhausted() >> true + 1 * session.execute({ statement -> + statement.query == 'SELECT sha FROM migrations WHERE name=?' && + statement.positionalValues.get(0) == 'add-column.cql' + }) >> resultSet + 1 * resultSet.isFullyFetched() >> true _ * lock.isMine() >> true _ * lock.keepAlive() 0 * _ diff --git a/src/test/groovy/smartthings/migration/MigrationRunnerSpec.groovy b/src/test/groovy/smartthings/migration/MigrationRunnerSpec.groovy index 1739c81..7f9a124 100644 --- a/src/test/groovy/smartthings/migration/MigrationRunnerSpec.groovy +++ b/src/test/groovy/smartthings/migration/MigrationRunnerSpec.groovy @@ -1,7 +1,8 @@ package smartthings.migration -import com.datastax.driver.core.Cluster -import com.datastax.driver.core.ResultSet +import com.datastax.oss.driver.api.core.CqlSession +import com.datastax.oss.driver.api.core.CqlSessionBuilder +import com.datastax.oss.driver.api.core.cql.ResultSet import org.cassandraunit.CassandraCQLUnit import org.cassandraunit.dataset.CQLDataSet import org.cassandraunit.dataset.cql.ClassPathCQLDataSet @@ -73,12 +74,13 @@ class MigrationRunnerSpec extends Specification { def 'run migrations via session'() { given: - - Cluster.Builder builder = Cluster.builder().addContactPoint('localhost').withPort(9142); - - def params = new MigrationParameters.Builder() - .setSession(builder.build().connect()) + .setSession( + CqlSessionBuilder.newInstance() + .addContactPoint(new InetSocketAddress('localhost', 9142)) + .withLocalDatacenter("datacenter1") + .build() + ) .setKeyspace(keyspace) .setMigrationsLogFile('/cassandra/success.changelog') .build() @@ -206,8 +208,8 @@ class MigrationRunnerSpec extends Specification { List processRows(ResultSet results) { results.all().collect { row -> row.columnDefinitions.collect { column -> - assert column.keyspace == keyspace - [column.name, row.getString(column.name)] + assert column.keyspace.asCql(true) == keyspace + [column.name.asCql(true), row.getString(column.name)] }.collectEntries() } } diff --git a/src/test/resources/test-cassandra.yaml b/src/test/resources/test-cassandra.yaml index 3872747..a3f9069 100644 --- a/src/test/resources/test-cassandra.yaml +++ b/src/test/resources/test-cassandra.yaml @@ -586,3 +586,5 @@ encryption_options: # cipher_suites: [TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA] hints_directory: /tmp/hints + +cdc_raw_directory: build/embeddedCassandra/raw diff --git a/version.txt b/version.txt index b8181ab..e15743d 100644 --- a/version.txt +++ b/version.txt @@ -1,2 +1,2 @@ -0.1.7-SNAPSHOT +0.2.0-SNAPSHOT