Fix Sonar Issues

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -144,6 +144,8 @@
 public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton {
 
+	private static final String UNCHECKED = "unchecked";
+
 	private static final String THE_LEFT = "The [";
 
 	private static final String RESOLVED_TO_LEFT = "Resolved to [";
@@ -643,7 +645,7 @@ private void resolveContentTypeConverter(MethodKafkaListenerEndpoint<?, ?> endpo
 		}
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@SuppressWarnings({ "rawtypes", UNCHECKED })
 	private void resolveFilter(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
 		Object filter = resolveExpression(kafkaListener.filter());
 		if (filter instanceof RecordFilterStrategy) {
@@ -701,7 +703,7 @@ protected String noBeanFoundMessage(Object target, String listenerBeanName, Stri
 				+ requestedBeanName + "' was found in the application context";
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings(UNCHECKED)
 	private void resolveKafkaProperties(MethodKafkaListenerEndpoint<?, ?> endpoint, String[] propertyStrings) {
 		if (propertyStrings.length > 0) {
 			Properties properties = new Properties();
@@ -870,7 +872,7 @@ else if (relativeToCurrentValue instanceof Boolean) {
 		return relativeToCurrent;
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings(UNCHECKED)
 	private void resolveAsString(Object resolvedValue, List<String> result) {
 		if (resolvedValue instanceof String[]) {
 			for (Object object : (String[]) resolvedValue) {
@@ -891,7 +893,7 @@ else if (resolvedValue instanceof Iterable) {
 		}
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings(UNCHECKED)
 	private void resolvePartitionAsInteger(String topic, Object resolvedValue,
 			List<TopicPartitionOffset> result, @Nullable Long offset, boolean isRelative, boolean checkDups) {
 

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

@@ -152,6 +152,10 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 
 	private static final String UNUSED = "unused";
 
+	private static final String UNCHECKED = "unchecked";
+
+	private static final String RAWTYPES = "rawtypes";
+
 	private static final int DEFAULT_ACK_TIME = 5000;
 
 	private static final Map<String, Object> CONSUMER_CONFIG_DEFAULTS = ConsumerConfig.configDef().defaultValues();
@@ -547,12 +551,6 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private static final String ERROR_HANDLER_THREW_AN_EXCEPTION = "Error handler threw an exception";
 
-		private static final String UNCHECKED = "unchecked";
-
-		private static final String RAWTYPES = "rawtypes";
-
-		private static final String RAW_TYPES = RAWTYPES;
-
 		private final LogAccessor logger = KafkaMessageListenerContainer.this.logger; // NOSONAR hide
 
 		private final ContainerProperties containerProperties = getContainerProperties();
@@ -611,7 +609,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final PlatformTransactionManager transactionManager = this.containerProperties.getTransactionManager();
 
-		@SuppressWarnings(RAW_TYPES)
+		@SuppressWarnings(RAWTYPES)
 		private final KafkaAwareTransactionManager kafkaTxManager =
 				this.transactionManager instanceof KafkaAwareTransactionManager
 						? ((KafkaAwareTransactionManager) this.transactionManager) : null;
@@ -708,10 +706,10 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final Map<TopicPartition, Boolean> wasIdlePartition;
 
-		private final byte[] listenerinfo = getListenerInfo();
-
 		private final Header infoHeader = new RecordHeader(KafkaHeaders.LISTENER_INFO, this.listenerinfo);
 
+		private final byte[] listenerinfo = getListenerInfo();
+
 		private final Set<TopicPartition> pausedForNack = new HashSet<>();
 
 		private Map<TopicPartition, OffsetMetadata> definedPartitions;
@@ -1666,16 +1664,20 @@ private void pausePartitionsIfNecessary() {
 		}
 
 		private void resumePartitionsIfNecessary() {
-			List<TopicPartition> partitionsToResume = getAssignedPartitions()
-					.stream()
-					.filter(tp -> !isPartitionPauseRequested(tp)
-							&& this.pausedPartitions.contains(tp))
-					.collect(Collectors.toList());
-			if (partitionsToResume.size() > 0) {
-				this.consumer.resume(partitionsToResume);
-				this.pausedPartitions.removeAll(partitionsToResume);
-				this.logger.debug(() -> "Resumed consumption from " + partitionsToResume);
-				partitionsToResume.forEach(KafkaMessageListenerContainer.this::publishConsumerPartitionResumedEvent);
+			Collection<TopicPartition> assigned = getAssignedPartitions();
+			if (assigned != null) {
+				List<TopicPartition> partitionsToResume = assigned
+						.stream()
+						.filter(tp -> !isPartitionPauseRequested(tp)
+								&& this.pausedPartitions.contains(tp))
+						.collect(Collectors.toList());
+				if (partitionsToResume.size() > 0) {
+					this.consumer.resume(partitionsToResume);
+					this.pausedPartitions.removeAll(partitionsToResume);
+					this.logger.debug(() -> "Resumed consumption from " + partitionsToResume);
+					partitionsToResume
+							.forEach(KafkaMessageListenerContainer.this::publishConsumerPartitionResumedEvent);
+				}
 			}
 		}
 
@@ -1988,7 +1990,7 @@ private void invokeBatchListener(final ConsumerRecords<K, V> recordsArg) {
 			}
 		}
 
-		@SuppressWarnings(RAW_TYPES)
+		@SuppressWarnings(RAWTYPES)
 		private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
 				@Nullable final List<ConsumerRecord<K, V>> recordList) {
 
@@ -2299,7 +2301,7 @@ private void invokeRecordListener(final ConsumerRecords<K, V> records) {
 		 * Invoke the listener with each record in a separate transaction.
 		 * @param records the records.
 		 */
-		@SuppressWarnings(RAW_TYPES) // NOSONAR complexity
+		@SuppressWarnings(RAWTYPES) // NOSONAR complexity
 		private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -2485,7 +2487,10 @@ private void pauseForNackSleep() {
 				this.nackWake = System.currentTimeMillis() + this.nackSleep;
 				this.nackSleep = -1;
 				Set<TopicPartition> alreadyPaused = this.consumer.paused();
-				this.pausedForNack.addAll(getAssignedPartitions());
+				Collection<TopicPartition> assigned = getAssignedPartitions();
+				if (assigned != null) {
+					this.pausedForNack.addAll(assigned);
+				}
 				this.pausedForNack.removeAll(alreadyPaused);
 				this.logger.debug(() -> "Pausing for nack sleep: " + ListenerConsumer.this.pausedForNack);
 				try {

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java

@@ -357,6 +357,7 @@ void shouldInstantiateIfNotInContainer() {
 	}
 
 	@Test
+	@SuppressWarnings("deprecation")
 	void shouldLogConsumerRecordMessage() {
 		RetryTopicConfigurer.LoggingDltListenerHandlerMethod method =
 				new RetryTopicConfigurer.LoggingDltListenerHandlerMethod();