Fix Issues Reported by Sonar

Author: garyrussell

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -119,21 +119,21 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 
 	public static final int DEFAULT_ZK_SESSION_TIMEOUT = 6000;
 
-	private static final Method getBrokerState;
+	private static final Method GET_BROKER_STATE_METHOD;
 
 	static {
 		try {
 			Method method = KafkaServer.class.getDeclaredMethod("brokerState");
 			if (method.getReturnType().equals(AtomicReference.class)) {
-				getBrokerState = method;
+				GET_BROKER_STATE_METHOD = method;
 			}
 			else {
-				getBrokerState = null;
+				GET_BROKER_STATE_METHOD = null;
 			}
 		}
 		catch (NoSuchMethodException | SecurityException e) {
-			throw new RuntimeException("Failed to determine KafkaServer.brokerState() method; client version: "
-					+ AppInfoParser.getVersion());
+			throw new IllegalStateException("Failed to determine KafkaServer.brokerState() method; client version: "
+					+ AppInfoParser.getVersion(), e);
 		}
 	}
 
@@ -579,9 +579,9 @@ public void destroy() {
 	}
 
 	private boolean brokerRunning(KafkaServer kafkaServer) {
-		if (getBrokerState != null) {
+		if (GET_BROKER_STATE_METHOD != null) {
 			try {
-				return !getBrokerState.invoke(kafkaServer).toString().equals("NOT_RUNNING");
+				return !GET_BROKER_STATE_METHOD.invoke(kafkaServer).toString().equals("NOT_RUNNING");
 			}
 			catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException ex) {
 				throw new IllegalStateException(ex);

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

@@ -19,8 +19,6 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 
-import org.springframework.lang.Nullable;
-
 /**
  * Interface for backing off a {@link MessageListenerContainer}
  * until a given dueTimestamp, if such timestamp is in the future.
@@ -34,7 +32,7 @@ public interface KafkaConsumerBackoffManager {
 	void backOffIfNecessary(Context context);
 
 	default Context createContext(long dueTimestamp, String listenerId, TopicPartition topicPartition,
-								@Nullable Consumer<?, ?> messageConsumer) {
+								Consumer<?, ?> messageConsumer) {
 		return new Context(dueTimestamp, topicPartition, listenerId, messageConsumer);
 	}
 
@@ -66,7 +64,7 @@ class Context {
 		private final Consumer<?, ?> consumerForTimingAdjustment;
 
 		Context(long dueTimestamp, TopicPartition topicPartition, String listenerId,
-				@Nullable Consumer<?, ?> consumerForTimingAdjustment) {
+				Consumer<?, ?> consumerForTimingAdjustment) {
 
 			this.dueTimestamp = dueTimestamp;
 			this.listenerId = listenerId;
@@ -86,7 +84,7 @@ public TopicPartition getTopicPartition() {
 			return this.topicPartition;
 		}
 
-		public @Nullable Consumer<?, ?> getConsumerForTimingAdjustment() {
+		public Consumer<?, ?> getConsumerForTimingAdjustment() {
 			return this.consumerForTimingAdjustment;
 		}
 

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

@@ -196,7 +196,7 @@ private long applyTimingAdjustment(Context context, long timeUntilDue, long poll
 					"Skipping timing adjustment for TopicPartition %s.", context.getTopicPartition()));
 			return 0L;
 		}
-		return this.kafkaConsumerTimingAdjuster.adjustTiming(
+		return this.kafkaConsumerTimingAdjuster.adjustTiming( // NOSONAR
 						context.getConsumerForTimingAdjustment(),
 						context.getTopicPartition(), pollTimeout, timeUntilDue);
 	}

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

@@ -116,7 +116,9 @@ private void invokeDelegateOnMessage(ConsumerRecord<K, V> consumerRecord, Acknow
 		}
 	}
 
-	private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord<K, V> data, long nextExecutionTimestamp, Consumer<?, ?> consumer) {
+	private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord<K, V> data, long nextExecutionTimestamp,
+			Consumer<?, ?> consumer) {
+
 		return this.kafkaConsumerBackoffManager.createContext(nextExecutionTimestamp, this.listenerId,
 				new TopicPartition(data.topic(), data.partition()), consumer);
 	}

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

@@ -135,7 +135,7 @@ public void setRawRecordHeader(boolean rawRecordHeader) {
 		this.rawRecordHeader = rawRecordHeader;
 	}
 
-	@Override
+	@Override // NOSONAR
 	public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknowledgment acknowledgment,
 			Consumer<?, ?> consumer, Type type) {