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

Add 'spring.kafka.consumer.max-poll-interval' configuration property

Browse Source 
See gh-44901

Signed-off-by: Yanming Zhou <zhouyanming@gmail.com>
This commit is contained in:
Yanming Zhou
2025-03-26 15:05:00 +08:00
committed by Moritz Halbritter
parent 617244b473
commit 4e1fa66420
2 changed files with 23 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -59,6 +59,7 @@ import org.springframework.util.unit.DataSize;
* @author Tomaz Fernandes
* @author Andy Wilkinson
* @author Scott Frederick
* @author Yanming Zhou
* @since 1.5.0
*/
@ConfigurationProperties("spring.kafka")
@@ -337,6 +338,12 @@ public class KafkaProperties {
*/
private Integer maxPollRecords;
/**
* Maximum delay between invocations of poll() when using consumer group
* management.
*/
private Duration maxPollInterval;
/**
* Additional consumer-specific properties used to configure the client.
*/
@@ -454,6 +461,14 @@ public class KafkaProperties {
this.maxPollRecords = maxPollRecords;
}
public Duration getMaxPollInterval() {
return this.maxPollInterval;
}
public void setMaxPollInterval(Duration maxPollInterval) {
this.maxPollInterval = maxPollInterval;
}
public Map<String, String> getProperties() {
return this.properties;
}
@@ -483,6 +498,9 @@ public class KafkaProperties {
map.from(this::getKeyDeserializer).to(properties.in(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
map.from(this::getValueDeserializer).to(properties.in(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
map.from(this::getMaxPollRecords).to(properties.in(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
map.from(this::getMaxPollInterval)
.asInt(Duration::toMillis)
.to(properties.in(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG));
return properties.with(this.ssl, this.security, this.properties, sslBundles);
}

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

@@ -115,6 +115,7 @@ import static org.mockito.Mockito.never;
* @author Andy Wilkinson
* @author Phillip Webb
* @author Scott Frederick
* @author Yanming Zhou
*/
class KafkaAutoConfigurationTests {
@@ -132,8 +133,9 @@ class KafkaAutoConfigurationTests {
"spring.kafka.ssl.key-store-type=PKCS12", "spring.kafka.ssl.trust-store-location=classpath:tsLoc",
"spring.kafka.ssl.trust-store-password=p3", "spring.kafka.ssl.trust-store-type=PKCS12",
"spring.kafka.ssl.protocol=TLSv1.2", "spring.kafka.consumer.auto-commit-interval=123",
"spring.kafka.consumer.max-poll-records=42", "spring.kafka.consumer.auto-offset-reset=earliest",
"spring.kafka.consumer.client-id=ccid", // test override common
"spring.kafka.consumer.max-poll-records=42", "spring.kafka.consumer.max-poll-interval=30s",
"spring.kafka.consumer.auto-offset-reset=earliest", "spring.kafka.consumer.client-id=ccid",
// test override common
"spring.kafka.consumer.enable-auto-commit=false", "spring.kafka.consumer.fetch-max-wait=456",
"spring.kafka.consumer.properties.fiz.buz=fix.fox", "spring.kafka.consumer.fetch-min-size=1KB",
"spring.kafka.consumer.group-id=bar", "spring.kafka.consumer.heartbeat-interval=234",
@@ -172,6 +174,7 @@ class KafkaAutoConfigurationTests {
assertThat(configs).containsEntry(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
IntegerDeserializer.class);
assertThat(configs).containsEntry(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 42);
assertThat(configs).containsEntry(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30000);
assertThat(configs).containsEntry("foo", "bar");
assertThat(configs).containsEntry("baz", "qux");
assertThat(configs).containsEntry("foo.bar.baz", "qux.fiz.buz");