Sonar Fixes

spring-projects · Jul 14, 2021 · 2cab170 · 2cab170
1 parent 1875a6c
commit 2cab170
Show file tree

Hide file tree

Showing 14 changed files with 43 additions and 32 deletions.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java b/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);

diff --git a/.../main/java/org/springframework/kafka/listener/ConditionalDelegatingBatchErrorHandler.java b/.../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;

diff --git a/...a/src/main/java/org/springframework/kafka/listener/ConditionalDelegatingErrorHandler.java b/...a/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;
 				}

diff --git a/...afka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java b/...afka/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,

diff --git a/...fka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java b/...fka/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);
 	}

diff --git a/.../src/main/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandler.java b/.../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

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/GenericErrorHandler.java b/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);
 	}
 

diff --git a/...kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/...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);
+				}
 			}
 		}
 

diff --git a/...a/src/main/java/org/springframework/kafka/listener/ListenerInvokingBatchErrorHandler.java b/...a/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);
 
 }
diff --git a/...g-kafka/src/main/java/org/springframework/kafka/listener/RecoveringBatchErrorHandler.java b/...g-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<>();

diff --git a/...-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java b/...-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java
@@ -47,7 +47,7 @@ default void handle(Exception thrownException, @Nullable ConsumerRecord<?, ?> da
 	 * @param records the remaining records including the one that failed.
 	 * @param consumer the consumer.
 	 */
-	void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer);
+	void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer);
 
 	@Override
 	default void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/RetryingBatchErrorHandler.java
@@ -93,7 +93,7 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	}
 
 	@Override
-	public void handle(Exception thrownException, ConsumerRecords<?, ?> records,
+	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
 
 		if (records == null || records.count() == 0) {

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java
@@ -109,7 +109,7 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {
 
-		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
+		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(), // NOSONAR
 				getRecoveryStrategy(records, thrownException), this.logger, getLogLevel());
 	}
 

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java
@@ -202,7 +202,7 @@ public static void seekOrRecover(Exception thrownException, @Nullable List<Consu
 			}
 		}
 
-		if (records == null || !doSeeks(records, consumer, thrownException, true, recovery, container, logger)) {
+		if (records == null || !doSeeks(records, consumer, thrownException, true, recovery, container, logger)) { // NOSONAR
 			throw new KafkaException("Seek to current after exception", level, thrownException);
 		}
 		if (commitRecovered) {