correct many defects in DefaultKafkaProducerFactory#updateConfigs()

Author: NathanQingyangXu

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -23,6 +23,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
@@ -110,6 +111,7 @@
  * @author Artem Bilan
  * @author Chris Gilbert
  * @author Thomas Strauß
+ * @author Nathan Xu
  */
 public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 		implements ProducerFactory<K, V>, ApplicationContextAware,
@@ -660,23 +662,31 @@ public boolean removePostProcessor(ProducerPostProcessor<K, V> postProcessor) {
 
 	@Override
 	public void updateConfigs(Map<String, Object> updates) {
-		updates.entrySet().forEach(entry -> {
-			if (entry.getKey().equals(ProducerConfig.TRANSACTIONAL_ID_CONFIG)) {
-				Assert.isTrue(entry.getValue() instanceof String, () -> "'" + ProducerConfig.TRANSACTIONAL_ID_CONFIG
-						+ "' must be a String, not a " + entry.getClass().getName());
-				Assert.isTrue(this.transactionIdPrefix != null
-								? entry.getValue() != null
-								: entry.getValue() == null,
-								"Cannot change transactional capability");
-				this.transactionIdPrefix = (String) entry.getValue();
+		updates.forEach((key, value) -> {
+			if (key == null) { // ConcurrentHashMap doesn't accept null key
+				return;
 			}
-			else if (entry.getKey().equals(ProducerConfig.CLIENT_ID_CONFIG)) {
-				Assert.isTrue(entry.getValue() instanceof String, () -> "'" + ProducerConfig.CLIENT_ID_CONFIG
-						+ "' must be a String, not a " + entry.getClass().getName());
-				this.clientIdPrefix = (String) entry.getValue();
+			if (key.equals(ProducerConfig.TRANSACTIONAL_ID_CONFIG)) {
+				Assert.isTrue(
+						value == null || value instanceof String,
+						() -> "'" + ProducerConfig.TRANSACTIONAL_ID_CONFIG
+								+ "' must be null or a String, not a " + Objects.requireNonNull(value)
+								.getClass()
+								.getName()
+				);
+				Assert.isTrue(
+						(this.transactionIdPrefix != null) == (value != null),
+						"Cannot change transactional capability"
+				);
+				this.transactionIdPrefix = (String) value;
 			}
-			else {
-				this.configs.put(entry.getKey(), entry.getValue());
+			else if (key.equals(ProducerConfig.CLIENT_ID_CONFIG)) {
+				Assert.isTrue(value == null || value instanceof String, () -> "'" + ProducerConfig.CLIENT_ID_CONFIG
+						+ "' must be null or a String, not a " + Objects.requireNonNull(value).getClass().getName());
+				this.clientIdPrefix = (String) value;
+			}
+			else if (value != null) { // ConcurrentHashMap doesn't accept null value
+				this.configs.put(key, value);
 			}
 		});
 	}

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.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.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.mockito.ArgumentMatchers.any;
@@ -545,6 +546,7 @@ void configUpdates() {
 		assertThat(KafkaTestUtils.getPropertyValue(pf1, "transactionIdPrefix")).isEqualTo("tx2-");
 		configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, null);
 		assertThatIllegalArgumentException().isThrownBy(() -> pf1.updateConfigs(configs));
+		assertThatCode(() -> pf1.updateConfigs(Collections.singletonMap(null, null))).doesNotThrowAnyException();
 	}
 
 	@SuppressWarnings("unchecked")