Fix Sonar issues for previous commit

Author: garyrussell

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

@@ -41,7 +41,7 @@ default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> d
 	void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer);
 
 	@Override
-	default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	default void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
 		handle(thrownException, data, consumer);

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

@@ -50,8 +50,9 @@ default void handle(Exception thrownException, @Nullable ConsumerRecord<?, ?> da
 	void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer);
 
 	@Override
-	default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
-			MessageListenerContainer container) {
+	default void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
+			Consumer<?, ?> consumer, MessageListenerContainer container) {
+
 		handle(thrownException, records, consumer);
 	}
 

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

@@ -23,6 +23,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
 import org.springframework.kafka.KafkaException;
+import org.springframework.lang.Nullable;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.BackOffExecution;
 
@@ -55,26 +56,28 @@ public void setBackOff(BackOff backOff) {
 	}
 
 	@Override
-	public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
-		data.partitions()
-				.stream()
-				.collect(
-						Collectors.toMap(tp -> tp,
-								tp -> data.records(tp).get(0).offset(), (u, v) -> (long) v, LinkedHashMap::new))
-				.forEach(consumer::seek);
+		if (data != null) {
+			data.partitions()
+					.stream()
+					.collect(
+							Collectors.toMap(tp -> tp,
+									tp -> data.records(tp).get(0).offset(), (u, v) -> (long) v, LinkedHashMap::new))
+					.forEach(consumer::seek);
 
-		if (this.backOff != null) {
-			try {
-				ListenerUtils.unrecoverableBackOff(this.backOff, this.backOffs, this.lastIntervals, container);
+			if (this.backOff != null) {
+				try {
+					ListenerUtils.unrecoverableBackOff(this.backOff, this.backOffs, this.lastIntervals, container);
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+				}
 			}
-			catch (InterruptedException e) {
-				Thread.currentThread().interrupt();
-			}
-		}
 
-		throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);
+			throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);
+		}
 	}
 
 	@Override

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

@@ -106,11 +106,13 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	}
 
 	@Override
-	public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
+	public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {
 
-		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
-				getRecoveryStrategy(records, thrownException), this.logger, getLogLevel());
+		if (records != null) {
+			SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
+					getRecoveryStrategy(records, thrownException), this.logger, getLogLevel());
+		}
 	}
 
 }