GH-2622: Fix Possible NPE

MLC can be null.

Other Sonar fixes.

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

@@ -29,6 +29,8 @@
  */
 public class ContainerPausingBackOffHandler implements BackOffHandler {
 
+	private final DefaultBackOffHandler defaultBackOffHandler = new DefaultBackOffHandler();
+
 	private final ListenerContainerPauseService pauser;
 
 	/**
@@ -41,7 +43,12 @@ public ContainerPausingBackOffHandler(ListenerContainerPauseService pauser) {
 
 	@Override
 	public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
-		this.pauser.pause(container, Duration.ofMillis(nextBackOff));
+		if (container == null) {
+			this.defaultBackOffHandler.onNextBackOff(container, exception, nextBackOff);
+		}
+		else {
+			this.pauser.pause(container, Duration.ofMillis(nextBackOff));
+		}
 	}
 
 }

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

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Default {@link BackOffHandler}; suspends the thread for the back off. If a container is
+ * provided, {@link ListenerUtils#stoppableSleep(MessageListenerContainer, long)} is used,
+ * to terminate the suspension if the container is stopped.
+ *
+ * @author Jan Marincek
+ * @author Gary Russell
+ * @since 2.9
+ */
+public class DefaultBackOffHandler implements BackOffHandler {
+
+	@Override
+	public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
+		try {
+			if (container == null) {
+				Thread.sleep(nextBackOff);
+			}
+			else {
+				ListenerUtils.stoppableSleep(container, nextBackOff);
+			}
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+	}
+
+}

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

@@ -307,21 +307,4 @@ void setLastException(Exception lastException) {
 
 	}
 
-	static class DefaultBackOffHandler implements BackOffHandler {
-		@Override
-		public void onNextBackOff(@Nullable MessageListenerContainer container, Exception exception, long nextBackOff) {
-			try {
-				if (container == null) {
-					Thread.sleep(nextBackOff);
-				}
-				else {
-					ListenerUtils.stoppableSleep(container, nextBackOff);
-				}
-			}
-			catch (InterruptedException e) {
-				throw new RuntimeException(e);
-			}
-		}
-	}
-
 }

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

@@ -24,7 +24,6 @@
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.log.LogAccessor;
-import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
@@ -93,7 +92,7 @@ public void pause(MessageListenerContainer messageListenerContainer, Duration pa
 	 * Resume the listener container by given id.
 	 * @param listenerId the id of the listener
 	 */
-	public void resume(@NonNull String listenerId) {
+	public void resume(String listenerId) {
 		Assert.notNull(this.registry, "Resume by id is only supported when a registry is provided");
 		getListenerContainer(listenerId).ifPresent(this::resume);
 	}
@@ -102,7 +101,7 @@ public void resume(@NonNull String listenerId) {
 	 * Resume the listener container.
 	 * @param messageListenerContainer the listener container
 	 */
-	public void resume(@NonNull MessageListenerContainer messageListenerContainer) {
+	public void resume(MessageListenerContainer messageListenerContainer) {
 		if (messageListenerContainer.isPauseRequested()) {
 			LOGGER.debug(() -> "Resuming container " + messageListenerContainer);
 			messageListenerContainer.resume();