Sonar fixes

Author: garyrussell

spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java

@@ -235,7 +235,7 @@ public static OffsetAndMetadata getCurrentOffset(String brokerAddresses, String
 
 		try (AdminClient client = AdminClient
 				.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerAddresses))) {
-			return client.listConsumerGroupOffsets(group).partitionsToOffsetAndMetadata().get()
+			return client.listConsumerGroupOffsets(group).partitionsToOffsetAndMetadata().get() // NOSONAR false positive
 					.get(new TopicPartition(topic, partition));
 		}
 	}

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -337,7 +337,9 @@ public synchronized void stop() {
 					if (this.cleanupConfig.cleanupOnStop()) {
 						this.kafkaStreams.cleanUp();
 					}
-					this.listeners.forEach(listener -> listener.streamsRemoved(this.beanName, this.kafkaStreams));
+					for (Listener listener : this.listeners) {
+						listener.streamsRemoved(this.beanName, this.kafkaStreams);
+					}
 					this.kafkaStreams = null;
 				}
 			}

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

@@ -34,7 +34,7 @@ public interface KafkaFailureCallback<K, V> extends FailureCallback {
 
 	@Override
 	default void onFailure(Throwable ex) {
-		onFailure((KafkaProducerException) ex);
+		onFailure((KafkaProducerException) ex); // NOSONAR (unchecked cast)
 	}
 
 	/**

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

@@ -44,6 +44,8 @@
  */
 public class RoutingKafkaTemplate extends KafkaTemplate<Object, Object> {
 
+	private static final String THIS_METHOD_IS_NOT_SUPPORTED = "This method is not supported";
+
 	private final Map<Pattern, ProducerFactory<Object, Object>> factoryMatchers;
 
 	private final ConcurrentMap<String, ProducerFactory<Object, Object>> factoryMap = new ConcurrentHashMap<>();
@@ -72,7 +74,7 @@ public Producer<Object, Object> createProducer() {
 
 	@Override
 	public ProducerFactory<Object, Object> getProducerFactory() {
-		throw new UnsupportedOperationException("This method is not supported");
+		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
 	@Override
@@ -91,27 +93,27 @@ public ProducerFactory<Object, Object> getProducerFactory(String topic) {
 
 	@Override
 	public <T> T execute(ProducerCallback<Object, Object, T> callback) {
-		throw new UnsupportedOperationException("This method is not supported");
+		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
 	@Override
 	public <T> T executeInTransaction(OperationsCallback<Object, Object, T> callback) {
-		throw new UnsupportedOperationException("This method is not supported");
+		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
 	@Override
 	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
-		throw new UnsupportedOperationException("This method is not supported");
+		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
 	@Override
 	public Map<MetricName, ? extends Metric> metrics() {
-		throw new UnsupportedOperationException("This method is not supported");
+		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
 	@Override
 	public void flush() {
-		throw new UnsupportedOperationException("This method is not supported");
+		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
 }

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

@@ -348,7 +348,7 @@ protected void checkTopics() {
 					.filter(entry -> AdminClientConfig.configNames().contains(entry.getKey()))
 					.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
 			List<String> missing = null;
-			try (AdminClient client = AdminClient.create(configs)) {
+			try (AdminClient client = AdminClient.create(configs)) { // NOSONAR - false positive null check
 				if (client != null) {
 					String[] topics = this.containerProperties.getTopics();
 					if (topics == null) {

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConsumerRecordMetadata.java

@@ -85,8 +85,9 @@ public boolean equals(Object obj) {
 		if (!(obj instanceof ConsumerRecordMetadata)) {
 			return false;
 		}
-		return this.delegate.equals(obj)
-				&& this.timestampType.equals(((ConsumerRecordMetadata) obj).timestampType());
+		ConsumerRecordMetadata crm = (ConsumerRecordMetadata) obj;
+		return this.delegate.equals(crm.delegate)
+				&& this.timestampType.equals(crm.timestampType());
 	}
 
 	@Override

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java

@@ -73,10 +73,19 @@ public static <P, T> BiFunction<P, Headers, T> propertyToMethodInvokingFunction(
 				method = clazz.getDeclaredMethod(methodName, payloadType);
 			}
 			catch (@SuppressWarnings("unused") NoSuchMethodException e1) {
-				throw new IllegalStateException("the parser method must take '(String, Headers)' or '(String)'");
+				IllegalStateException ise =
+						new IllegalStateException("the parser method must take '("
+								+ payloadType.getSimpleName()
+								+ ", Headers)' or '("
+								+ payloadType.getSimpleName()
+								+ ")'", e1);
+				ise.addSuppressed(e);
+				throw ise;
 			}
 			catch (SecurityException e1) {
-				throw new IllegalStateException(e1);
+				IllegalStateException ise = new IllegalStateException(e1);
+				ise.addSuppressed(e);
+				throw ise;
 			}
 		}
 		catch (SecurityException e) {