Fix new Sonar smells

Author: artembilan

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java

@@ -21,8 +21,6 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
-import org.apache.kafka.clients.producer.ProducerRecord;
-
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.dsl.ComponentsRegistration;
@@ -239,8 +237,8 @@ public S timestampExpression(Expression timestampExpression) {
 	}
 
 	/**
-	 * Configure a SpEL expression to determine whether or not to flush the producer after
-	 * a send. By default the producer is flushed if a header {@code kafka_flush} has a
+	 * Configure a SpEL expression to determine whether to flush the producer after
+	 * a send. By default, the producer is flushed if a header {@code kafka_flush} has a
 	 * value {@link Boolean#TRUE}.
 	 * @param flushExpression the timestamp expression to use.
 	 * @return the spec.
@@ -251,8 +249,8 @@ public S flushExpression(String flushExpression) {
 
 	/**
 	 * Configure a {@link Function} that will be invoked at runtime to determine whether
-	 * or not to flush the producer after a send. By default the producer is flushed if a
-	 * header {@code kafka_flush} has a value {@link Boolean#TRUE}. Typically used with a
+	 * to flush the producer after send. By default, the producer is flushed if a
+	 * header {@code kafka_flush} has a value {@link Boolean#TRUE}. Typically, used with a
 	 * Java 8 Lambda expression:
 	 * <pre class="code">
 	 * {@code
@@ -268,8 +266,8 @@ public <P> S flush(Function<Message<P>, Boolean> flushFunction) {
 	}
 
 	/**
-	 * Configure an {@link Expression} to determine whether or not to flush the producer
-	 * after a send. By default the producer is flushed if a header {@code kafka_flush}
+	 * Configure an {@link Expression} to determine whether to flush the producer
+	 * after a send. By default, the producer is flushed if a header {@code kafka_flush}
 	 * has a value {@link Boolean#TRUE}.
 	 * @param flushExpression the timestamp expression to use.
 	 * @return the spec.
@@ -293,7 +291,7 @@ public S sync(boolean sync) {
 
 	/**
 	 * Specify a timeout in milliseconds how long {@link KafkaProducerMessageHandler}
-	 * should wait wait for send operation results. Defaults to 10 seconds.
+	 * should wait for send operation results. Defaults to 10 seconds.
 	 * @param sendTimeout the timeout to wait for result fo send operation.
 	 * @return the spec.
 	 */
@@ -379,8 +377,8 @@ public S futuresChannel(String futuresChannel) {
 	}
 
 	/**
-	 * Set a {@link ProducerRecordCreator} to create the {@link ProducerRecord}. Ignored
-	 * if {@link #useTemplateConverter(boolean) useTemplateConverter} is true.
+	 * Set a {@link ProducerRecordCreator} to create the {@link org.apache.kafka.clients.producer.ProducerRecord}.
+	 * Ignored if {@link #useTemplateConverter(boolean) useTemplateConverter} is true.
 	 * @param creator the creator.
 	 * @return the spec.
 	 * @since 5.5.5
@@ -392,7 +390,7 @@ public S producerRecordCreator(ProducerRecordCreator<K, V> creator) {
 
 	/**
 	 * Set to true to use the template's message converter to create the
-	 * {@link ProducerRecord} instead of the
+	 * {@link org.apache.kafka.clients.producer.ProducerRecord} instead of the
 	 * {@link #producerRecordCreator(ProducerRecordCreator) producerRecordCreator}.
 	 * @param use true to use the converter.
 	 * @return the spec.
@@ -409,7 +407,8 @@ public S useTemplateConverter(boolean use) {
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 */
-	public static class KafkaProducerMessageHandlerTemplateSpec<K, V> extends KafkaProducerMessageHandlerSpec<K, V, KafkaProducerMessageHandlerTemplateSpec<K, V>>
+	public static class KafkaProducerMessageHandlerTemplateSpec<K, V>
+			extends KafkaProducerMessageHandlerSpec<K, V, KafkaProducerMessageHandlerTemplateSpec<K, V>>
 			implements ComponentsRegistration {
 
 		private final KafkaTemplateSpec<K, V> kafkaTemplateSpec;
@@ -428,6 +427,7 @@ public static class KafkaProducerMessageHandlerTemplateSpec<K, V> extends KafkaP
 		 */
 		public KafkaProducerMessageHandlerTemplateSpec<K, V> configureKafkaTemplate(
 				Consumer<KafkaTemplateSpec<K, V>> configurer) {
+
 			Assert.notNull(configurer, "The 'configurer' cannot be null");
 			configurer.accept(this.kafkaTemplateSpec);
 			return _this();

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -138,8 +138,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private Duration closeTimeout = Duration.ofSeconds(DEFAULT_CLOSE_TIMEOUT);
 
-	public boolean newAssignment;
-
 	private volatile Consumer<K, V> consumer;
 
 	private volatile boolean pausing;
@@ -150,6 +148,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private volatile boolean stopped;
 
+	public volatile boolean newAssignment; // NOSONAR - direct access from inner
+
 	/**
 	 * Construct an instance with the supplied parameters. Fetching multiple
 	 * records per poll will be disabled.
@@ -415,7 +415,7 @@ public boolean isPaused() {
 		return this.paused;
 	}
 
-	@Override
+	@Override // NOSONAR - not so complex
 	protected synchronized Object doReceive() {
 		if (this.stopped) {
 			this.logger.debug("Message source is stopped; no records will be returned");

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

@@ -254,7 +254,7 @@ public void removeTopic(String... topic) {
 		}
 	}
 
-	private synchronized void connectAndSubscribe() throws MqttException {
+	private synchronized void connectAndSubscribe() throws MqttException { // NOSONAR
 		MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
 		this.cleanSession = connectionOptions.isCleanSession();
 		this.consumerStopAction = this.clientFactory.getConsumerStopAction();

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttHeaderMapper.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.mqtt.support;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -61,7 +62,9 @@ public class MqttHeaderMapper implements HeaderMapper<MqttProperties> {
 	 */
 	public void setInboundHeaderNames(String... inboundHeaderNames) {
 		Assert.notNull(inboundHeaderNames, "'inboundHeaderNames' must not be null");
-		this.inboundHeaderNames = inboundHeaderNames;
+		String[] copy = Arrays.copyOf(inboundHeaderNames, inboundHeaderNames.length);
+		Arrays.sort(copy);
+		this.inboundHeaderNames = copy;
 	}
 
 	/**
@@ -73,7 +76,9 @@ public void setInboundHeaderNames(String... inboundHeaderNames) {
 	 */
 	public void setOutboundHeaderNames(String... outboundHeaderNames) {
 		Assert.notNull(outboundHeaderNames, "'outboundHeaderNames' must not be null");
-		this.outboundHeaderNames = outboundHeaderNames;
+		String[] copy = Arrays.copyOf(outboundHeaderNames, outboundHeaderNames.length);
+		Arrays.sort(copy);
+		this.outboundHeaderNames = copy;
 	}
 
 	@Override