GH-1876: Fix new Sonar Issues

Author: garyrussell

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

@@ -2712,6 +2712,18 @@ private void initPartitionsIfNeeded() {
 					partitions.put(tp, new OffsetMetadata(off.offset(), false, SeekPosition.TIMESTAMP));
 				}
 			});
+			doInitialSeeks(partitions, beginnings, ends);
+			if (this.consumerSeekAwareListener != null) {
+				this.consumerSeekAwareListener.onPartitionsAssigned(partitions.keySet().stream()
+							.map(tp -> new SimpleEntry<>(tp, this.consumer.position(tp)))
+							.collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue())),
+						this.seekCallback);
+			}
+		}
+
+		private void doInitialSeeks(Map<TopicPartition, OffsetMetadata> partitions, Set<TopicPartition> beginnings,
+				Set<TopicPartition> ends) {
+
 			if (beginnings.size() > 0) {
 				this.consumer.seekToBeginning(beginnings);
 			}
@@ -2746,12 +2758,6 @@ else if (metadata.relativeToCurrent) {
 					}
 				}
 			}
-			if (this.consumerSeekAwareListener != null) {
-				this.consumerSeekAwareListener.onPartitionsAssigned(partitions.keySet().stream()
-							.map(tp -> new SimpleEntry<>(tp, this.consumer.position(tp)))
-							.collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue())),
-						this.seekCallback);
-			}
 		}
 
 		private void logReset(TopicPartition topicPartition, long newOffset) {

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

@@ -103,7 +103,7 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	}
 
 	@Override
-	public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
 		doHandle(thrownException, data, consumer, container, () -> { });