fix legacy format issue in DefaultKafkaProducerFactory

Author: NathanQingyangXu

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

@@ -165,6 +165,7 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	/**
 	 * Construct a factory with the provided configuration.
+	 *
 	 * @param configs the configuration.
 	 */
 	public DefaultKafkaProducerFactory(Map<String, Object> configs) {
@@ -178,11 +179,13 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs) {
 	 * This config is going to be overridden with a suffix for target {@link Producer} instance.
 	 * The serializers' {@code configure()} methods will be called with the
 	 * configuration map.
+	 *
 	 * @param configs the configuration.
 	 * @param keySerializer the key {@link Serializer}.
 	 * @param valueSerializer the value {@link Serializer}.
 	 */
-	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+	public DefaultKafkaProducerFactory(
+			Map<String, Object> configs,
 			@Nullable Serializer<K> keySerializer,
 			@Nullable Serializer<V> valueSerializer) {
 
@@ -196,14 +199,17 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 	 * be overridden with a suffix for target {@link Producer} instance. The serializers'
 	 * {@code configure()} methods will be called with the configuration map unless
 	 * {@code configureSerializers} is false..
+	 *
 	 * @param configs the configuration.
 	 * @param keySerializer the key {@link Serializer}.
 	 * @param valueSerializer the value {@link Serializer}.
 	 * @param configureSerializers set to false if serializers are already fully
 	 * configured.
+	 *
 	 * @since 2.8.7
 	 */
-	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+	public DefaultKafkaProducerFactory(
+			Map<String, Object> configs,
 			@Nullable Serializer<K> keySerializer,
 			@Nullable Serializer<V> valueSerializer, boolean configureSerializers) {
 
@@ -217,12 +223,15 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 	 * be overridden with a suffix for target {@link Producer} instance. When the
 	 * suppliers are invoked to get an instance, the serializers' {@code configure()}
 	 * methods will be called with the configuration map.
+	 *
 	 * @param configs the configuration.
 	 * @param keySerializerSupplier the key {@link Serializer} supplier function.
 	 * @param valueSerializerSupplier the value {@link Serializer} supplier function.
+	 *
 	 * @since 2.3
 	 */
-	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+	public DefaultKafkaProducerFactory(
+			Map<String, Object> configs,
 			@Nullable Supplier<Serializer<K>> keySerializerSupplier,
 			@Nullable Supplier<Serializer<V>> valueSerializerSupplier) {
 
@@ -244,7 +253,8 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 	 * configured.
 	 * @since 2.8.7
 	 */
-	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+	public DefaultKafkaProducerFactory(
+			Map<String, Object> configs,
 			@Nullable Supplier<Serializer<K>> keySerializerSupplier,
 			@Nullable Supplier<Serializer<V>> valueSerializerSupplier, boolean configureSerializers) {
 
@@ -308,7 +318,9 @@ public void setBeanName(String name) {
 	 * Set a key serializer. The serializer will be configured using the producer
 	 * configuration, unless {@link #setConfigureSerializers(boolean)
 	 * configureSerializers} is false.
+	 *
 	 * @param keySerializer the key serializer.
+	 *
 	 * @see #setConfigureSerializers(boolean)
 	 */
 	public void setKeySerializer(@Nullable Serializer<K> keySerializer) {
@@ -319,7 +331,9 @@ public void setKeySerializer(@Nullable Serializer<K> keySerializer) {
 	 * Set a value serializer. The serializer will be configured using the producer
 	 * configuration, unless {@link #setConfigureSerializers(boolean)
 	 * configureSerializers} is false.
+	 *
 	 * @param valueSerializer the value serializer.
+	 *
 	 * @see #setConfigureSerializers(boolean)
 	 */
 	public void setValueSerializer(@Nullable Serializer<V> valueSerializer) {
@@ -330,6 +344,7 @@ public void setValueSerializer(@Nullable Serializer<V> valueSerializer) {
 	 * Set a supplier to supply instances of the key serializer. The serializer will be
 	 * configured using the producer configuration, unless
 	 * {@link #setConfigureSerializers(boolean) configureSerializers} is false.
+	 *
 	 * @param keySerializerSupplier the supplier.
 	 * @since 2.8
 	 * @see #setConfigureSerializers(boolean)
@@ -340,6 +355,7 @@ public void setKeySerializerSupplier(Supplier<Serializer<K>> keySerializerSuppli
 
 	/**
 	 * Set a supplier to supply instances of the value serializer.
+	 *
 	 * @param valueSerializerSupplier the supplier. The serializer will be configured
 	 * using the producer configuration, unless {@link #setConfigureSerializers(boolean)
 	 * configureSerializers} is false.
@@ -354,12 +370,14 @@ public void setValueSerializerSupplier(Supplier<Serializer<V>> valueSerializerSu
 	 * If true (default), programmatically provided serializers (via constructor or
 	 * setters) will be configured using the producer configuration. Set to false if the
 	 * serializers are already fully configured.
+	 *
 	 * @return true to configure.
 	 * @since 2.8.7
 	 * @see #setKeySerializer(Serializer)
 	 * @see #setKeySerializerSupplier(Supplier)
 	 * @see #setValueSerializer(Serializer)
 	 * @see #setValueSerializerSupplier(Supplier)
+
 	 */
 	public boolean isConfigureSerializers() {
 		return this.configureSerializers;
@@ -369,6 +387,7 @@ public boolean isConfigureSerializers() {
 	 * Set to false (default true) to prevent programmatically provided serializers (via
 	 * constructor or setters) from being configured using the producer configuration,
 	 * e.g. if the serializers are already fully configured.
+	 *
 	 * @param configureSerializers false to not configure.
 	 * @since 2.8.7
 	 * @see #setKeySerializer(Serializer)
@@ -385,6 +404,7 @@ public void setConfigureSerializers(boolean configureSerializers) {
 	 * closing the producer itself (when {@link #reset()}, {@link #destroy()
 	 * #closeProducerFor(String)}, or {@link #closeThreadBoundProducer()} are invoked).
 	 * Specified in seconds; default {@link #DEFAULT_PHYSICAL_CLOSE_TIMEOUT}.
+
 	 * @param physicalCloseTimeout the timeout in seconds.
 	 * @since 1.0.7
 	 */
@@ -425,6 +445,7 @@ public final void setTransactionIdPrefix(String transactionIdPrefix) {
 	 * all clients. Clients <b>must</b> call {@link #closeThreadBoundProducer()} to
 	 * physically close the producer when it is no longer needed. These producers will not
 	 * be closed by {@link #destroy()} or {@link #reset()}.
+	 *
 	 * @param producerPerThread true for a producer per thread.
 	 * @since 2.3
 	 * @see #closeThreadBoundProducer()
@@ -526,19 +547,23 @@ public int getPhase() {
 	 * you want to change the ID config, add a new
 	 * {@link org.apache.kafka.clients.producer.ProducerConfig#TRANSACTIONAL_ID_CONFIG}
 	 * key to the override config.</p>
+	 *
 	 * @param overrideProperties the properties to be applied to the new factory
+	 *
 	 * @return {@link org.springframework.kafka.core.DefaultKafkaProducerFactory} with
-	 *  properties applied
+	 * properties applied
 	 */
 	@Override
 	public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
 		Map<String, Object> producerProperties = new HashMap<>(getConfigurationProperties());
 		producerProperties.putAll(overrideProperties);
 		producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties);
-		DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(producerProperties,
-						getKeySerializerSupplier(),
-						getValueSerializerSupplier(),
-						isConfigureSerializers());
+		DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(
+				producerProperties,
+				getKeySerializerSupplier(),
+				getValueSerializerSupplier(),
+				isConfigureSerializers()
+		);
 		newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds());
 		newFactory.setProducerPerThread(isProducerPerThread());
 		for (ProducerPostProcessor<K, V> templatePostProcessor : getPostProcessors()) {
@@ -559,7 +584,9 @@ public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> o
 	 * new factory, the transactionId has to be reinserted prior use.
 	 * The incoming properties are checked for a transactionId key. If none is
 	 * there, the one existing in the factory is added.
+	 *
 	 * @param producerProperties the properties to be used for the new factory
+	 *
 	 * @return the producerProperties or a copy with the transaction ID set
 	 */
 	private Map<String, Object> ensureExistingTransactionIdPrefixInProperties(Map<String, Object> producerProperties) {
@@ -576,7 +603,9 @@ private Map<String, Object> ensureExistingTransactionIdPrefixInProperties(Map<St
 
 	/**
 	 * Add a listener.
+	 *
 	 * @param listener the listener.
+	 *
 	 * @since 2.5
 	 */
 	@Override
@@ -587,8 +616,10 @@ public void addListener(Listener<K, V> listener) {
 
 	/**
 	 * Add a listener at a specific index.
+	 *
 	 * @param index the index (list position).
 	 * @param listener the listener.
+	 *
 	 * @since 2.5
 	 */
 	@Override
@@ -604,8 +635,11 @@ public void addListener(int index, Listener<K, V> listener) {
 
 	/**
 	 * Remove a listener.
+	 *
 	 * @param listener the listener.
+	 *
 	 * @return true if removed.
+	 *
 	 * @since 2.5
 	 */
 	@Override
@@ -633,7 +667,7 @@ public void updateConfigs(Map<String, Object> updates) {
 				Assert.isTrue(this.transactionIdPrefix != null
 								? entry.getValue() != null
 								: entry.getValue() == null,
-						"Cannot change transactional capability");
+								"Cannot change transactional capability");
 				this.transactionIdPrefix = (String) entry.getValue();
 			}
 			else if (entry.getKey().equals(ProducerConfig.CLIENT_ID_CONFIG)) {
@@ -723,6 +757,7 @@ public void onApplicationEvent(ContextStoppedEvent event) {
 	/**
 	 * Close the {@link Producer}(s) and clear the cache of transactional
 	 * {@link Producer}(s).
+	 *
 	 * @since 2.2
 	 */
 	@Override
@@ -769,7 +804,8 @@ private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 			}
 			if (this.producer == null) {
 				this.producer = new CloseSafeProducer<>(createKafkaProducer(), this::removeProducer,
-						this.physicalCloseTimeout, this.beanName, this.epoch.get());
+														this.physicalCloseTimeout, this.beanName, this.epoch.get()
+				);
 				this.listeners.forEach(listener -> listener.producerAdded(this.producer.clientId, this.producer));
 			}
 			return this.producer;
@@ -787,7 +823,8 @@ private Producer<K, V> getOrCreateThreadBoundProducer() {
 		}
 		if (tlProducer == null) {
 			tlProducer = new CloseSafeProducer<>(createKafkaProducer(), this::removeProducer,
-					this.physicalCloseTimeout, this.beanName, this.epoch.get());
+												this.physicalCloseTimeout, this.beanName, this.epoch.get()
+			);
 			for (Listener<K, V> listener : this.listeners) {
 				listener.producerAdded(tlProducer.clientId, tlProducer);
 			}
@@ -798,6 +835,7 @@ private Producer<K, V> getOrCreateThreadBoundProducer() {
 
 	/**
 	 * Subclasses must return a raw producer which will be wrapped in a {@link CloseSafeProducer}.
+	 *
 	 * @return the producer.
 	 */
 	protected Producer<K, V> createKafkaProducer() {
@@ -806,9 +844,12 @@ protected Producer<K, V> createKafkaProducer() {
 
 	/**
 	 * Remove the single shared producer and a thread-bound instance if present.
+	 *
 	 * @param producerToRemove the producer.
 	 * @param timeout the close timeout.
+	 *
 	 * @return true if the producer was closed.
+	 *
 	 * @since 2.2.13
 	 */
 	protected final boolean removeProducer(CloseSafeProducer<K, V> producerToRemove, Duration timeout) {
@@ -818,7 +859,9 @@ protected final boolean removeProducer(CloseSafeProducer<K, V> producerToRemove,
 	/**
 	 * Subclasses must return a producer from the {@link #getCache()} or a
 	 * new raw producer wrapped in a {@link CloseSafeProducer}.
+	 *
 	 * @return the producer - cannot be null.
+	 *
 	 * @since 1.3
 	 */
 	protected Producer<K, V> createTransactionalProducer() {
@@ -838,7 +881,11 @@ protected Producer<K, V> createTransactionalProducer(String txIdPrefix) {
 			}
 		}
 		if (cachedProducer == null) {
-			return doCreateTxProducer(txIdPrefix, "" + this.transactionIdSuffix.getAndIncrement(), this::cacheReturner);
+			return doCreateTxProducer(
+					txIdPrefix,
+					"" + this.transactionIdSuffix.getAndIncrement(),
+					this::cacheReturner
+			);
 		}
 		else {
 			return cachedProducer;
@@ -876,7 +923,8 @@ boolean cacheReturner(CloseSafeProducer<K, V> producerToRemove, Duration timeout
 		}
 	}
 
-	private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
+	private CloseSafeProducer<K, V> doCreateTxProducer(
+			String prefix, String suffix,
 			BiPredicate<CloseSafeProducer<K, V>, Duration> remover) {
 		Producer<K, V> newProducer = createRawProducer(getTxProducerConfigs(prefix + suffix));
 		try {
@@ -895,7 +943,8 @@ private CloseSafeProducer<K, V> doCreateTxProducer(String prefix, String suffix,
 		}
 		CloseSafeProducer<K, V> closeSafeProducer =
 				new CloseSafeProducer<>(newProducer, remover, prefix, this.physicalCloseTimeout, this.beanName,
-						this.epoch.get());
+										this.epoch.get()
+				);
 		this.listeners.forEach(listener -> listener.producerAdded(closeSafeProducer.clientId, closeSafeProducer));
 		return closeSafeProducer;
 	}
@@ -939,6 +988,7 @@ public void closeThreadBoundProducer() {
 
 	/**
 	 * Return the configuration of a producer.
+	 *
 	 * @return the configuration of a producer.
 	 * @since 2.8.3
 	 * @see #createKafkaProducer()
@@ -947,15 +997,19 @@ protected Map<String, Object> getProducerConfigs() {
 		final Map<String, Object> newProducerConfigs = new HashMap<>(this.configs);
 		checkBootstrap(newProducerConfigs);
 		if (this.clientIdPrefix != null) {
-			newProducerConfigs.put(ProducerConfig.CLIENT_ID_CONFIG,
-					this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet());
+			newProducerConfigs.put(
+					ProducerConfig.CLIENT_ID_CONFIG,
+					this.clientIdPrefix + "-" + this.clientIdCounter.incrementAndGet()
+			);
 		}
 		return newProducerConfigs;
 	}
 
 	/**
 	 * Return the configuration of a transactional producer.
+	 *
 	 * @param transactionId the transactionId.
+	 *
 	 * @return the configuration of a transactional producer.
 	 * @since 2.8.3
 	 * @see #doCreateTxProducer(String, String, BiPredicate)
@@ -971,7 +1025,6 @@ protected Map<String, Object> getTxProducerConfigs(String transactionId) {
 	 *
 	 * @param <K> the key type.
 	 * @param <V> the value type.
-	 *
 	 */
 	protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
@@ -995,14 +1048,16 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		volatile boolean closed; // NOSONAR
 
-		CloseSafeProducer(Producer<K, V> delegate,
+		CloseSafeProducer(
+				Producer<K, V> delegate,
 				BiPredicate<CloseSafeProducer<K, V>, Duration> removeConsumerProducer, Duration closeTimeout,
 				String factoryName, int epoch) {
 
 			this(delegate, removeConsumerProducer, null, closeTimeout, factoryName, epoch);
 		}
 
-		CloseSafeProducer(Producer<K, V> delegate,
+		CloseSafeProducer(
+				Producer<K, V> delegate,
 				BiPredicate<CloseSafeProducer<K, V>, Duration> removeProducer, @Nullable String txIdPrefix,
 				Duration closeTimeout, String factoryName, int epoch) {
 
@@ -1097,7 +1152,8 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
 		}
 
 		@Override
-		public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
+		public void sendOffsetsToTransaction(
+				Map<TopicPartition, OffsetAndMetadata> offsets,
 				ConsumerGroupMetadata groupMetadata) throws ProducerFencedException {
 
 			LOGGER.trace(() -> toString() + " sendOffsetsToTransaction(" + offsets + ", " + groupMetadata + ")");

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