From e26f3e2a65cba16cc0b47b997828456cf4ec3e66 Mon Sep 17 00:00:00 2001 From: Stephane Nicoll Date: Tue, 21 Apr 2020 17:29:38 +0200 Subject: [PATCH] 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 --- .../cassandra/CassandraAutoConfiguration.java | 44 +-- .../cassandra/CassandraProperties.java | 271 ++++++++++++------ .../CassandraAutoConfigurationTests.java | 40 ++- 3 files changed, 231 insertions(+), 124 deletions(-) diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java index 53d6d3cef7..2e370765b8 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfiguration.java +++ b/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 mapContactPoints(CassandraProperties properties) { diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java index eb1e199c79..f4f2ce0859 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/cassandra/CassandraProperties.java +++ b/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. */ diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java index 61e7205288..e2e041b272 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/cassandra/CassandraAutoConfigurationTests.java +++ b/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))