Upgrade to Kafka 3.4.0

Closes gh-34284

Author: wilkinsona

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.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 static class Streams {
 		 */
 		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 void setBootstrapServers(List<String> bootstrapServers) {
 			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 Map<String, Object> buildProperties() {
 			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"));

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java

@@ -261,7 +261,7 @@ void streamsProperties() {
 		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 @@ void streamsProperties() {
 							.asProperties();
 					assertThat((List<String>) 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 @@ void streamsProperties() {
 				});
 	}
 
+	@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() {

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",