Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade to Kafka 3.4.0

Browse Source 
Closes gh-34284
This commit is contained in:
Andy Wilkinson
2023-02-20 16:30:26 +00:00
parent c0b5bfeb40
commit a73973f5f5
3 changed files with 35 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
View File

@@ -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"));

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

@@ -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<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 @@ 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() {

2
spring-boot-project/spring-boot-dependencies/build.gradle
View File

@@ -700,7 +700,7 @@ bom {
]
}
}
library("Kafka", "3.3.2") {
library("Kafka", "3.4.0") {
group("org.apache.kafka") {
modules = [
"connect",