diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java index 73dd0391c..1e567ff2e 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java @@ -28,8 +28,8 @@ import com.datastax.oss.driver.api.core.cql.Row; *

* The retrieval of the rows of a {@link ReactiveResultSet} is generally paged (a first page of result is fetched and * the next one is only fetched once all the results of the first one has been consumed). The size of the pages can be - * configured either globally through {@link com.datastax.driver.core.QueryOptions#setFetchSize} or per-statement with - * {@link com.datastax.oss.driver.api.core.cql.Statement#setFetchSize}. + * configured either globally or per-statement with + * {@link com.datastax.oss.driver.api.core.cql.SimpleStatement#setPageSize(int)}. *

* Please note however that this {@link ReactiveResultSet} paging is not available with the version 1 of the native * protocol (i.e. with Cassandra 1.2 or if version 1 has been explicitly requested). If the protocol version 1 is in @@ -42,7 +42,7 @@ import com.datastax.oss.driver.api.core.cql.Row; * @since 2.0 * @see Flux * @see ReactiveSession - * @see com.datastax.driver.core.ResultSet + * @see com.datastax.oss.driver.api.core.cql.AsyncResultSet */ public interface ReactiveResultSet { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java index 7ea4fecd1..fa6dbac55 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractSessionConfiguration.java @@ -16,7 +16,9 @@ package org.springframework.data.cassandra.config; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -31,8 +33,16 @@ import org.springframework.data.cassandra.core.cql.keyspace.DropKeyspaceSpecific import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory; import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverOption; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.internal.core.config.typesafe.DefaultDriverConfigLoader; +import com.datastax.oss.driver.internal.core.config.typesafe.DefaultProgrammaticDriverConfigLoaderBuilder; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; /** * Spring {@link @Configuration} class used to configure a Cassandra client application {@link CqlSession} connected to @@ -76,7 +86,7 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { * @since 1.5 */ @Nullable - protected SessionBuilderConfigurer getClusterBuilderConfigurer() { + protected SessionBuilderConfigurer getSessionBuilderConfigurer() { return null; } @@ -85,12 +95,25 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { * * @return the cluster name; may be {@literal null}. * @since 1.5 + * @deprecated since 3.0, use {@link #getSessionName()} instead. */ @Nullable + @Deprecated protected String getClusterName() { return null; } + /** + * Returns the session name. + * + * @return the session name; may be {@literal null}. + * @since 3.0 + */ + @Nullable + protected String getSessionName() { + return null; + } + /** * Returns the {@link CompressionType}. * @@ -139,6 +162,16 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { return Collections.emptyList(); } + /** + * Returns the local data center name used for + * {@link com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy}. + * + * @return the local data center name. + */ + protected String getLocalDataCenter() { + return null; + } + /** * Returns the list of startup scripts to be run after {@link #getKeyspaceCreations() keyspace creations} and after * initialization in the {@code system} keyspace. @@ -193,10 +226,13 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { bean.setContactPoints(getContactPoints()); bean.setPort(getPort()); + bean.setLocalDatacenter(getLocalDataCenter()); bean.setKeyspaceCreations(getKeyspaceCreations()); bean.setKeyspaceDrops(getKeyspaceDrops()); + bean.setSessionSessionBuilderConfigurer(getBuilderConfigurer()); + bean.setKeyspaceName(getKeyspaceName()); bean.setKeyspaceStartupScripts(getStartupScripts()); bean.setKeyspaceShutdownScripts(getShutdownScripts()); @@ -204,6 +240,43 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { return bean; } + private SessionBuilderConfigurer getBuilderConfigurer() { + + SessionBuilderConfigurer configurer = getSessionBuilderConfigurer(); + + return sessionBuilder -> { + + ProgrammaticDriverConfigLoaderBuilder builder = new DefaultProgrammaticDriverConfigLoaderBuilder(() -> { + + CassandraDriverOptions options = new CassandraDriverOptions(); + + if (StringUtils.hasText(getClusterName())) { + options.add(DefaultDriverOption.SESSION_NAME, getClusterName()); + } else if (StringUtils.hasText(getSessionName())) { + options.add(DefaultDriverOption.SESSION_NAME, getSessionName()); + } + + CompressionType compressionType = getCompressionType(); + if (compressionType != null) { + options.add(DefaultDriverOption.PROTOCOL_COMPRESSION, compressionType); + } + + ConfigFactory.invalidateCaches(); + return ConfigFactory.defaultOverrides().withFallback(options.build()) + .withFallback(ConfigFactory.defaultReference()).resolve(); + + }, DefaultDriverConfigLoader.DEFAULT_ROOT_PATH); + + sessionBuilder.withConfigLoader(builder.build()); + + if (configurer != null) { + return configurer.configure(sessionBuilder); + } + + return sessionBuilder; + }; + } + /** * Creates a {@link CqlTemplate} configured with {@link #getRequiredSessionFactory()}. * @@ -220,4 +293,38 @@ public abstract class AbstractSessionConfiguration implements BeanFactoryAware { this.beanFactory = beanFactory; } + private static class CassandraDriverOptions { + + private final Map options = new LinkedHashMap<>(); + + private CassandraDriverOptions add(DriverOption option, String value) { + String key = createKeyFor(option); + this.options.put(key, value); + return this; + } + + private CassandraDriverOptions add(DriverOption option, int value) { + return add(option, String.valueOf(value)); + } + + private CassandraDriverOptions add(DriverOption option, Enum value) { + return add(option, value.name()); + } + + private CassandraDriverOptions add(DriverOption option, List values) { + for (int i = 0; i < values.size(); i++) { + this.options.put(String.format("%s.%s", createKeyFor(option), i), values.get(i)); + } + return this; + } + + private Config build() { + return ConfigFactory.parseMap(this.options, "Environment"); + } + + private static String createKeyFor(DriverOption option) { + return String.format("%s.%s", DefaultDriverConfigLoader.DEFAULT_ROOT_PATH, option.getPath()); + } + } + } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java index 6ab81252b..0a81ff5aa 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CqlSessionFactoryBean.java @@ -91,6 +91,8 @@ public class CqlSessionFactoryBean private @Nullable String keyspaceName; private @Nullable String localDatacenter; + private @Nullable SessionBuilderConfigurer sessionSessionBuilderConfigurer; + private List keyspaceActions = new ArrayList<>(); private Set keyspaceSpecifications = new HashSet<>(); @@ -294,6 +296,16 @@ public class CqlSessionFactoryBean return session; } + /** + * Sets the {@link SessionBuilderConfigurer} to configure the + * {@link com.datastax.oss.driver.api.core.session.SessionBuilder}. + * + * @param sessionSessionBuilderConfigurer + */ + public void setSessionSessionBuilderConfigurer(@Nullable SessionBuilderConfigurer sessionSessionBuilderConfigurer) { + this.sessionSessionBuilderConfigurer = sessionSessionBuilderConfigurer; + } + /** * Sets CQL scripts to be executed immediately after the session is connected. * @@ -490,6 +502,10 @@ public class CqlSessionFactoryBean builder.withLocalDatacenter(this.localDatacenter); } + if (this.sessionSessionBuilderConfigurer != null) { + return this.sessionSessionBuilderConfigurer.configure(builder); + } + return builder; } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionBuilderConfigurer.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionBuilderConfigurer.java index 92f7ed00e..067d945ec 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionBuilderConfigurer.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/SessionBuilderConfigurer.java @@ -15,6 +15,7 @@ */ package org.springframework.data.cassandra.config; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; import com.datastax.oss.driver.api.core.session.SessionBuilder; /** @@ -35,5 +36,5 @@ public interface SessionBuilderConfigurer { * @return the argument to the {@code sessionBuilder} parameter. * @see SessionBuilder */ - SessionBuilder configure(SessionBuilder sessionBuilder); + CqlSessionBuilder configure(CqlSessionBuilder sessionBuilder); } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlTemplateConfigIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlTemplateConfigIntegrationTests.java index f93c9063f..431455ab9 100755 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlTemplateConfigIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CqlTemplateConfigIntegrationTests.java @@ -46,6 +46,11 @@ public class CqlTemplateConfigIntegrationTests extends AbstractEmbeddedCassandra return "system"; } + @Override + protected String getLocalDataCenter() { + return "datacenter1"; + } + @Override protected int getPort() { return cassandraEnvironment.getPort(); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/AbstractTestJavaConfig.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/AbstractTestJavaConfig.java index b80b75e10..350d27dc5 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/AbstractTestJavaConfig.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/support/AbstractTestJavaConfig.java @@ -34,4 +34,8 @@ public abstract class AbstractTestJavaConfig extends AbstractSessionConfiguratio return PROPERTIES.getCassandraPort(); } + @Override + protected String getLocalDataCenter() { + return "datacenter1"; + } } diff --git a/src/main/asciidoc/reference/migration-guide-2.2-to-3.0.adoc b/src/main/asciidoc/reference/migration-guide-2.2-to-3.0.adoc index 16517f85d..59fac897c 100644 --- a/src/main/asciidoc/reference/migration-guide-2.2-to-3.0.adoc +++ b/src/main/asciidoc/reference/migration-guide-2.2-to-3.0.adoc @@ -6,6 +6,7 @@ Spring Data for Apache Cassandra 3.0 introduces a set of breaking changes when u == Review dependencies Upgrading to Spring Data Cassandra requires an upgrade to the DataStax Driver version 4. Upgrading to the new driver comes with transitive dependency changes, most notably, Google Guava is bundled and shaded by the driver. +Check out the https://docs.datastax.com/en/developer/java-driver/4.3/upgrade_guide/[DataStax Java Driver for Apache Cassandra 4 Upgrade Guide] for details on the Driver-related changes. == Adapt Configuration @@ -15,6 +16,9 @@ This means that `SocketOptions`, `AddressTranslator` and many more options are c If you're using XML-based configuration, make sure to migrate all configuration files from the `cql` namespace (`http://www.springframework.org/schema/cql https://www.springframework.org/schema/cql/spring-cql.xsd`) to the `cassandra` namespace (`http://www.springframework.org/schema/data/cassandra https://www.springframework.org/schema/data/cassandra/spring-cassandra.xsd`). +To reflect the change in configuration builders, `ClusterBuilderConfigurer` was renamed to `SessionBuilderConfigurer` accepting now `CqlSessionBuilder` instead of the `Cluster.Builder`. +Make sure to also provide the local data center in your configuration as it is required to properly configure load balancing. + === Connectivity The configuration elements for `Cluster` (`cassandra:cluster`) and `Session` (`cassandra:session`) were merged into a single `CqlSession` (`cassandra:session`) element that configures both, the keyspace and endpoints. @@ -39,7 +43,7 @@ With the upgrade, schema support was moved to a new namespace element: `cassandr ==== [source,xml] ---- - + @@ -140,6 +144,7 @@ Paging state now uses `ByteBuffer`. * `KeyspaceIdentifier` and `CqlIdentifier`, use `com.datastax.oss.driver.api.core.CqlIdentifier` instead. * `CassandraSessionFactoryBean`, use `CqlSessionFactoryBean` instead. * `AbstractCqlTemplateConfiguration`, use `AbstractSessionConfiguration` instead. +* `AbstractSessionConfiguration.getClusterName()`, use `AbstractSessionConfiguration.getSessionName()` instead. * `CodecRegistryTupleTypeFactory`, use `SimpleTupleTypeFactory` instead. * Spring Data's `CqlIdentifier`, use the driver `CqlIdentifier` instead. * `forceQuote` attributes as quoting is no longer required. `CqlIdentifier` properly escapes reserved keywords and takes care of case-sensitivity.