DATACASS-656 - Polishing.

Pick up compression type and session name properties along with SessionBuilderConfigurer.

Include LocalDataCenter in migration guide.

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;
  * <p>
  * 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)}.
  * <p>
  * 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 {
 

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<String, String> 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<String> 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());
+		}
+	}
+
 }

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> keyspaceActions = new ArrayList<>();
 	private Set<KeyspaceActionSpecification> 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;
 	}
 

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);
 }

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();

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";
+	}
 }

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]
 ----
-<cassandra:session contact-points="localhost" port="9042" keyspace="mykeyspace">
+<cassandra:session contact-points="localhost" port="9042" keyspace="mykeyspace" local-datacenter="datacenter1">
   <cassandra:keyspace action="CREATE_DROP" name="mykeyspace" />
 </cassandra:session>
 
@@ -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.