Resolve new Sonar Issues

Author: garyrussell

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

@@ -399,7 +399,7 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 				.findRetryConfigurationFor(kafkaListener.topics(), methodToUse, bean);
 
 		if (retryTopicConfiguration == null) {
-			this.logger.debug("No retry topic configuration found for topics " + kafkaListener.topics() + ".");
+			this.logger.debug("No retry topic configuration found for topics " + Arrays.asList(kafkaListener.topics()));
 			return false;
 		}
 
@@ -413,15 +413,14 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 	}
 
 	private RetryTopicConfigurer getRetryTopicConfigurer() {
-		RetryTopicConfigurer retryTopicConfigurer;
 		try {
 			return this.beanFactory.getBean(RetryTopicConfigurer.class);
 		}
 		catch (NoSuchBeanDefinitionException e) {
 			if (!(this.beanFactory instanceof AutowireCapableBeanFactory)) {
 				throw new IllegalStateException("BeanFactory must be an instance of "
 						+ AutowireCapableBeanFactory.class.getSimpleName()
-						+ " Provided beanFactory: " + this.beanFactory.getClass().getSimpleName());
+						+ " Provided beanFactory: " + this.beanFactory.getClass().getSimpleName(), e);
 			}
 			((AutowireCapableBeanFactory) this.beanFactory)
 					.createBean(RetryTopicBootstrapper.class)
@@ -471,9 +470,9 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 	}
 
 	protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener,
