Review Cassandra's timeout options

This commit reviews the current timeout options. It creates a connection
and request sub-namespace to separate concerns a bit more.

Closes gh-19673

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java

@@ -39,6 +39,8 @@ import com.typesafe.config.ConfigFactory;
 
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.cassandra.CassandraProperties.Connection;
+import org.springframework.boot.autoconfigure.cassandra.CassandraProperties.Request;
 import org.springframework.boot.autoconfigure.cassandra.CassandraProperties.Throttler;
 import org.springframework.boot.autoconfigure.cassandra.CassandraProperties.ThrottlerType;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -115,10 +117,9 @@ public class CassandraAutoConfiguration {
 						.add(DefaultDriverOption.AUTH_PROVIDER_PASSWORD, properties.getPassword()));
 		map.from(properties::getCompression).whenNonNull()
 				.to((compression) -> options.add(DefaultDriverOption.PROTOCOL_COMPRESSION, compression));
-		mapQueryOptions(properties, options);
-		mapSocketOptions(properties, options);
+		mapConnectionOptions(properties, options);
 		mapPoolingOptions(properties, options);
-		mapThrottlingOptions(properties, options);
+		mapRequestOptions(properties, options);
 		map.from(mapContactPoints(properties))
 				.to((contactPoints) -> options.add(DefaultDriverOption.CONTACT_POINTS, contactPoints));
 		map.from(properties.getLocalDatacenter()).to(
@@ -128,22 +129,13 @@ public class CassandraAutoConfiguration {
 				.withFallback(ConfigFactory.defaultReference()).resolve();
 	}
 
-	private void mapQueryOptions(CassandraProperties properties, CassandraDriverOptions options) {
-		PropertyMapper map = PropertyMapper.get();
-		map.from(properties::getConsistencyLevel).whenNonNull()
-				.to(((consistency) -> options.add(DefaultDriverOption.REQUEST_CONSISTENCY, consistency)));
-		map.from(properties::getSerialConsistencyLevel).whenNonNull().to(
-				(serialConsistency) -> options.add(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY, serialConsistency));
-		map.from(properties::getPageSize)
-				.to((pageSize) -> options.add(DefaultDriverOption.REQUEST_PAGE_SIZE, pageSize));
-	}
-
-	private void mapSocketOptions(CassandraProperties properties, CassandraDriverOptions options) {
-		PropertyMapper map = PropertyMapper.get();
-		map.from(properties::getConnectTimeout).whenNonNull().asInt(Duration::toMillis)
-				.to((connectTimeout) -> options.add(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT, connectTimeout));
-		map.from(properties::getReadTimeout).whenNonNull().asInt(Duration::toMillis)
-				.to((readTimeout) -> options.add(DefaultDriverOption.REQUEST_TIMEOUT, readTimeout));
+	private void mapConnectionOptions(CassandraProperties properties, CassandraDriverOptions options) {
+		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		Connection connectionProperties = properties.getConnection();
+		map.from(connectionProperties::getConnectTimeout).asInt(Duration::toMillis)
+				.to((connectTimeout) -> options.add(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT, connectTimeout));
+		map.from(connectionProperties::getInitQueryTimeout).asInt(Duration::toMillis).to(
+				(initQueryTimeout) -> options.add(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT, initQueryTimeout));
 	}
 
 	private void mapPoolingOptions(CassandraProperties properties, CassandraDriverOptions options) {
@@ -155,9 +147,18 @@ public class CassandraAutoConfiguration {
 				.to((heartBeatInterval) -> options.add(DefaultDriverOption.HEARTBEAT_INTERVAL, heartBeatInterval));
 	}
 
-	private void mapThrottlingOptions(CassandraProperties properties, CassandraDriverOptions options) {
+	private void mapRequestOptions(CassandraProperties properties, CassandraDriverOptions options) {
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-		Throttler throttlerProperties = properties.getThrottler();
+		Request requestProperties = properties.getRequest();
+		map.from(requestProperties::getTimeout).asInt(Duration::toMillis)
+				.to(((timeout) -> options.add(DefaultDriverOption.REQUEST_TIMEOUT, timeout)));
+		map.from(requestProperties::getConsistency)
+				.to(((consistency) -> options.add(DefaultDriverOption.REQUEST_CONSISTENCY, consistency)));
+		map.from(requestProperties::getSerialConsistency).to(
+				(serialConsistency) -> options.add(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY, serialConsistency));
+		map.from(requestProperties::getPageSize)
+				.to((pageSize) -> options.add(DefaultDriverOption.REQUEST_PAGE_SIZE, pageSize));
+		Throttler throttlerProperties = requestProperties.getThrottler();
 		map.from(throttlerProperties::getType).as(ThrottlerType::type)
 				.to((type) -> options.add(DefaultDriverOption.REQUEST_THROTTLER_CLASS, type));
 		map.from(throttlerProperties::getMaxQueueSize)
@@ -168,7 +169,6 @@ public class CassandraAutoConfiguration {
 				.add(DefaultDriverOption.REQUEST_THROTTLER_MAX_REQUESTS_PER_SECOND, maxRequestsPerSecond));
 		map.from(throttlerProperties::getDrainInterval).asInt(Duration::toMillis).to(
 				(drainInterval) -> options.add(DefaultDriverOption.REQUEST_THROTTLER_DRAIN_INTERVAL, drainInterval));
-
 	}
 
 	private List<String> mapContactPoints(CassandraProperties properties) {

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java

@@ -82,31 +82,6 @@ public class CassandraProperties {
 	 */
 	private Compression compression = Compression.NONE;
 
-	/**
-	 * Queries consistency level.
-	 */
-	private DefaultConsistencyLevel consistencyLevel;
-
-	/**
-	 * Queries serial consistency level.
-	 */
-	private DefaultConsistencyLevel serialConsistencyLevel;
-
-	/**
-	 * Queries default page size.
-	 */
-	private int pageSize = 5000;
-
-	/**
-	 * Socket option: connection time out.
-	 */
-	private Duration connectTimeout;
-
-	/**
-	 * Socket option: read time out.
-	 */
-	private Duration readTimeout;
-
 	/**
 	 * Schema action to take at startup.
 	 */
@@ -117,15 +92,20 @@ public class CassandraProperties {
 	 */
 	private boolean ssl = false;
 
+	/**
+	 * Connection configuration.
+	 */
+	private final Connection connection = new Connection();
+
 	/**
 	 * Pool configuration.
 	 */
 	private final Pool pool = new Pool();
 
 	/**
-	 * Request throttling configuration.
+	 * Request configuration.
 	 */
-	private final Throttler throttler = new Throttler();
+	private final Request request = new Request();
 
 	public String getKeyspaceName() {
 		return this.keyspaceName;
@@ -198,55 +178,59 @@ public class CassandraProperties {
 		this.compression = compression;
 	}
 
+	@Deprecated
+	@DeprecatedConfigurationProperty(replacement = "spring.data.cassandra.request.consistency")
 	public DefaultConsistencyLevel getConsistencyLevel() {
-		return this.consistencyLevel;
-	}
-
-	public void setConsistencyLevel(DefaultConsistencyLevel consistency) {
-		this.consistencyLevel = consistency;
-	}
-
-	public DefaultConsistencyLevel getSerialConsistencyLevel() {
-		return this.serialConsistencyLevel;
-	}
-
-	public void setSerialConsistencyLevel(DefaultConsistencyLevel serialConsistency) {
-		this.serialConsistencyLevel = serialConsistency;
-	}
-
-	public int getPageSize() {
-		return this.pageSize;
-	}
-
-	public void setPageSize(int pageSize) {
-		this.pageSize = pageSize;
+		return getRequest().getConsistency();
 	}
 
 	@Deprecated
-	@DeprecatedConfigurationProperty(replacement = "spring.data.cassandra.page-size")
+	public void setConsistencyLevel(DefaultConsistencyLevel consistency) {
+		getRequest().setConsistency(consistency);
+	}
+
+	@Deprecated
+	@DeprecatedConfigurationProperty(replacement = "spring.data.cassandra.request.serial-consistency")
+	public DefaultConsistencyLevel getSerialConsistencyLevel() {
+		return getRequest().getSerialConsistency();
+	}
+
+	@Deprecated
+	public void setSerialConsistencyLevel(DefaultConsistencyLevel serialConsistency) {
+		getRequest().setSerialConsistency(serialConsistency);
+	}
+
+	@Deprecated
+	@DeprecatedConfigurationProperty(replacement = "spring.data.cassandra.request.page-size")
 	public int getFetchSize() {
-		return getPageSize();
+		return getRequest().getPageSize();
 	}
 
 	@Deprecated
 	public void setFetchSize(int fetchSize) {
-		setPageSize(fetchSize);
+		getRequest().setPageSize(fetchSize);
 	}
 
+	@Deprecated
+	@DeprecatedConfigurationProperty(replacement = "spring.data.cassandra.connection.init-query-timeout")
 	public Duration getConnectTimeout() {
-		return this.connectTimeout;
+		return getConnection().getInitQueryTimeout();
 	}
 
+	@Deprecated
 	public void setConnectTimeout(Duration connectTimeout) {
-		this.connectTimeout = connectTimeout;
+		getConnection().setInitQueryTimeout(connectTimeout);
 	}
 
+	@Deprecated
+	@DeprecatedConfigurationProperty(replacement = "spring.data.cassandra.request.timeout")
 	public Duration getReadTimeout() {
-		return this.readTimeout;
+		return getRequest().getTimeout();
 	}
 
+	@Deprecated
 	public void setReadTimeout(Duration readTimeout) {
-		this.readTimeout = readTimeout;
+		getRequest().setTimeout(readTimeout);
 	}
 
 	public boolean isSsl() {
@@ -265,12 +249,147 @@ public class CassandraProperties {
 		this.schemaAction = schemaAction;
 	}
 
+	public Connection getConnection() {
+		return this.connection;
+	}
+
 	public Pool getPool() {
 		return this.pool;
 	}
 
-	public Throttler getThrottler() {
-		return this.throttler;
+	public Request getRequest() {
+		return this.request;
+	}
+
+	public static class Connection {
+
+		/**
+		 * Timeout to use when establishing driver connections.
+		 */
+		private Duration connectTimeout = Duration.ofSeconds(5);
+
+		/**
+		 * Timeout to use for internal queries that run as part of the initialization
+		 * process, just after a connection is opened.
+		 */
+		private Duration initQueryTimeout = Duration.ofMillis(500);
+
+		public Duration getConnectTimeout() {
+			return this.connectTimeout;
+		}
+
+		public void setConnectTimeout(Duration connectTimeout) {
+			this.connectTimeout = connectTimeout;
+		}
+
+		public Duration getInitQueryTimeout() {
+			return this.initQueryTimeout;
+		}
+
+		public void setInitQueryTimeout(Duration initQueryTimeout) {
+			this.initQueryTimeout = initQueryTimeout;
+		}
+
+	}
+
+	public static class Request {
+
+		/**
+		 * How long the driver waits for a request to complete.
+		 */
+		private Duration timeout = Duration.ofSeconds(2);
+
+		/**
+		 * Queries consistency level.
+		 */
+		private DefaultConsistencyLevel consistency;
+
+		/**
+		 * Queries serial consistency level.
+		 */
+		private DefaultConsistencyLevel serialConsistency;
+
+		/**
+		 * How many rows will be retrieved simultaneously in a single network roundtrip.
+		 */
+		private int pageSize = 5000;
+
+		private final Throttler throttler = new Throttler();
+
+		public Duration getTimeout() {
+			return this.timeout;
+		}
+
+		public void setTimeout(Duration timeout) {
+			this.timeout = timeout;
+		}
+
+		public DefaultConsistencyLevel getConsistency() {
+			return this.consistency;
+		}
+
+		public void setConsistency(DefaultConsistencyLevel consistency) {
+			this.consistency = consistency;
+		}
+
+		public DefaultConsistencyLevel getSerialConsistency() {
+			return this.serialConsistency;
+		}
+
+		public void setSerialConsistency(DefaultConsistencyLevel serialConsistency) {
+			this.serialConsistency = serialConsistency;
+		}
+
+		public int getPageSize() {
+			return this.pageSize;
+		}
+
+		public void setPageSize(int pageSize) {
+			this.pageSize = pageSize;
+		}
+
+		public Throttler getThrottler() {
+			return this.throttler;
+		}
+
+	}
+
+	/**
+	 * Pool properties.
+	 */
+	public static class Pool {
+
+		/**
+		 * Idle timeout before an idle connection is removed. If a duration suffix is not
+		 * specified, seconds will be used.
+		 */
+		@DurationUnit(ChronoUnit.SECONDS)
+		private Duration idleTimeout = Duration.ofSeconds(120);
+
+		/**
+		 * Heartbeat interval after which a message is sent on an idle connection to make
+		 * sure it's still alive. If a duration suffix is not specified, seconds will be
+		 * used.
+		 */
+		@DurationUnit(ChronoUnit.SECONDS)
+		private Duration heartbeatInterval = Duration.ofSeconds(30);
+
+		public Duration getIdleTimeout() {
+			return this.idleTimeout;
+		}
+
+		public void setIdleTimeout(Duration idleTimeout) {
+			this.idleTimeout = idleTimeout;
+		}
+
+		public Duration getHeartbeatInterval() {
+			return this.heartbeatInterval;
+		}
+
+		public void setHeartbeatInterval(Duration heartbeatInterval) {
+			this.heartbeatInterval = heartbeatInterval;
+		}
+
 	}
 
 	public static class Throttler {
@@ -345,44 +464,6 @@ public class CassandraProperties {
 
 	}
 
-	/**
-	 * Pool properties.
-	 */
-	public static class Pool {
-
-		/**
-		 * Idle timeout before an idle connection is removed. If a duration suffix is not
-		 * specified, seconds will be used.
-		 */
-		@DurationUnit(ChronoUnit.SECONDS)
-		private Duration idleTimeout = Duration.ofSeconds(120);
-
-		/**
-		 * Heartbeat interval after which a message is sent on an idle connection to make
-		 * sure it's still alive. If a duration suffix is not specified, seconds will be
-		 * used.
-		 */
-		@DurationUnit(ChronoUnit.SECONDS)
-		private Duration heartbeatInterval = Duration.ofSeconds(30);
-
-		public Duration getIdleTimeout() {
-			return this.idleTimeout;
-		}
-
-		public void setIdleTimeout(Duration idleTimeout) {
-			this.idleTimeout = idleTimeout;
-		}
-
-		public Duration getHeartbeatInterval() {
-			return this.heartbeatInterval;
-		}
-
-		public void setHeartbeatInterval(Duration heartbeatInterval) {
-			this.heartbeatInterval = heartbeatInterval;
-		}
-
-	}
-
 	/**
 	 * Name of the algorithm used to compress protocol frames.
 	 */

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java

@@ -129,6 +129,17 @@ class CassandraAutoConfigurationTests {
 				});
 	}
 
+	@Test
+	void driverConfigLoaderCustomizeConnectionOptions() {
+		this.contextRunner.withPropertyValues("spring.data.cassandra.connection.connect-timeout=200ms",
+				"spring.data.cassandra.connection.init-query-timeout=10").run((context) -> {
+					DriverExecutionProfile config = context.getBean(DriverConfigLoader.class).getInitialConfig()
+							.getDefaultProfile();
+					assertThat(config.getInt(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT)).isEqualTo(200);
+					assertThat(config.getInt(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT)).isEqualTo(10);
+				});
+	}
+
 	@Test
 	void driverConfigLoaderCustomizePoolOptions() {
 		this.contextRunner.withPropertyValues("spring.data.cassandra.pool.idle-timeout=42",
@@ -140,6 +151,21 @@ class CassandraAutoConfigurationTests {
 				});
 	}
 
+	@Test
+	void driverConfigLoaderCustomizeRequestOptions() {
+		this.contextRunner.withPropertyValues("spring.data.cassandra.request.timeout=5s",
+				"spring.data.cassandra.request.consistency=two",
+				"spring.data.cassandra.request.serial-consistency=quorum", "spring.data.cassandra.request.page-size=42")
+				.run((context) -> {
+					DriverExecutionProfile config = context.getBean(DriverConfigLoader.class).getInitialConfig()
+							.getDefaultProfile();
+					assertThat(config.getInt(DefaultDriverOption.REQUEST_TIMEOUT)).isEqualTo(5000);
+					assertThat(config.getString(DefaultDriverOption.REQUEST_CONSISTENCY)).isEqualTo("TWO");
+					assertThat(config.getString(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY)).isEqualTo("QUORUM");
+					assertThat(config.getInt(DefaultDriverOption.REQUEST_PAGE_SIZE)).isEqualTo(42);
+				});
+	}
+
 	@Test
 	void driverConfigLoaderUsePassThroughLimitingRequestThrottlerByDefault() {
 		this.contextRunner.withPropertyValues().run((context) -> {
@@ -152,9 +178,9 @@ class CassandraAutoConfigurationTests {
 
 	@Test
 	void driverConfigLoaderCustomizeConcurrencyLimitingRequestThrottler() {
-		this.contextRunner.withPropertyValues("spring.data.cassandra.throttler.type=concurrency-limiting",
-				"spring.data.cassandra.throttler.max-concurrent-requests=62",
-				"spring.data.cassandra.throttler.max-queue-size=72").run((context) -> {
+		this.contextRunner.withPropertyValues("spring.data.cassandra.request.throttler.type=concurrency-limiting",
+				"spring.data.cassandra.request.throttler.max-concurrent-requests=62",
+				"spring.data.cassandra.request.throttler.max-queue-size=72").run((context) -> {
 					DriverExecutionProfile config = context.getBean(DriverConfigLoader.class).getInitialConfig()
 							.getDefaultProfile();
 					assertThat(config.getString(DefaultDriverOption.REQUEST_THROTTLER_CLASS))
@@ -167,10 +193,10 @@ class CassandraAutoConfigurationTests {
 
 	@Test
 	void driverConfigLoaderCustomizeRateLimitingRequestThrottler() {
-		this.contextRunner.withPropertyValues("spring.data.cassandra.throttler.type=rate-limiting",
-				"spring.data.cassandra.throttler.max-requests-per-second=62",
-				"spring.data.cassandra.throttler.max-queue-size=72",
-				"spring.data.cassandra.throttler.drain-interval=16ms").run((context) -> {
+		this.contextRunner.withPropertyValues("spring.data.cassandra.request.throttler.type=rate-limiting",
+				"spring.data.cassandra.request.throttler.max-requests-per-second=62",
+				"spring.data.cassandra.request.throttler.max-queue-size=72",
+				"spring.data.cassandra.request.throttler.drain-interval=16ms").run((context) -> {
 					DriverExecutionProfile config = context.getBean(DriverConfigLoader.class).getInitialConfig()
 							.getDefaultProfile();
 					assertThat(config.getString(DefaultDriverOption.REQUEST_THROTTLER_CLASS))