Fix Sonar Issues

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/PartitionPausingBackOffManagerFactory.java

@@ -118,7 +118,7 @@ public void setTimingAdjustmentEnabled(boolean timingAdjustmentEnabled) {
 	 *
 	 * @param timingAdjustmentManager the adjustmentManager to be used.
 	 */
-	public void setTimingAdjustmentManager(KafkaConsumerTimingAdjuster timingAdjustmentManager) {
+	public final void setTimingAdjustmentManager(KafkaConsumerTimingAdjuster timingAdjustmentManager) {
 		Assert.isTrue(this.timingAdjustmentEnabled, () -> "TimingAdjustment is disabled for this factory.");
 		this.timingAdjustmentManager = timingAdjustmentManager;
 	}
@@ -127,7 +127,7 @@ public void setTimingAdjustmentManager(KafkaConsumerTimingAdjuster timingAdjustm
 	 * Set the {@link TaskExecutor} that will be used in the {@link KafkaConsumerTimingAdjuster}.
 	 * @param taskExecutor the taskExecutor to be used.
 	 */
-	public void setTaskExecutor(TaskExecutor taskExecutor) {
+	public final void setTaskExecutor(TaskExecutor taskExecutor) {
 		Assert.isTrue(this.timingAdjustmentEnabled, () -> "TimingAdjustment is disabled for this factory.");
 		this.taskExecutor = taskExecutor;
 	}

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java

@@ -18,6 +18,7 @@
 
 import java.time.Clock;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -362,7 +363,7 @@ public static class BlockingRetriesConfigurer {
 		@SuppressWarnings("varargs")
 		@SafeVarargs
 		public final BlockingRetriesConfigurer retryOn(Class<? extends Exception>... exceptions) {
-			this.retryableExceptions = exceptions;
+			this.retryableExceptions = Arrays.copyOf(exceptions, exceptions.length);
 			return this;
 		}
 

spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java

@@ -157,18 +157,9 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 		List<Headers> natives = new ArrayList<>();
 		List<ConsumerRecord<?, ?>> raws = new ArrayList<>();
 		List<ConversionException> conversionFailures = new ArrayList<>();
-		if (this.headerMapper != null) {
-			rawHeaders.put(KafkaHeaders.BATCH_CONVERTED_HEADERS, convertedHeaders);
-		}
-		else {
-			rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, natives);
-		}
-		if (this.rawRecordHeader) {
-			rawHeaders.put(KafkaHeaders.RAW_DATA, raws);
-		}
+		addToRawHeaders(rawHeaders, convertedHeaders, natives, raws, conversionFailures);
 		commonHeaders(acknowledgment, consumer, rawHeaders, keys, topics, partitions, offsets, timestampTypes,
 				timestamps);
-		rawHeaders.put(KafkaHeaders.CONVERSION_FAILURES, conversionFailures);
 		boolean logged = false;
 		String info = null;
 		for (ConsumerRecord<?, ?> record : records) {
@@ -210,6 +201,21 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 		return MessageBuilder.createMessage(payloads, kafkaMessageHeaders);
 	}
 
+	private void addToRawHeaders(Map<String, Object> rawHeaders, List<Map<String, Object>> convertedHeaders,
+			List<Headers> natives, List<ConsumerRecord<?, ?>> raws, List<ConversionException> conversionFailures) {
+
+		if (this.headerMapper != null) {
+			rawHeaders.put(KafkaHeaders.BATCH_CONVERTED_HEADERS, convertedHeaders);
+		}
+		else {
+			rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, natives);
+		}
+		if (this.rawRecordHeader) {
+			rawHeaders.put(KafkaHeaders.RAW_DATA, raws);
+		}
+		rawHeaders.put(KafkaHeaders.CONVERSION_FAILURES, conversionFailures);
+	}
+
 	private Object obtainPayload(Type type, ConsumerRecord<?, ?> record, List<ConversionException> conversionFailures) {
 		return this.recordConverter == null || !containerType(type)
 				? extractAndConvertValue(record, type)