-								Object bean, Object adminTarget, String beanName,
-								Function<MethodKafkaListenerEndpoint<?, ?>, MethodKafkaListenerEndpoint<?, ?>> endpointCustomizer,
-								Function<KafkaListenerContainerFactory<?>, KafkaListenerContainerFactory<?>> containerFactoryCustomizer) {
+			Object bean, Object adminTarget, String beanName,
+			Function<MethodKafkaListenerEndpoint<?, ?>, MethodKafkaListenerEndpoint<?, ?>> endpointCustomizer,
+			Function<KafkaListenerContainerFactory<?>, KafkaListenerContainerFactory<?>> containerFactoryCustomizer) {
 
 		String beanRef = kafkaListener.beanRef();
 		if (StringUtils.hasText(beanRef)) {
@@ -505,18 +504,18 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 		resolveKafkaProperties(endpoint, kafkaListener.properties());
 		endpoint.setSplitIterables(kafkaListener.splitIterables());
 
-		endpoint = endpointCustomizer.apply(endpoint);
+		MethodKafkaListenerEndpoint<?, ?> customizedEndpoint = endpointCustomizer.apply(endpoint);
 
 		String containerFactory = resolve(kafkaListener.containerFactory());
 		KafkaListenerContainerFactory<?> containerFactoryInstanceOrNull = containerFactoryCustomizer
 				.apply(resolveContainerFactory(kafkaListener, containerFactory, beanName));
 
-		this.registrar.registerEndpoint(endpoint, containerFactoryInstanceOrNull);
+		this.registrar.registerEndpoint(customizedEndpoint, containerFactoryInstanceOrNull);
 
-		endpoint.setBeanFactory(this.beanFactory);
+		customizedEndpoint.setBeanFactory(this.beanFactory);
 		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
 		if (StringUtils.hasText(errorHandlerBeanName)) {
-			resolveErrorHandler(endpoint, kafkaListener);
+			resolveErrorHandler(customizedEndpoint, kafkaListener);
 		}
 		if (StringUtils.hasText(beanRef)) {
 			this.listenerScope.removeListener(beanRef);

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

@@ -24,21 +24,22 @@
 
 import org.springframework.kafka.retrytopic.DltStrategy;
 import org.springframework.kafka.retrytopic.FixedDelayStrategy;
-import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
 import org.springframework.kafka.retrytopic.RetryTopicConstants;
 import org.springframework.kafka.retrytopic.TopicSuffixingStrategy;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.policy.MaxAttemptsRetryPolicy;
 
 /**
  *
- * Annotation to create the retry and dlt topics for a {@link KafkaListener} annotated listener.
- * See {@link RetryTopicConfigurer for usage examples.}
+ * Annotation to create the retry and dlt topics for a {@link KafkaListener} annotated
+ * listener. See {@link org.springframework.kafka.retrytopic.RetryTopicConfigurer} for
+ * usage examples.
  *
  * @author Tomaz Fernandes
+ * @author Gary Russell
  * @since 2.7
  *
- * @see RetryTopicConfigurer
+ * @see org.springframework.kafka.retrytopic.RetryTopicConfigurer
  */
 @Target({ ElementType.METHOD })
 @Retention(RetentionPolicy.RUNTIME)
@@ -62,8 +63,8 @@
 
 	/**
 	 *
-	 * The amount of time in milliseconds after which message retrying should give up
-	 * and send the message to the DLT.
+	 * The amount of time in milliseconds after which message retrying should give up and
+	 * send the message to the DLT.
 	 *
 	 * @return the timeout value.
 	 *
@@ -72,27 +73,30 @@
 
 	/**
 	 *
-	 * The bean name of the {@link org.springframework.kafka.core.KafkaTemplate} bean
-	 * that will be used to forward the message to the retry and Dlt topics. If not specified,
+	 * The bean name of the {@link org.springframework.kafka.core.KafkaTemplate} bean that
+	 * will be used to forward the message to the retry and Dlt topics. If not specified,
 	 * a bean with name retryTopicDefaultKafkaTemplate will be looked up.
 	 *
 	 * @return the kafkaTemplate bean name.
 	 */
 	String kafkaTemplate() default "";
 
 	/**
-	 *  The bean name of the {@link org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory} that will be used to create
-	 *  the consumers for the retry and dlt topics. If none is provided, the one from the {@link KafkaListener} annotation
-	 *  is used, or else a default one, if any.
+	 * The bean name of the
+	 * {@link org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory}
+	 * that will be used to create the consumers for the retry and dlt topics. If none is
+	 * provided, the one from the {@link KafkaListener} annotation is used, or else a
+	 * default one, if any.
 	 *
 	 * @return the listenerContainerFactory bean name.
 	 */
 	String listenerContainerFactory() default "";
 
 	/**
-	 * Whether or not the topic should be created after registration with the provided configurations.
-	 * Not to be confused with the ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG from Kafka
-	 * configuration, which is handled by the {@link org.apache.kafka.clients.consumer.KafkaConsumer}.
+	 * Whether or not the topic should be created after registration with the provided
+	 * configurations. Not to be confused with the
+	 * ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG from Kafka configuration, which is
+	 * handled by the {@link org.apache.kafka.clients.consumer.KafkaConsumer}.
 	 *
 	 * @return the configuration.
 	 */
@@ -120,50 +124,46 @@
 	Class<? extends Throwable>[] include() default {};
 
 	/**
-	 * The exceptions that should not be retried.
-	 * When the message processing throws these exceptions
-	 * the message goes straight to the DLT.
+	 * The exceptions that should not be retried. When the message processing throws these
+	 * exceptions the message goes straight to the DLT.
 	 *
 	 * @return the exceptions not to be retried.
 	 */
 	Class<? extends Throwable>[] exclude() default {};
 
 	/**
-	 * Whether or not the captured exception should be
-	 * traversed to look for the exceptions provided above.
+	 * Whether or not the captured exception should be traversed to look for the
+	 * exceptions provided above.
 	 *
 	 * @return the value.
 	 */
 	boolean traversingCauses() default false;
 
 	/**
-	 * The suffix that will be appended to the main topic in order to generate
-	 * the retry topics. The corresponding delay value is also appended.
+	 * The suffix that will be appended to the main topic in order to generate the retry
+	 * topics. The corresponding delay value is also appended.
 	 *
 	 * @return the retry topics' suffix.
 	 */
 	String retryTopicSuffix() default RetryTopicConstants.DEFAULT_RETRY_SUFFIX;
 
 	/**
-	 * The suffix that will be appended to the main topic in order to generate
-	 * the dlt topic.
+	 * The suffix that will be appended to the main topic in order to generate the dlt
+	 * topic.
 	 *
 	 * @return the dlt suffix.
 	 */
 	String dltTopicSuffix() default RetryTopicConstants.DEFAULT_DLT_SUFFIX;
 
-
 	/**
-	 * The suffix that will be appended to the main topic in order to generate
-	 * the dlt topic.
+	 * The suffix that will be appended to the main topic in order to generate the dlt
+	 * topic.
 	 *
 	 * @return the dlt suffix.
 	 */
 	TopicSuffixingStrategy topicSuffixingStrategy() default TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE;
 
-	DltStrategy dltStrategy()
-			default DltStrategy.ALWAYS_RETRY_ON_ERROR;
+	DltStrategy dltStrategy() default DltStrategy.ALWAYS_RETRY_ON_ERROR;
 
-	FixedDelayStrategy fixedDelayTopicStrategy()
-			default FixedDelayStrategy.MULTIPLE_TOPICS;
+	FixedDelayStrategy fixedDelayTopicStrategy() default FixedDelayStrategy.MULTIPLE_TOPICS;
 }

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

@@ -22,29 +22,25 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationListener;
-import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.event.ListenerContainerPartitionIdleEvent;
 
 /**
  *
- * A manager that backs off consumption for a given topic if the timestamp provided is not due.
- * Use with {@link SeekToCurrentErrorHandler} to guarantee that the message is read again after
- * partition consumption is resumed (or seek it manually by other means).
+ * A manager that backs off consumption for a given topic if the timestamp provided is not
+ * due. Use with {@link SeekToCurrentErrorHandler} to guarantee that the message is read
+ * again after partition consumption is resumed (or seek it manually by other means).
  *
  * @author Tomaz Fernandes
  * @since 2.7
  * @see SeekToCurrentErrorHandler
  */
 public class KafkaConsumerBackoffManager implements ApplicationListener<ListenerContainerPartitionIdleEvent> {
 
-	private static final LogAccessor logger = new LogAccessor(LogFactory.getLog(KafkaConsumerBackoffManager.class));
-
 	/**
 	 * Internal Back Off Clock Bean Name.
 	 */
@@ -121,17 +117,17 @@ public static class Context {
 		/**
 		 * The time after which the message should be processed.
 		 */
-		final long dueTimestamp;
+		final long dueTimestamp; // NOSONAR
 
 		/**
 		 * The id for the listener that should be paused.
 		 */
-		final String listenerId;
+		final String listenerId; // NOSONAR
 
 		/**
 		 * The topic that contains the partition to be paused.
 		 */
-		final TopicPartition topicPartition;
+		final TopicPartition topicPartition; // NOSONAR
 
 		Context(long dueTimestamp, String listenerId, TopicPartition topicPartition) {
 			this.dueTimestamp = dueTimestamp;

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

@@ -688,7 +688,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private volatile long lastPoll = System.currentTimeMillis();
 
-		private Set<TopicPartition> pausedPartitions;
+		private final Set<TopicPartition> pausedPartitions;
 
 		@SuppressWarnings(UNCHECKED)
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
@@ -1262,12 +1262,12 @@ private void checkIdlePartitions() {
 		private void checkIdlePartition(TopicPartition topicPartition) {
 			if (this.containerProperties.getIdlePartitionEventInterval() != null) {
 				long now = System.currentTimeMillis();
-				Long lastReceive = this.lastReceivePartition.computeIfAbsent(topicPartition, newTopicPartition -> now);
-				Long lastAlertAt = this.lastAlertPartition.computeIfAbsent(topicPartition, newTopicPartition -> now);
-				if (now > lastReceive + this.containerProperties.getIdlePartitionEventInterval()
-						&& now > lastAlertAt + this.containerProperties.getIdlePartitionEventInterval()) {
+				Long lstReceive = this.lastReceivePartition.computeIfAbsent(topicPartition, newTopicPartition -> now);
+				Long lstAlertAt = this.lastAlertPartition.computeIfAbsent(topicPartition, newTopicPartition -> now);
+				if (now > lstReceive + this.containerProperties.getIdlePartitionEventInterval()
+						&& now > lstAlertAt + this.containerProperties.getIdlePartitionEventInterval()) {
 					this.wasIdlePartition.put(topicPartition, true);
-					publishIdlePartitionEvent(now - lastReceive, topicPartition, this.consumer,
+					publishIdlePartitionEvent(now - lstReceive, topicPartition, this.consumer,
 							isPartitionPauseRequested(topicPartition));
 					this.lastAlertPartition.put(topicPartition, now);
 					if (this.consumerSeekAwareListener != null) {
@@ -1285,11 +1285,11 @@ private void notIdlePartitions(Set<TopicPartition> partitions) {
 
 		private void notIdlePartition(TopicPartition topicPartition) {
 			long now = System.currentTimeMillis();
-			Boolean wasIdle = this.wasIdlePartition.get(topicPartition);
-			if (wasIdle != null && wasIdle) {
+			Boolean partitionWasIdle = this.wasIdlePartition.get(topicPartition);
+			if (partitionWasIdle != null && partitionWasIdle) {
 				this.wasIdlePartition.put(topicPartition, false);
-				Long lastReceive = this.lastReceivePartition.computeIfAbsent(topicPartition, newTopicPartition -> now);
-				publishNoLongerIdlePartitionEvent(now - this.lastReceive, this.consumer, topicPartition);
+				Long lstReceive = this.lastReceivePartition.computeIfAbsent(topicPartition, newTopicPartition -> now);
+				publishNoLongerIdlePartitionEvent(now - lstReceive, this.consumer, topicPartition);
 			}
 			this.lastReceivePartition.put(topicPartition, now);
 		}

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/BackOffValuesGenerator.java

@@ -49,9 +49,9 @@ public class BackOffValuesGenerator {
 
 	public BackOffValuesGenerator(int providedMaxAttempts, BackOffPolicy providedBackOffPolicy) {
 		this.numberOfvaluesToCreate = getMaxAttemps(providedMaxAttempts) - 1;
-		BackOffPolicy backOffPolicy = providedBackOffPolicy != null ? providedBackOffPolicy : DEFAULT_BACKOFF_POLICY;
-		checkBackOffPolicyTipe(backOffPolicy);
-		this.backOffPolicy = backOffPolicy;
+		BackOffPolicy policy = providedBackOffPolicy != null ? providedBackOffPolicy : DEFAULT_BACKOFF_POLICY;
+		checkBackOffPolicyTipe(policy);
+		this.backOffPolicy = policy;
 	}
 
 	public int getMaxAttemps(int providedMaxAttempts) {

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

@@ -58,9 +58,12 @@ public DeadLetterPublishingRecovererFactory(DestinationTopicResolver destination
 	@SuppressWarnings("unchecked")
 	public DeadLetterPublishingRecoverer create(Configuration configuration) {
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(configuration.template,
-					((cr, e) -> this.resolveDestination(cr, e, configuration))) {
+					((cr, e) -> this.resolveDestination(cr, e))) {
+
 			@Override
-			protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate) {
+			protected void publish(ProducerRecord<Object, Object> outRecord,
+					KafkaOperations<Object, Object> kafkaTemplate) {
+
 				if (NO_OPS_RETRY_TOPIC.equals(outRecord.topic())) {
 					this.logger.warn(() -> "Processing failed for last topic, giving up.");
 					return;
@@ -72,6 +75,7 @@ protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations
 								.getKafkaOperations();
 				super.publish(outRecord, kafkaOperationsForTopic);
 			}
+
 		};
 
 		recoverer.setHeadersFunction((consumerRecord, e) -> addHeaders(consumerRecord, e, getAttempts(consumerRecord)));
@@ -83,7 +87,7 @@ public void setDeadLetterPublishingRecovererCustomizer(Consumer<DeadLetterPublis
 		this.recovererCustomizer = customizer;
 	}
 
-	private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e, Configuration configuration) {
+	private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e) {
 		if (isBackoffException(e)) {
 			throw (NestedRuntimeException) e; // Necessary to not commit the offset and seek to current again
 		}

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java

@@ -57,10 +57,12 @@ public class ListenerContainerFactoryConfigurer {
 		configuredFactoriesCache = new HashSet<>();
 	}
 
-	private final static String INTERNAL_KAFKA_CONSUMER_BACKOFF_BEAN_NAME = "kafkaconsumerbackoff-internal";
-	private final static long DEFAULT_IDLE_PARTITION_EVENT_INTERVAL = 1000L;
+	private static final long DEFAULT_IDLE_PARTITION_EVENT_INTERVAL = 1000L;
+
 	private final DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory;
+
 	private Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer = container -> { };
+
 	private Consumer<ErrorHandler> errorHandlerCustomizer = errorHandler -> { };
 
 	ListenerContainerFactoryConfigurer(KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
@@ -94,6 +96,7 @@ private void addToFactoriesCache(ConcurrentKafkaListenerContainerFactory<?, ?> c
 	}
 
 	public void setContainerCustomizer(Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer) {
+		Assert.notNull(containerCustomizer, "'containerCustomizer' cannot be null");
 		this.containerCustomizer = containerCustomizer;
 	}