From a73973f5f54ca97515b1d783c28ebeaaf6496d8e Mon Sep 17 00:00:00 2001 From: Andy Wilkinson Date: Mon, 20 Feb 2023 16:30:26 +0000 Subject: [PATCH] Upgrade to Kafka 3.4.0 Closes gh-34284 --- .../autoconfigure/kafka/KafkaProperties.java | 19 +++++++++++++++++++ .../kafka/KafkaAutoConfigurationTests.java | 17 +++++++++++++++-- .../spring-boot-dependencies/build.gradle | 2 +- 3 files changed, 35 insertions(+), 3 deletions(-) diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java index bfa7cf1631..de10b115bf 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java @@ -34,6 +34,7 @@ import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.DeprecatedConfigurationProperty; import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException; import org.springframework.boot.convert.DurationUnit; @@ -769,6 +770,11 @@ public class KafkaProperties { */ private DataSize cacheMaxSizeBuffering; + /** + * Maximum size of the in-memory state store cache across all threads. + */ + private DataSize stateStoreCacheMaxSize; + /** * ID to pass to the server when making requests. Used for server-side logging. */ @@ -826,14 +832,25 @@ public class KafkaProperties { this.bootstrapServers = bootstrapServers; } + @DeprecatedConfigurationProperty(replacement = "spring.kafka.streams.state-store-cache-max-size") + @Deprecated(since = "3.1.0", forRemoval = true) public DataSize getCacheMaxSizeBuffering() { return this.cacheMaxSizeBuffering; } + @Deprecated(since = "3.1.0", forRemoval = true) public void setCacheMaxSizeBuffering(DataSize cacheMaxSizeBuffering) { this.cacheMaxSizeBuffering = cacheMaxSizeBuffering; } + public DataSize getStateStoreCacheMaxSize() { + return this.stateStoreCacheMaxSize; + } + + public void setStateStoreCacheMaxSize(DataSize stateStoreCacheMaxSize) { + this.stateStoreCacheMaxSize = stateStoreCacheMaxSize; + } + public String getClientId() { return this.clientId; } @@ -869,6 +886,8 @@ public class KafkaProperties { map.from(this::getBootstrapServers).to(properties.in(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)); map.from(this::getCacheMaxSizeBuffering).asInt(DataSize::toBytes) .to(properties.in("cache.max.bytes.buffering")); + map.from(this::getStateStoreCacheMaxSize).asInt(DataSize::toBytes) + .to(properties.in("statestore.cache.max.bytes")); map.from(this::getClientId).to(properties.in(CommonClientConfigs.CLIENT_ID_CONFIG)); map.from(this::getReplicationFactor).to(properties.in("replication.factor")); map.from(this::getStateDir).to(properties.in("state.dir")); diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java index ccbb9e1040..3ad6e46308 100644 --- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java +++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java @@ -261,7 +261,7 @@ class KafkaAutoConfigurationTests { this.contextRunner.withUserConfiguration(EnableKafkaStreamsConfiguration.class).withPropertyValues( "spring.kafka.client-id=cid", "spring.kafka.bootstrap-servers=localhost:9092,localhost:9093", "spring.application.name=appName", "spring.kafka.properties.foo.bar.baz=qux.fiz.buz", - "spring.kafka.streams.auto-startup=false", "spring.kafka.streams.cache-max-size-buffering=1KB", + "spring.kafka.streams.auto-startup=false", "spring.kafka.streams.state-store-cache-max-size=1KB", "spring.kafka.streams.client-id=override", "spring.kafka.streams.properties.fiz.buz=fix.fox", "spring.kafka.streams.replication-factor=2", "spring.kafka.streams.state-dir=/tmp/state", "spring.kafka.streams.security.protocol=SSL", "spring.kafka.streams.ssl.key-password=p7", @@ -276,7 +276,7 @@ class KafkaAutoConfigurationTests { .asProperties(); assertThat((List) configs.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) .containsExactly("localhost:9092", "localhost:9093"); - assertThat(configs).containsEntry(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 1024); + assertThat(configs).containsEntry(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 1024); assertThat(configs).containsEntry(StreamsConfig.CLIENT_ID_CONFIG, "override"); assertThat(configs).containsEntry(StreamsConfig.REPLICATION_FACTOR_CONFIG, 2); assertThat(configs).containsEntry(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); @@ -299,6 +299,19 @@ class KafkaAutoConfigurationTests { }); } + @SuppressWarnings("deprecation") + @Deprecated(since = "3.1.0", forRemoval = true) + void streamsCacheMaxSizeBuffering() { + this.contextRunner.withUserConfiguration(EnableKafkaStreamsConfiguration.class) + .withPropertyValues("spring.kafka.streams.cache-max-size-buffering=1KB").run((context) -> { + Properties configs = context + .getBean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME, + KafkaStreamsConfiguration.class) + .asProperties(); + assertThat(configs).containsEntry(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 1024); + }); + } + @SuppressWarnings("unchecked") @Test void streamsApplicationIdUsesMainApplicationNameByDefault() { diff --git a/spring-boot-project/spring-boot-dependencies/build.gradle b/spring-boot-project/spring-boot-dependencies/build.gradle index d6128f39ed..b3ff0ac522 100644 --- a/spring-boot-project/spring-boot-dependencies/build.gradle +++ b/spring-boot-project/spring-boot-dependencies/build.gradle @@ -700,7 +700,7 @@ bom { ] } } - library("Kafka", "3.3.2") { + library("Kafka", "3.4.0") { group("org.apache.kafka") { modules = [ "connect",