root
/
spring-boot
mirror of https://github.com/spring-projects/spring-boot.git
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

Add config property for KafkaAdmin modifyTopicConfigs 

See gh-31679
This commit is contained in:
Michael Kreis 2022-07-12 08:17:27 +02:00 committed by Stephane Nicoll
parent 8e3aaf1cf4
commit 677c05a5b1
3 changed files with 23 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java
View File

@ -142,6 +142,7 @@ public class KafkaAutoConfiguration {
public KafkaAdmin kafkaAdmin() {
KafkaAdmin kafkaAdmin = new KafkaAdmin(this.properties.buildAdminProperties());
kafkaAdmin.setFatalIfBrokerNotAvailable(this.properties.getAdmin().isFailFast());
kafkaAdmin.setModifyTopicConfigs(this.properties.getAdmin().isModifyTopicConfigs());
return kafkaAdmin;
}

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

@ -647,6 +647,11 @@ public class KafkaProperties {
*/
private boolean failFast;
/**
* Whether to enable modification of existing topic configuration.
*/
private boolean modifyTopicConfigs;
public Ssl getSsl() {
return this.ssl;
}
@ -671,6 +676,14 @@ public class KafkaProperties {
this.failFast = failFast;
}
public boolean isModifyTopicConfigs() {
return this.modifyTopicConfigs;
}
public void setModifyTopicConfigs(boolean modifyTopicConfigs) {
this.modifyTopicConfigs = modifyTopicConfigs;
}
public Map<String, String> getProperties() {
return this.properties;
}

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

@ -209,15 +209,14 @@ class KafkaAutoConfigurationTests {
@Test
void adminProperties() {
this.contextRunner
.withPropertyValues("spring.kafka.clientId=cid", "spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
"spring.kafka.admin.fail-fast=true", "spring.kafka.admin.properties.fiz.buz=fix.fox",
"spring.kafka.admin.security.protocol=SSL", "spring.kafka.admin.ssl.key-password=p4",
"spring.kafka.admin.ssl.key-store-location=classpath:ksLocP",
"spring.kafka.admin.ssl.key-store-password=p5", "spring.kafka.admin.ssl.key-store-type=PKCS12",
"spring.kafka.admin.ssl.trust-store-location=classpath:tsLocP",
"spring.kafka.admin.ssl.trust-store-password=p6",
"spring.kafka.admin.ssl.trust-store-type=PKCS12", "spring.kafka.admin.ssl.protocol=TLSv1.2")
this.contextRunner.withPropertyValues("spring.kafka.clientId=cid",
"spring.kafka.properties.foo.bar.baz=qux.fiz.buz", "spring.kafka.admin.fail-fast=true",
"spring.kafka.admin.properties.fiz.buz=fix.fox", "spring.kafka.admin.security.protocol=SSL",
"spring.kafka.admin.ssl.key-password=p4", "spring.kafka.admin.ssl.key-store-location=classpath:ksLocP",
"spring.kafka.admin.ssl.key-store-password=p5", "spring.kafka.admin.ssl.key-store-type=PKCS12",
"spring.kafka.admin.ssl.trust-store-location=classpath:tsLocP",
"spring.kafka.admin.ssl.trust-store-password=p6", "spring.kafka.admin.ssl.trust-store-type=PKCS12",
"spring.kafka.admin.ssl.protocol=TLSv1.2", "spring.kafka.admin.modify-topic-configs=true")
.run((context) -> {
KafkaAdmin admin = context.getBean(KafkaAdmin.class);
Map<String, Object> configs = admin.getConfigurationProperties();
@ -239,6 +238,7 @@ class KafkaAutoConfigurationTests {
assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
assertThat(admin).hasFieldOrPropertyWithValue("fatalIfBrokerNotAvailable", true);
assertThat(admin).hasFieldOrPropertyWithValue("modifyTopicConfigs", true);
});
}