Sonar Fixes

Author: garyrussell

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

@@ -19,6 +19,8 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
+import org.springframework.lang.Nullable;
+
 /**
  * Handles errors thrown during the execution of a {@link BatchMessageListener}.
  * The listener should communicate which position(s) in the list failed in the
@@ -37,7 +39,7 @@ public interface BatchErrorHandler extends GenericErrorHandler<ConsumerRecords<?
 	 * @param consumer the consumer.
 	 * @param container the container.
 	 */
-	default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 		handle(thrownException, data);
 	}
@@ -51,7 +53,7 @@ default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consu
 	 * @param invokeListener a callback to re-invoke the listener.
 	 * @since 2.3.7
 	 */
-	default void handle(Exception thrownException, ConsumerRecords<?, ?> data,
+	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data,
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
 
 		handle(thrownException, data);

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

@@ -34,7 +34,7 @@
  * @since 2.7.4
  *
  */
-public class ConditionalDelegatingBatchErrorHandler implements ContainerAwareBatchErrorHandler {
+public class ConditionalDelegatingBatchErrorHandler implements ListenerInvokingBatchErrorHandler {
 
 	private final ContainerAwareBatchErrorHandler defaultErrorHandler;
 
@@ -70,21 +70,21 @@ public void addDelegate(Class<? extends Throwable> throwable, ContainerAwareBatc
 	}
 
 	@Override
-	public void handle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
+	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
 		// Never called but, just in case
 		doHandle(thrownException, records, consumer, container, null);
 	}
 
 	@Override
-	public void handle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
+	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container, Runnable invokeListener) {
 
 		doHandle(thrownException, records, consumer, container, invokeListener);
 	}
 
-	protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
+	protected void doHandle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container, @Nullable Runnable invokeListener) {
 
 		Throwable cause = thrownException;

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

@@ -74,7 +74,6 @@ public void addDelegate(Class<? extends Throwable> throwable, ContainerAwareErro
 	public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
-		boolean handled = false;
 		Throwable cause = thrownException;
 		if (cause instanceof ListenerExecutionFailedException) {
 			cause = thrownException.getCause();
@@ -83,7 +82,6 @@ public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?
 			Class<? extends Throwable> causeClass = cause.getClass();
 			for (Entry<Class<? extends Throwable>, ContainerAwareErrorHandler> entry : this.delegates.entrySet()) {
 				if (entry.getKey().isAssignableFrom(causeClass)) {
-					handled = true;
 					entry.getValue().handle(thrownException, records, consumer, container);
 					return;
 				}

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

@@ -38,7 +38,7 @@ default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> d
 	}
 
 	@Override
-	void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer);
+	void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer);
 
 	@Override
 	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,

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

@@ -19,6 +19,8 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
+import org.springframework.lang.Nullable;
+
 /**
  * An error handler that has access to the batch of records from the last poll the
  * consumer, and the container.
@@ -31,12 +33,12 @@
 public interface ContainerAwareBatchErrorHandler extends ConsumerAwareBatchErrorHandler {
 
 	@Override
-	default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer) {
+	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer) {
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 
 	@Override
-	void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container);
 
 	/**
@@ -50,8 +52,8 @@ void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?
 	 */
 	@Override
 	@SuppressWarnings("unused")
-	default void handle(Exception thrownException, ConsumerRecords<?, ?> data,
-			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
+	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data,
+			Consumer<?, ?> consumer, MessageListenerContainer container, @Nullable Runnable invokeListener) {
 
 		handle(thrownException, data, consumer, container);
 	}

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

@@ -23,6 +23,7 @@
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.KafkaException;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -55,7 +56,7 @@ public ContainerStoppingBatchErrorHandler(Executor executor) {
 	}
 
 	@Override
-	public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 		this.executor.execute(() -> container.stop());
 		// isRunning is false before the container.stop() waits for listener thread

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

@@ -45,7 +45,7 @@ public interface GenericErrorHandler<T> {
 	 * @param data the data.
 	 * @param consumer the consumer.
 	 */
-	default void handle(Exception thrownException, T data, Consumer<?, ?> consumer) {
+	default void handle(Exception thrownException, @Nullable T data, Consumer<?, ?> consumer) {
 		handle(thrownException, data);
 	}
 

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

@@ -1534,16 +1534,19 @@ private void doResumeConsumerIfNeccessary() {
 
 		private void pausePartitionsIfNecessary() {
 			Set<TopicPartition> pausedConsumerPartitions = this.consumer.paused();
-			List<TopicPartition> partitionsToPause = getAssignedPartitions()
-					.stream()
-					.filter(tp -> isPartitionPauseRequested(tp)
-							&& !pausedConsumerPartitions.contains(tp))
-					.collect(Collectors.toList());
-			if (partitionsToPause.size() > 0) {
-				this.consumer.pause(partitionsToPause);
-				this.pausedPartitions.addAll(partitionsToPause);
-				this.logger.debug(() -> "Paused consumption from " + partitionsToPause);
-				partitionsToPause.forEach(KafkaMessageListenerContainer.this::publishConsumerPartitionPausedEvent);
+			Collection<TopicPartition> partitions = getAssignedPartitions();
+			if (partitions != null) {
+				List<TopicPartition> partitionsToPause = partitions
+						.stream()
+						.filter(tp -> isPartitionPauseRequested(tp)
+								&& !pausedConsumerPartitions.contains(tp))
+						.collect(Collectors.toList());
+				if (partitionsToPause.size() > 0) {
+					this.consumer.pause(partitionsToPause);
+					this.pausedPartitions.addAll(partitionsToPause);
+					this.logger.debug(() -> "Paused consumption from " + partitionsToPause);
+					partitionsToPause.forEach(KafkaMessageListenerContainer.this::publishConsumerPartitionPausedEvent);
+				}
 			}
 		}
 

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

@@ -19,6 +19,8 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
+import org.springframework.lang.Nullable;
+
 /**
  * A batch error handler that is capable of invoking the listener during error handling.
  *
@@ -30,14 +32,14 @@
 public interface ListenerInvokingBatchErrorHandler extends ContainerAwareBatchErrorHandler {
 
 	@Override
-	default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 
 	@Override
-	void handle(Exception thrownException, ConsumerRecords<?, ?> records,
+	void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener);
 
 }

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

@@ -155,9 +155,12 @@ private int findIndex(ConsumerRecords<?, ?> data, ConsumerRecord<?, ?> record) {
 		return i;
 	}
 
-	private void seekOrRecover(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	private void seekOrRecover(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container, int indexArg) {
 
+		if (data == null) {
+			return;
+		}
 		Iterator<?> iterator = data.iterator();
 		List<ConsumerRecord<?, ?>> toCommit = new ArrayList<>();
 		List<ConsumerRecord<?, ?>> remaining = new ArrayList<>();