spring-projectsGH-2826: Fix KafkaAdmin NPE With Bad Config Name

Resolves spring-projects#2832 **cherry-pick to 3.0.x, 2.9.x**
garyrussell · Oct 9, 2023 · cdc957e · cdc957e
1 parent c98e4cd
commit cdc957e
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 deletions.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java
@@ -415,6 +415,10 @@ private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
 				if (topicOptional.isPresent() && topicOptional.get().configs() != null) {
 					for (Map.Entry<String, String> desiredConfigParameter : topicOptional.get().configs().entrySet()) {
 						ConfigEntry actualConfigParameter = topicConfig.getValue().get(desiredConfigParameter.getKey());
+						if (actualConfigParameter == null) {
+							throw new IllegalStateException("Topic property '" + desiredConfigParameter.getKey()
+									+ "' does not exist");
+						}
 						if (!desiredConfigParameter.getValue().equals(actualConfigParameter.value())) {
 							configMismatchesEntries.add(actualConfigParameter);
 						}

diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java
@@ -17,6 +17,7 @@
 package org.springframework.kafka.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.awaitility.Awaitility.await;
 
 import java.lang.reflect.Method;
@@ -176,6 +177,15 @@ public void testAddTopicsAndAddPartitions() throws Exception {
 					&& configResourceConfigMap.get(new ConfigResource(Type.TOPIC, "noConfigAddLater"))
 					.get("retention.ms").value().equals("1000");
 		});
+
+		assertThatIllegalStateException().isThrownBy(() -> this.admin.createOrModifyTopics(mismatchconfig,
+				TopicBuilder.name("noConfigAddLater")
+						.partitions(2)
+						.replicas(1)
+						.config("no.such.config.key", "1000")
+						.build()))
+				.withMessageContaining("no.such.config.key");
+
 	}
 
 	@Test