Fix new and some old Sonar smells

spring-integration-core/src/main/java/org/springframework/integration/filter/ExpressionEvaluatingSelector.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -30,17 +30,19 @@ import org.springframework.integration.handler.ExpressionEvaluatingMessageProces
  * @author Mark Fisher
  * @author Liujiong
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class ExpressionEvaluatingSelector extends AbstractMessageProcessingSelector {
 
-	private static final ExpressionParser expressionParser =
+	private static final ExpressionParser EXPRESSION_PARSER =
 			new SpelExpressionParser(new SpelParserConfiguration(true, true));
 
 	private final String expressionString;
 
 	public ExpressionEvaluatingSelector(String expressionString) {
-		super(new ExpressionEvaluatingMessageProcessor<Boolean>(expressionParser.parseExpression(expressionString),
+		super(new ExpressionEvaluatingMessageProcessor<Boolean>(EXPRESSION_PARSER.parseExpression(expressionString),
 				Boolean.class));
 		this.expressionString = expressionString;
 	}

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -397,11 +397,10 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
 		private void processMapArgument(Object messageOrPayload, boolean foundPayloadAnnotation,
 				Map<String, Object> headersToPopulate, Map<?, ?> argumentValue) {
 
-			if (messageOrPayload instanceof Map && !foundPayloadAnnotation) {
+			if (messageOrPayload instanceof Map && !foundPayloadAnnotation
-				if (GatewayMethodInboundMessageMapper.this.payloadExpression == null) {
+					&& GatewayMethodInboundMessageMapper.this.payloadExpression == null) {
-					throw new MessagingException("Ambiguous method parameters; found more than one " +
+				throw new MessagingException("Ambiguous method parameters; found more than one " +
-							"Map-typed parameter and neither one contains a @Payload annotation");
+						"Map-typed parameter and neither one contains a @Payload annotation");
-				}
 			}
 			copyHeaders(argumentValue, headersToPopulate);
 		}

spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -41,7 +41,7 @@ import org.springframework.beans.BeanUtils;
  */
 public final class SimpleJsonSerializer {
 
-	private static final Log logger = LogFactory.getLog(SimpleJsonSerializer.class);
+	private static final Log LOGGER = LogFactory.getLog(SimpleJsonSerializer.class);
 
 	private SimpleJsonSerializer() {
 	}
@@ -69,12 +69,12 @@ public final class SimpleJsonSerializer {
 				}
 				catch (InvocationTargetException | IllegalAccessException | IllegalArgumentException e) {
 					Throwable exception = e;
-					if (e instanceof InvocationTargetException) {
+					if (e instanceof InvocationTargetException) { // NOSONAR
 						exception = e.getCause();
 					}
 
-					if (logger.isDebugEnabled()) {
+					if (LOGGER.isDebugEnabled()) {
-						logger.debug("Failed to serialize property " + propertyName, exception);
+						LOGGER.debug("Failed to serialize property " + propertyName, exception);
 					}
 
 					result =

spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -35,6 +35,7 @@ import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.core.DestinationResolver;
@@ -210,7 +211,7 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 			MessageSelector selector = next.getSelector();
 			MessageChannel channel = next.getChannel();
 			if (selector instanceof ExpressionEvaluatingSelector
-					&& channel.equals(targetChannel)
+					&& targetChannel.equals(channel)
 					&& ((ExpressionEvaluatingSelector) selector).getExpressionString().equals(selectorExpression)) {
 				it.remove();
 				counter++;
@@ -307,14 +308,13 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi
 			return this.selector;
 		}
 
+		@Nullable
 		public MessageChannel getChannel() {
 			if (this.channel == null) {
 				String channelNameForInitialization = this.channelName;
-				if (channelNameForInitialization != null) {
+				if (channelNameForInitialization != null && this.channelResolver != null) {
-					if (this.channelResolver != null) {
+					this.channel = this.channelResolver.resolveDestination(channelNameForInitialization);
-						this.channel = this.channelResolver.resolveDestination(channelNameForInitialization);
+					this.channelName = null;
-						this.channelName = null;
-					}
 				}
 			}
 			return this.channel;

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2020 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.
@@ -34,7 +34,7 @@ import org.apache.commons.logging.LogFactory;
  */
 public class PollSkipAdvice implements MethodInterceptor {
 
-	private static final Log logger = LogFactory.getLog(PollSkipAdvice.class);
+	private static final Log LOGGER = LogFactory.getLog(PollSkipAdvice.class);
 
 	private final PollSkipStrategy pollSkipStrategy;
 
@@ -51,8 +51,8 @@ public class PollSkipAdvice implements MethodInterceptor {
 	@Override
 	public Object invoke(MethodInvocation invocation) throws Throwable {
 		if ("call".equals(invocation.getMethod().getName()) && this.pollSkipStrategy.skipPoll()) {
-			if (logger.isDebugEnabled()) {
+			if (LOGGER.isDebugEnabled()) {
-				logger.debug("Skipping poll because "
+				LOGGER.debug("Skipping poll because "
 						+ this.pollSkipStrategy.getClass().getName()
 						+ ".skipPoll() returned true");
 			}

spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -33,7 +33,7 @@ import org.springframework.messaging.Message;
  */
 class PersistentMessageGroup implements MessageGroup {
 
-	private static final Log logger = LogFactory.getLog(PersistentMessageGroup.class);
+	private static final Log LOGGER = LogFactory.getLog(PersistentMessageGroup.class);
 
 	private final MessageGroupStore messageGroupStore;
 
@@ -64,8 +64,8 @@ class PersistentMessageGroup implements MessageGroup {
 		if (this.oneMessage == null) {
 			synchronized (this) {
 				if (this.oneMessage == null) {
-					if (logger.isDebugEnabled()) {
+					if (LOGGER.isDebugEnabled()) {
-						logger.debug("Lazy loading of one message for messageGroup: " + this.original.getGroupId());
+						LOGGER.debug("Lazy loading of one message for messageGroup: " + this.original.getGroupId());
 					}
 					this.oneMessage = this.messageGroupStore.getOneMessageFromGroup(this.original.getGroupId());
 				}
@@ -97,8 +97,8 @@ class PersistentMessageGroup implements MessageGroup {
 		if (this.size == 0) {
 			synchronized (this) {
 				if (this.size == 0) {
-					if (logger.isDebugEnabled()) {
+					if (LOGGER.isDebugEnabled()) {
-						logger.debug("Lazy loading of group size for messageGroup: " + this.original.getGroupId());
+						LOGGER.debug("Lazy loading of group size for messageGroup: " + this.original.getGroupId());
 					}
 					this.size = this.messageGroupStore.messageGroupSize(this.original.getGroupId());
 				}
@@ -180,8 +180,8 @@ class PersistentMessageGroup implements MessageGroup {
 				synchronized (this) {
 					if (this.collection == null) {
 						Object groupId = PersistentMessageGroup.this.original.getGroupId();
-						if (logger.isDebugEnabled()) {
+						if (LOGGER.isDebugEnabled()) {
-							logger.debug("Lazy loading of messages for messageGroup: " + groupId);
+							LOGGER.debug("Lazy loading of messages for messageGroup: " + groupId);
 						}
 						this.collection = PersistentMessageGroup.this.messageGroupStore.getMessagesForGroup(groupId);
 					}

spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -41,11 +41,11 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Artem Bilan
  *
- * @see org.springframework.beans.factory.BeanFactory
+ * @see BeanFactory
  */
 public class BeanFactoryChannelResolver implements DestinationResolver<MessageChannel>, BeanFactoryAware {
 
-	private static final Log logger = LogFactory.getLog(BeanFactoryChannelResolver.class);
+	private static final Log LOGGER = LogFactory.getLog(BeanFactoryChannelResolver.class);
 
 	private BeanFactory beanFactory;
 
@@ -102,7 +102,7 @@ public class BeanFactoryChannelResolver implements DestinationResolver<MessageCh
 									HeaderChannelRegistry.class);
 						}
 						catch (Exception ex) {
-							logger.debug("No HeaderChannelRegistry found");
+							LOGGER.debug("No HeaderChannelRegistry found");
 						}
 						this.initialized = true;
 					}

spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonObjectMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2020 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.
@@ -43,7 +43,7 @@ import org.springframework.util.ClassUtils;
  */
 public abstract class AbstractJacksonJsonObjectMapper<N, P, J> implements JsonObjectMapper<N, P>, BeanClassLoaderAware {
 
-	protected static final Collection<Class<?>> supportedJsonTypes =
+	protected static final Collection<Class<?>> SUPPORTED_JSON_TYPES =
 			Arrays.asList(String.class, byte[].class, File.class, URL.class, InputStream.class, Reader.class);
 
 	private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader();

spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2020 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.
@@ -150,7 +150,7 @@ public class Jackson2JsonObjectMapper extends AbstractJacksonJsonObjectMapper<Js
 			return this.objectMapper.readValue((Reader) json, type);
 		}
 		else {
-			throw new IllegalArgumentException("'json' argument must be an instance of: " + supportedJsonTypes
+			throw new IllegalArgumentException("'json' argument must be an instance of: " + SUPPORTED_JSON_TYPES
 					+ " , but gotten: " + json.getClass());
 		}
 	}

spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonPresent.java

@@ -29,28 +29,14 @@ public final class JacksonPresent {
 
 	private static final ClassLoader CLASS_LOADER = ClassUtils.getDefaultClassLoader();
 
-	private static final boolean jackson2Present =
+	private static final boolean JACKSON_2_PRESENT =
 			ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper", CLASS_LOADER) &&
 					ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator", CLASS_LOADER);
 
-	private static final boolean jacksonPresent =
-			ClassUtils.isPresent("org.codehaus.jackson.map.ObjectMapper", CLASS_LOADER) &&
-					ClassUtils.isPresent("org.codehaus.jackson.JsonGenerator", CLASS_LOADER);
-
 	public static boolean isJackson2Present() {
-		return jackson2Present;
+		return JACKSON_2_PRESENT;
 	}
 
-	/**
-	 * @return true if Jackson 1.x is present on classpath
-	 * @deprecated Jackson 1.x is not supported any more. Use Jackson 2.x.
-	 */
-	@Deprecated
-	public static boolean isJacksonPresent() {
-		return jacksonPresent;
-	}
-
-
 	private JacksonPresent() {
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java

@@ -66,7 +66,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 
 	public static final long DEFAULT_BUSY_WAIT_TIME = 50L;
 
-	private static final Log logger = LogFactory.getLog(LockRegistryLeaderInitiator.class);
+	private static final Log LOGGER = LogFactory.getLog(LockRegistryLeaderInitiator.class);
 
 	private final Object lifecycleMonitor = new Object();
 
@@ -295,7 +295,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 				this.leaderSelector = new LeaderSelector(buildLeaderPath());
 				this.running = true;
 				this.future = this.executorService.submit(this.leaderSelector);
-				logger.debug("Started LeaderInitiator");
+				LOGGER.debug("Started LeaderInitiator");
 			}
 		}
 	}
@@ -321,7 +321,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 					this.future.cancel(true);
 				}
 				this.future = null;
-				logger.debug("Stopped LeaderInitiator for " + getContext());
+				LOGGER.debug("Stopped LeaderInitiator for " + getContext());
 			}
 		}
 	}
@@ -369,7 +369,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 						this.lock.unlock();
 					}
 					catch (Exception e) {
-						logger.debug("Could not unlock during stop for " + this.context
+						LOGGER.debug("Could not unlock during stop for " + this.context
 								+ " - treat as broken. Revoking...", e);
 					}
 					// We are stopping, therefore not leading any more
@@ -380,8 +380,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 		}
 
 		private void tryAcquireLock() throws InterruptedException {
-			if (logger.isDebugEnabled()) {
+			if (LOGGER.isDebugEnabled()) {
-				logger.debug("Acquiring the lock for " + this.context);
+				LOGGER.debug("Acquiring the lock for " + this.context);
 			}
 			// We always try to acquire the lock, in case it expired
 			boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis,
@@ -423,7 +423,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 					this.lock.unlock();
 				}
 				catch (Exception e1) {
-					logger.debug("Could not unlock - treat as broken " + this.context +
+					LOGGER.debug("Could not unlock - treat as broken " + this.context +
 							". Revoking " + (isRunning() ? " and retrying..." : "..."), e1);
 
 				}
@@ -449,8 +449,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 						Thread.currentThread().interrupt();
 					}
 				}
-				if (logger.isDebugEnabled()) {
+				if (LOGGER.isDebugEnabled()) {
-					logger.debug("Error acquiring the lock for " + this.context +
+					LOGGER.debug("Error acquiring the lock for " + this.context +
 							". " + (isRunning() ? "Retrying..." : ""), ex);
 				}
 			}
@@ -458,7 +458,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 		}
 
 		private void restartSelectorBecauseOfError(Exception ex) {
-			logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", ex);
+			LOGGER.warn("Restarting LeaderSelector for " + this.context + " because of error.", ex);
 			LockRegistryLeaderInitiator.this.future =
 					LockRegistryLeaderInitiator.this.executorService.submit(
 							() -> {
@@ -480,7 +480,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 							LockRegistryLeaderInitiator.this, this.context, this.lockKey);
 				}
 				catch (Exception e) {
-					logger.warn("Error publishing OnGranted event.", e);
+					LOGGER.warn("Error publishing OnGranted event.", e);
 				}
 			}
 		}
@@ -494,7 +494,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 							LockRegistryLeaderInitiator.this.candidate.getRole());
 				}
 				catch (Exception e) {
-					logger.warn("Error publishing OnRevoked event.", e);
+					LOGGER.warn("Error publishing OnRevoked event.", e);
 				}
 			}
 		}
@@ -508,7 +508,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 							LockRegistryLeaderInitiator.this.candidate.getRole());
 				}
 				catch (Exception e) {
-					logger.warn("Error publishing OnFailedToAcquire event.", e);
+					LOGGER.warn("Error publishing OnFailedToAcquire event.", e);
 				}
 			}
 		}
@@ -530,8 +530,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 
 		@Override
 		public void yield() {
-			if (logger.isDebugEnabled()) {
+			if (LOGGER.isDebugEnabled()) {
-				logger.debug("Yielding leadership from " + this);
+				LOGGER.debug("Yielding leadership from " + this);
 			}
 			if (LockRegistryLeaderInitiator.this.future != null) {
 				LockRegistryLeaderInitiator.this.future.cancel(true);

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java

@@ -37,6 +37,7 @@ import org.springframework.util.Assert;
  * Pollable channel backed by a Kafka topic.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -68,8 +69,9 @@ public class PollableKafkaChannel extends AbstractKafkaChannel
 
 	private static String topic(KafkaMessageSource<?, ?> source) {
 		Assert.notNull(source, "'source' cannot be null");
-		Assert.isTrue(source.getConsumerProperties().getTopics().length == 1, "Only one topic is allowed");
+		String[] topics = source.getConsumerProperties().getTopics();
-		return source.getConsumerProperties().getTopics()[0];
+		Assert.isTrue(topics != null && topics.length == 1, "Only one topic is allowed");
+		return topics[0];
 	}
 
 	@Override

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java

@@ -19,7 +19,6 @@ package org.springframework.integration.kafka.channel;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
-import org.springframework.context.Phased;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
@@ -37,6 +36,7 @@ import org.springframework.util.Assert;
  * Subscribable channel backed by a Kafka topic.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -79,7 +79,7 @@ public class SubscribableKafkaChannel extends AbstractKafkaChannel implements Su
 	/**
 	 * Set the phase.
 	 * @param phase the phase.
-	 * @see Phased
+	 * @see org.springframework.context.Phased
 	 */
 	public void setPhase(int phase) {
 		this.phase = phase;

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java

@@ -18,8 +18,6 @@ package org.springframework.integration.kafka.dsl;
 
 import java.util.regex.Pattern;
 
-import org.apache.kafka.common.TopicPartition;
-
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory;
@@ -225,7 +223,7 @@ public final class Kafka {
 	 * Create an initial
 	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
 	 * @param consumerFactory the {@link ConsumerFactory}.
-	 * @param topicPartitions the {@link TopicPartition} vararg.
+	 * @param topicPartitions the {@link TopicPartitionOffset} vararg.
 	 * @param <K> the Kafka message key type.
 	 * @param <V> the Kafka message value type.
 	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
@@ -244,7 +242,7 @@ public final class Kafka {
 	 * {@link KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec}.
 	 * @param consumerFactory the {@link ConsumerFactory}.
 	 * @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
-	 * @param topicPartitions the {@link TopicPartition} vararg.
+	 * @param topicPartitions the {@link TopicPartitionOffset} vararg.
 	 * @param <K> the Kafka message key type.
 	 * @param <V> the Kafka message value type.
 	 * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -16,16 +16,11 @@
 
 package org.springframework.integration.kafka.dsl;
 
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.dsl.MessageSourceSpec;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource.KafkaAckCallbackFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.ConsumerProperties;
-import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 
 /**
@@ -36,6 +31,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter;
  *
  * @author Gary Russell
  * @author Anshul Mehra
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -72,7 +68,7 @@ public class KafkaInboundChannelAdapterSpec<K, V>
 
 	/**
 	 * Set the message converter to replace the default.
-	 * {@link MessagingMessageConverter}.
+	 * {@link org.springframework.kafka.support.converter.MessagingMessageConverter}.
 	 * @param messageConverter the converter.
 	 * @return the spec.
 	 */
@@ -93,9 +89,9 @@ public class KafkaInboundChannelAdapterSpec<K, V>
 	}
 
 	/**
-	 * Set to true to include the raw {@link ConsumerRecord} as headers with keys
+	 * Set to true to include the raw {@link org.apache.kafka.clients.consumer.ConsumerRecord} as headers with keys
-	 * {@link KafkaHeaders#RAW_DATA} and
+	 * {@link org.springframework.kafka.support.KafkaHeaders#RAW_DATA} and
-	 * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. enabling callers to have
+	 * {@link org.springframework.integration.IntegrationMessageHeaderAccessor#SOURCE_DATA}. enabling callers to have
 	 * access to the record to process errors.
 	 * @param rawMessageHeader true to include the header.
 	 * @return the spec.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -29,7 +29,6 @@ import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
 import org.springframework.integration.support.ObjectStringMapBuilder;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
-import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ConsumerSeekAware;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.retry.RecoveryCallback;
@@ -116,8 +115,8 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
 	}
 
 	/**
-	 * A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaInboundGatewaySpec}
+	 * A {@link org.springframework.kafka.listener.ConcurrentMessageListenerContainer}
-	 * extension.
+	 * configuration {@link KafkaInboundGatewaySpec} extension.
 	 * @param <K> the key type.
 	 * @param <V> the request value type.
 	 * @param <R> the reply value type.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -27,7 +27,6 @@ import org.springframework.integration.dsl.ComponentsRegistration;
 import org.springframework.integration.dsl.MessageProducerSpec;
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
-import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ConsumerSeekAware;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.support.converter.BatchMessageConverter;
@@ -193,8 +192,8 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
 	}
 
 	/**
-	 * A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaMessageDrivenChannelAdapterSpec}
+	 * A {@link org.springframework.kafka.listener.ConcurrentMessageListenerContainer} configuration
-	 * extension.
+	 * {@link KafkaMessageDrivenChannelAdapterSpec} extension.
 	 * @param <K> the key type.
 	 * @param <V> the value type.
 	 */

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java

@@ -24,11 +24,8 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.springframework.core.task.AsyncListenableTaskExecutor;
 import org.springframework.integration.dsl.IntegrationComponentSpec;
 import org.springframework.kafka.core.ConsumerFactory;
-import org.springframework.kafka.listener.AbstractMessageListenerContainer;
-import org.springframework.kafka.listener.AcknowledgingMessageListener;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ContainerProperties;
-import org.springframework.kafka.listener.ErrorHandler;
 import org.springframework.kafka.listener.GenericErrorHandler;
 import org.springframework.kafka.support.TopicPartitionOffset;
 
@@ -68,12 +65,13 @@ public class KafkaMessageListenerContainerSpec<K, V>
 	}
 
 	@Override
-	public KafkaMessageListenerContainerSpec<K, V> id(String id) {
+	public KafkaMessageListenerContainerSpec<K, V> id(String id) { // NOSONAR - increase visibility
 		return super.id(id);
 	}
 
 	/**
-	 * Specify a concurrency maximum number for the {@link AbstractMessageListenerContainer}.
+	 * Specify a concurrency maximum number for the
+	 * {@link org.springframework.kafka.listener.AbstractMessageListenerContainer}.
 	 * @param concurrency the concurrency maximum number.
 	 * @return the spec.
 	 * @see ConcurrentMessageListenerContainer#setConcurrency(int)
@@ -84,10 +82,11 @@ public class KafkaMessageListenerContainerSpec<K, V>
 	}
 
 	/**
-	 * Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}.
+	 * Specify an {@link org.springframework.kafka.listener.ErrorHandler} for the
-	 * @param errorHandler the {@link ErrorHandler}.
+	 * {@link org.springframework.kafka.listener.AbstractMessageListenerContainer}.
+	 * @param errorHandler the {@link org.springframework.kafka.listener.ErrorHandler}.
 	 * @return the spec.
-	 * @see ErrorHandler
+	 * @see org.springframework.kafka.listener.ErrorHandler
 	 */
 	public KafkaMessageListenerContainerSpec<K, V> errorHandler(GenericErrorHandler<?> errorHandler) {
 		this.target.setGenericErrorHandler(errorHandler);
@@ -104,7 +103,7 @@ public class KafkaMessageListenerContainerSpec<K, V>
 	 * {@code #setPollTimeout(long) pollTimeout}.</li>
 	 * <li>COUNT: Ack after at least this number of records have been received</li>
 	 * <li>MANUAL: Listener is responsible for acking - use a
-	 * {@link AcknowledgingMessageListener}.
+	 * {@link org.springframework.kafka.listener.AcknowledgingMessageListener}.
 	 * </ul>
 	 * @param ackMode the {@link ContainerProperties.AckMode}; default BATCH.
 	 * @return the spec.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -22,10 +22,6 @@ import java.util.Map;
 import java.util.function.Consumer;
 
 import org.springframework.integration.dsl.ComponentsRegistration;
-import org.springframework.integration.dsl.IntegrationComponentSpec;
-import org.springframework.integration.dsl.MessageHandlerSpec;
-import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
-import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.GenericMessageListenerContainer;
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
@@ -34,7 +30,8 @@ import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 
 /**
- * A {@link MessageHandlerSpec} implementation for the {@link KafkaProducerMessageHandler}
+ * A {@link org.springframework.integration.dsl.MessageHandlerSpec}
+ * implementation for the {@link org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler}
  * as a gateway.
  * @param <K> the key type.
  * @param <V> the outbound value type.
@@ -64,7 +61,7 @@ public class KafkaOutboundGatewaySpec<K, V, R, S extends KafkaOutboundGatewaySpe
 	}
 
 	/**
-	 * A {@link KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension.
+	 * A {@link org.springframework.kafka.core.KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension.
 	 *
 	 * @param <K> the key type.
 	 * @param <V> the outbound value type.
@@ -107,7 +104,8 @@ public class KafkaOutboundGatewaySpec<K, V, R, S extends KafkaOutboundGatewaySpe
 	}
 
 	/**
-	 * An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}.
+	 * An {@link org.springframework.integration.dsl.IntegrationComponentSpec}
+	 * implementation for the {@link org.springframework.kafka.core.KafkaTemplate}.
 	 *
 	 * @param <K> the key type.
 	 * @param <V> the request value type.

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -19,7 +19,6 @@ package org.springframework.integration.kafka.dsl;
 import org.springframework.integration.dsl.IntegrationComponentSpec;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
-import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 
@@ -68,7 +67,7 @@ public class KafkaTemplateSpec<K, V>
 
 	/**
 	 * Set a {@link ProducerListener} which will be invoked when Kafka acknowledges
-	 * a send operation. By default a {@link LoggingProducerListener} is configured
+	 * a send operation. By default a {@link org.springframework.kafka.support.LoggingProducerListener} is configured
 	 * which logs errors only.
 	 * @param producerListener the listener; may be {@code null}.
 	 * @return the spec

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java

@@ -72,7 +72,7 @@ import org.springframework.util.Assert;
  */
 public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implements OrderlyShutdownCapable {
 
-	private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
+	private static final ThreadLocal<AttributeAccessor> ATTRIBUTES_HOLDER = new ThreadLocal<>();
 
 	private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener();
 
@@ -82,7 +82,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 
 	private RetryTemplate retryTemplate;
 
-	private RecoveryCallback<? extends Object> recoveryCallback;
+	private RecoveryCallback<?> recoveryCallback;
 
 	private BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback;
 
@@ -143,7 +143,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 	 * Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified.
 	 * @param recoveryCallback the recovery callback.
 	 */
-	public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
+	public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
 		this.recoveryCallback = recoveryCallback;
 	}
 
@@ -224,10 +224,10 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 		boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
 		boolean needAttributes = needHolder | this.retryTemplate != null;
 		if (needHolder) {
-			attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
+			ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null));
 		}
 		if (needAttributes) {
-			AttributeAccessor attributes = attributesHolder.get();
+			AttributeAccessor attributes = ATTRIBUTES_HOLDER.get();
 			if (attributes != null) {
 				attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
 				attributes.setAttribute(KafkaHeaders.RAW_DATA, record);
@@ -237,7 +237,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 
 	@Override
 	protected AttributeAccessor getErrorMessageAttributes(Message<?> message) {
-		AttributeAccessor attributes = attributesHolder.get();
+		AttributeAccessor attributes = ATTRIBUTES_HOLDER.get();
 		if (attributes == null) {
 			return super.getErrorMessageAttributes(message);
 		}
@@ -283,7 +283,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 				}
 				finally {
 					if (KafkaInboundGateway.this.retryTemplate == null) {
-						attributesHolder.remove();
+						ATTRIBUTES_HOLDER.remove();
 					}
 				}
 			}
@@ -298,7 +298,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 				Map<String, Object> rawHeaders = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders();
 				if (KafkaInboundGateway.this.retryTemplate != null) {
 					AtomicInteger deliveryAttempt =
-							new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
+							new AtomicInteger(((RetryContext) ATTRIBUTES_HOLDER.get()).getRetryCount() + 1);
 					rawHeaders.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
 				}
 				else if (KafkaInboundGateway.this.containerDeliveryAttemptPresent) {
@@ -314,7 +314,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 				MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
 				if (KafkaInboundGateway.this.retryTemplate != null) {
 					AtomicInteger deliveryAttempt =
-							new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
+							new AtomicInteger(((RetryContext) ATTRIBUTES_HOLDER.get()).getRetryCount() + 1);
 					builder.setHeader(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
 				}
 				else if (KafkaInboundGateway.this.containerDeliveryAttemptPresent) {
@@ -362,7 +362,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 		@Override
 		public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
 			if (KafkaInboundGateway.this.retryTemplate != null) {
-				attributesHolder.set(context);
+				ATTRIBUTES_HOLDER.set(context);
 			}
 			return true;
 		}
@@ -371,7 +371,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
 		public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
 				Throwable throwable) {
 
-			attributesHolder.remove();
+			ATTRIBUTES_HOLDER.remove();
 		}
 
 		@Override

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java

@@ -33,7 +33,6 @@ import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.integration.core.Pausable;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
-import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageUtils;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
@@ -54,6 +53,7 @@ import org.springframework.kafka.support.converter.KafkaMessageHeaders;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryCallback;
@@ -78,7 +78,7 @@ import org.springframework.util.Assert;
 public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable,
 		Pausable {
 
-	private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
+	private static final ThreadLocal<AttributeAccessor> ATTRIBUTES_HOLDER = new ThreadLocal<>();
 
 	private final AbstractMessageListenerContainer<K, V> messageListenerContainer;
 
@@ -354,7 +354,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 	 * If there's a retry template, it will set the attributes holder via the listener. If
 	 * there's no retry template, but there's an error channel, we create a new attributes
 	 * holder here. If an attributes holder exists (by either method), we set the
-	 * attributes for use by the {@link ErrorMessageStrategy}.
+	 * attributes for use by the {@link org.springframework.integration.support.ErrorMessageStrategy}.
 	 * @param record the record.
 	 * @param message the message.
 	 * @since 2.1.1
@@ -363,10 +363,10 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 		boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
 		boolean needAttributes = needHolder | this.retryTemplate != null;
 		if (needHolder) {
-			attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
+			ATTRIBUTES_HOLDER.set(ErrorMessageUtils.getAttributeAccessor(null, null));
 		}
 		if (needAttributes) {
-			AttributeAccessor attributes = attributesHolder.get();
+			AttributeAccessor attributes = ATTRIBUTES_HOLDER.get();
 			if (attributes != null) {
 				attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
 				attributes.setAttribute(KafkaHeaders.RAW_DATA, record);
@@ -376,7 +376,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 
 	@Override
 	protected AttributeAccessor getErrorMessageAttributes(Message<?> message) {
-		AttributeAccessor attributes = attributesHolder.get();
+		AttributeAccessor attributes = ATTRIBUTES_HOLDER.get();
 		if (attributes == null) {
 			return super.getErrorMessageAttributes(message);
 		}
@@ -392,7 +392,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 			}
 			finally {
 				if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
-					attributesHolder.remove();
+					ATTRIBUTES_HOLDER.remove();
 				}
 			}
 		}
@@ -456,7 +456,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 				Map<String, Object> rawHeaders = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders();
 				if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) {
 					AtomicInteger deliveryAttempt =
-							new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
+							new AtomicInteger(((RetryContext) ATTRIBUTES_HOLDER.get()).getRetryCount() + 1);
 					rawHeaders.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
 				}
 				else if (KafkaMessageDrivenChannelAdapter.this.containerDeliveryAttemptPresent) {
@@ -472,7 +472,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 				MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
 				if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) {
 					AtomicInteger deliveryAttempt =
-							new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
+							new AtomicInteger(((RetryContext) ATTRIBUTES_HOLDER.get()).getRetryCount() + 1);
 					builder.setHeader(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
 				}
 				else if (KafkaMessageDrivenChannelAdapter.this.containerDeliveryAttemptPresent) {
@@ -491,7 +491,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 		@Override
 		public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
 			if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) {
-				attributesHolder.set(context);
+				ATTRIBUTES_HOLDER.set(context);
 			}
 			return true;
 		}
@@ -500,7 +500,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 		public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
 				Throwable throwable) {
 
-			attributesHolder.remove();
+			ATTRIBUTES_HOLDER.remove();
 		}
 
 		@Override
@@ -536,8 +536,9 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 			}
 			catch (RuntimeException e) {
 				Exception exception = new ConversionException("Failed to convert to message for: " + records, e);
-				if (getErrorChannel() != null) {
+				MessageChannel errorChannel = getErrorChannel();
-					getMessagingTemplate().send(getErrorChannel(), new ErrorMessage(exception));
+				if (errorChannel != null) {
+					getMessagingTemplate().send(errorChannel, new ErrorMessage(exception));
 				}
 			}
 
@@ -547,7 +548,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 		@Override
 		public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
 			if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) {
-				attributesHolder.set(context);
+				ATTRIBUTES_HOLDER.set(context);
 			}
 			return true;
 		}
@@ -556,7 +557,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 		public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
 				Throwable throwable) {
 
-			attributesHolder.remove();
+			ATTRIBUTES_HOLDER.remove();
 		}
 
 		@Override

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -30,7 +30,7 @@ import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Supplier;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
@@ -107,9 +107,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	 */
 	public static final String REMAINING_RECORDS = KafkaHeaders.PREFIX + "remainingRecords";
 
-	private final Supplier<Duration> minTimeoutProvider =
-			() -> Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT));
-
 	private final ConsumerFactory<K, V> consumerFactory;
 
 	private final KafkaAckCallbackFactory<K, V> ackCallbackFactory;
@@ -124,7 +121,11 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private final Collection<TopicPartition> assignedPartitions = new LinkedHashSet<>();
 
-	private Duration pollTimeout;
+	private final Duration commitTimeout;
+
+	private final Duration assignTimeout;
+
+	private final Duration pollTimeout;
 
 	private RecordMessageConverter messageConverter = new MessagingMessageConverter();
 
@@ -132,12 +133,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private boolean rawMessageHeader;
 
-	private Duration commitTimeout;
-
 	private boolean running;
 
-	private Duration assignTimeout;
-
 	private volatile Consumer<K, V> consumer;
 
 	private volatile boolean pausing;
@@ -231,7 +228,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		this.consumerFactory = fixOrRejectConsumerFactory(consumerFactory, allowMultiFetch);
 		this.ackCallbackFactory = ackCallbackFactory;
 		this.pollTimeout = Duration.ofMillis(consumerProperties.getPollTimeout());
-		this.assignTimeout = this.minTimeoutProvider.get();
+		this.assignTimeout =
+				Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT)); // NOSONAR - magic
 		this.commitTimeout = consumerProperties.getSyncCommitTimeout();
 	}
 
@@ -431,7 +429,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 			this.logger.debug("Consumer is paused; no records will be returned");
 		}
 		ConsumerRecord<K, V> record;
-		TopicPartition topicPartition;
 		if (this.recordsIterator != null) {
 			record = nextRecord();
 		}
@@ -447,125 +444,30 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 				record = nextRecord();
 			}
 		}
-		topicPartition = new TopicPartition(record.topic(), record.partition());
+		return recordToMessage(record);
-		KafkaAckInfo<K, V> ackInfo = new KafkaAckInfoImpl(record, topicPartition);
-		AcknowledgmentCallback ackCallback = this.ackCallbackFactory.createCallback(ackInfo);
-		this.inflightRecords.computeIfAbsent(topicPartition, tp -> Collections.synchronizedSet(new TreeSet<>()))
-				.add(ackInfo);
-		Message<?> message = this.messageConverter.toMessage(record,
-				ackCallback instanceof Acknowledgment ? (Acknowledgment) ackCallback : null, this.consumer,
-				this.payloadType);
-		if (message.getHeaders() instanceof KafkaMessageHeaders) {
-			Map<String, Object> rawHeaders = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders();
-			rawHeaders.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, ackCallback);
-			rawHeaders.put(REMAINING_RECORDS, this.remainingCount.get());
-			if (this.rawMessageHeader) {
-				rawHeaders.put(KafkaHeaders.RAW_DATA, record);
-				rawHeaders.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, record);
-			}
-			return message;
-		}
-		else {
-			AbstractIntegrationMessageBuilder<?> builder = getMessageBuilderFactory().fromMessage(message)
-					.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, ackCallback)
-					.setHeader(REMAINING_RECORDS, this.remainingCount.get());
-			if (this.rawMessageHeader) {
-				builder.setHeader(KafkaHeaders.RAW_DATA, record);
-				builder.setHeader(IntegrationMessageHeaderAccessor.SOURCE_DATA, record);
-			}
-			return builder;
-		}
-	}
-
-	private ConsumerRecord<K, V> nextRecord() {
-		ConsumerRecord<K, V> record;
-		record = this.recordsIterator.next();
-		if (!this.recordsIterator.hasNext()) {
-			this.recordsIterator = null;
-		}
-		this.remainingCount.decrementAndGet();
-		return record;
 	}
 
 	protected void createConsumer() {
 		synchronized (this.consumerMonitor) {
 			this.consumer = this.consumerFactory.createConsumer(this.consumerProperties.getGroupId(),
 					this.consumerProperties.getClientId(), null, this.consumerProperties.getKafkaConsumerProperties());
-			ConsumerRebalanceListener providedRebalanceListener = this.consumerProperties
-					.getConsumerRebalanceListener();
-			boolean isConsumerAware = providedRebalanceListener instanceof ConsumerAwareRebalanceListener;
-			ConsumerRebalanceListener rebalanceCallback = new ConsumerRebalanceListener() {
 
-				@Override
+			ConsumerRebalanceListener rebalanceCallback =
-				public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+					new ItegrationConsumerRebalanceListener(this.consumerProperties.getConsumerRebalanceListener());
-					KafkaMessageSource.this.assignedPartitions.removeAll(partitions);
-					if (KafkaMessageSource.this.logger.isInfoEnabled()) {
-						KafkaMessageSource.this.logger
-								.info("Partitions revoked: " + partitions);
-					}
-					if (providedRebalanceListener != null) {
-						if (isConsumerAware) {
-							((ConsumerAwareRebalanceListener) providedRebalanceListener)
-									.onPartitionsRevokedAfterCommit(KafkaMessageSource.this.consumer, partitions);
-						}
-						else {
-							providedRebalanceListener.onPartitionsRevoked(partitions);
-						}
-					}
 
-				}
+			Pattern topicPattern = this.consumerProperties.getTopicPattern();
-
+			TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitions();
-				@Override
+			if (topicPattern != null) {
-				public void onPartitionsLost(Collection<TopicPartition> partitions) {
+				this.consumer.subscribe(topicPattern, rebalanceCallback);
-					if (providedRebalanceListener != null) {
-						if (isConsumerAware) {
-							((ConsumerAwareRebalanceListener) providedRebalanceListener).onPartitionsLost(partitions);
-						}
-						else {
-							providedRebalanceListener.onPartitionsLost(partitions);
-						}
-					}
-					onPartitionsRevoked(partitions);
-				}
-
-				@Override
-				public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-					KafkaMessageSource.this.assignedPartitions.addAll(partitions);
-					if (KafkaMessageSource.this.paused) {
-						KafkaMessageSource.this.consumer.pause(KafkaMessageSource.this.assignedPartitions);
-						KafkaMessageSource.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; "
-								+ "consumer paused again, so the initial poll() will never return any records");
-					}
-					if (KafkaMessageSource.this.logger.isInfoEnabled()) {
-						KafkaMessageSource.this.logger
-								.info("Partitions assigned: " + partitions);
-					}
-					if (providedRebalanceListener != null) {
-						if (isConsumerAware) {
-							((ConsumerAwareRebalanceListener) providedRebalanceListener)
-									.onPartitionsAssigned(KafkaMessageSource.this.consumer, partitions);
-						}
-						else {
-							providedRebalanceListener.onPartitionsAssigned(partitions);
-						}
-					}
-				}
-
-			};
-
-			if (this.consumerProperties.getTopicPattern() != null) {
-				this.consumer.subscribe(this.consumerProperties.getTopicPattern(), rebalanceCallback);
 			}
-			else if (this.consumerProperties.getTopicPartitions() != null) {
+			else if (partitions != null) {
-				List<TopicPartition> topicPartitionsToAssign = Arrays
+				List<TopicPartition> topicPartitionsToAssign =
-						.stream(this.consumerProperties.getTopicPartitions())
+						Arrays.stream(partitions)
-						.map(TopicPartitionOffset::getTopicPartition)
+								.map(TopicPartitionOffset::getTopicPartition)
-						.collect(Collectors.toList());
+								.collect(Collectors.toList());
 				this.consumer.assign(topicPartitionsToAssign);
 				this.assignedPartitions.addAll(topicPartitionsToAssign);
 
-				TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitions();
-
 				for (TopicPartitionOffset partition : partitions) {
 					if (TopicPartitionOffset.SeekPosition.BEGINNING.equals(partition.getPosition())) {
 						this.consumer.seekToBeginning(Collections.singleton(partition.getTopicPartition()));
@@ -603,11 +505,53 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 				}
 			}
 			else {
-				this.consumer.subscribe(Arrays.asList(this.consumerProperties.getTopics()), rebalanceCallback);
+				this.consumer.subscribe(Arrays.asList(this.consumerProperties.getTopics()), // NOSONAR
+						rebalanceCallback);
 			}
 		}
 	}
 
+	private ConsumerRecord<K, V> nextRecord() {
+		ConsumerRecord<K, V> record;
+		record = this.recordsIterator.next();
+		if (!this.recordsIterator.hasNext()) {
+			this.recordsIterator = null;
+		}
+		this.remainingCount.decrementAndGet();
+		return record;
+	}
+
+	private Object recordToMessage(ConsumerRecord<K, V> record) {
+		TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
+		KafkaAckInfo<K, V> ackInfo = new KafkaAckInfoImpl(record, topicPartition);
+		AcknowledgmentCallback ackCallback = this.ackCallbackFactory.createCallback(ackInfo);
+		this.inflightRecords.computeIfAbsent(topicPartition, tp -> Collections.synchronizedSet(new TreeSet<>()))
+				.add(ackInfo);
+		Message<?> message = this.messageConverter.toMessage(record,
+				ackCallback instanceof Acknowledgment ? (Acknowledgment) ackCallback : null, this.consumer,
+				this.payloadType);
+		if (message.getHeaders() instanceof KafkaMessageHeaders) {
+			Map<String, Object> rawHeaders = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders();
+			rawHeaders.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, ackCallback);
+			rawHeaders.put(REMAINING_RECORDS, this.remainingCount.get());
+			if (this.rawMessageHeader) {
+				rawHeaders.put(KafkaHeaders.RAW_DATA, record);
+				rawHeaders.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, record);
+			}
+			return message;
+		}
+		else {
+			AbstractIntegrationMessageBuilder<?> builder = getMessageBuilderFactory().fromMessage(message)
+					.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, ackCallback)
+					.setHeader(REMAINING_RECORDS, this.remainingCount.get());
+			if (this.rawMessageHeader) {
+				builder.setHeader(KafkaHeaders.RAW_DATA, record);
+				builder.setHeader(IntegrationMessageHeaderAccessor.SOURCE_DATA, record);
+			}
+			return builder;
+		}
+	}
+
 	@Override
 	public synchronized void destroy() {
 		stopConsumer();
@@ -623,6 +567,67 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		}
 	}
 
+	private class ItegrationConsumerRebalanceListener implements ConsumerRebalanceListener {
+
+		private final ConsumerRebalanceListener providedRebalanceListener;
+
+		private final boolean isConsumerAware;
+
+		ItegrationConsumerRebalanceListener(ConsumerRebalanceListener providedRebalanceListener) {
+			this.providedRebalanceListener = providedRebalanceListener;
+			this.isConsumerAware = providedRebalanceListener instanceof ConsumerAwareRebalanceListener;
+		}
+
+		@Override
+		public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+			KafkaMessageSource.this.assignedPartitions.removeAll(partitions);
+			if (KafkaMessageSource.this.logger.isInfoEnabled()) {
+				KafkaMessageSource.this.logger.info("Partitions revoked: " + partitions);
+			}
+			if (this.providedRebalanceListener != null) {
+				if (this.isConsumerAware) {
+					((ConsumerAwareRebalanceListener) this.providedRebalanceListener)
+							.onPartitionsRevokedAfterCommit(KafkaMessageSource.this.consumer, partitions);
+				}
+				else {
+					this.providedRebalanceListener.onPartitionsRevoked(partitions);
+				}
+			}
+
+		}
+
+		@Override
+		public void onPartitionsLost(Collection<TopicPartition> partitions) {
+			if (this.providedRebalanceListener != null) {
+				this.providedRebalanceListener.onPartitionsLost(partitions);
+			}
+			onPartitionsRevoked(partitions);
+		}
+
+		@Override
+		public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+			KafkaMessageSource.this.assignedPartitions.addAll(partitions);
+			if (KafkaMessageSource.this.paused) {
+				KafkaMessageSource.this.consumer.pause(KafkaMessageSource.this.assignedPartitions);
+				KafkaMessageSource.this.logger.warn("Paused consumer resumed by Kafka due to rebalance; "
+						+ "consumer paused again, so the initial poll() will never return any records");
+			}
+			if (KafkaMessageSource.this.logger.isInfoEnabled()) {
+				KafkaMessageSource.this.logger.info("Partitions assigned: " + partitions);
+			}
+			if (this.providedRebalanceListener != null) {
+				if (this.isConsumerAware) {
+					((ConsumerAwareRebalanceListener) this.providedRebalanceListener)
+							.onPartitionsAssigned(KafkaMessageSource.this.consumer, partitions);
+				}
+				else {
+					this.providedRebalanceListener.onPartitionsAssigned(partitions);
+				}
+			}
+		}
+
+	}
+
 	/**
 	 * AcknowledgmentCallbackFactory for KafkaAckInfo.
 	 * @param <K> the key type.
@@ -693,7 +698,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 					consumerProperties != null
 							? consumerProperties.getCommitLogLevel()
 							: LogIfLevelEnabled.Level.DEBUG);
-			this.logOnlyMetadata = consumerProperties.isOnlyLogRecordMetadata();
+			this.logOnlyMetadata = consumerProperties != null && consumerProperties.isOnlyLogRecordMetadata();
 		}
 
 		@Override
@@ -751,7 +756,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 			}
 		}
 
-		private void commitIfPossible(ConsumerRecord<K, V> record) {
+		private void commitIfPossible(ConsumerRecord<K, V> record) { // NOSONAR
 			if (this.ackInfo.isRolledBack()) {
 				if (this.logger.isWarnEnabled()) {
 					this.logger.warn("Cannot commit offset for "
@@ -783,7 +788,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 									+ ListenerUtils.recordToString(record, this.logOnlyMetadata)
 									+ " and all deferred to "
 									+ ListenerUtils.recordToString(ackInformationToLog.getRecord(),
-											this.logOnlyMetadata));
+									this.logOnlyMetadata));
 							candidates.removeAll(toCommit);
 						}
 						else {
@@ -845,7 +850,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	/**
 	 * Information for building an KafkaAckCallback.
 	 */
-	public class KafkaAckInfoImpl implements KafkaAckInfo<K, V> {
+	public class KafkaAckInfoImpl implements KafkaAckInfo<K, V> { // NOSONAR - no equals() impl
 
 		private final ConsumerRecord<K, V> record;
 

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

@@ -66,7 +66,6 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -151,7 +150,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	private ProducerRecordCreator<K, V> producerRecordCreator =
 			(message, topic, partition, timestamp, key, value, headers) ->
-				new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
+					new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
 
 	private volatile byte[] singleReplyTopic;
 
@@ -162,7 +161,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		if (this.isGateway) {
 			setAsync(true);
 			updateNotPropagatedHeaders(
-					new String[] { KafkaHeaders.TOPIC, KafkaHeaders.PARTITION_ID, KafkaHeaders.MESSAGE_KEY }, false);
+					new String[]{KafkaHeaders.TOPIC, KafkaHeaders.PARTITION_ID, KafkaHeaders.MESSAGE_KEY}, false);
 		}
 		if (JacksonPresent.isJackson2Present()) {
 			this.headerMapper = new DefaultKafkaHeaderMapper();
@@ -270,7 +269,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 	}
 
 	/**
-	 * Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent
+	 * Set the failure channel. After a send failure, an
+	 * {@link org.springframework.messaging.support.ErrorMessage} will be sent
 	 * to this channel with a payload of a {@link KafkaSendFailureException} with the
 	 * failed message and cause.
 	 * @param sendFailureChannel the failure channel.
@@ -281,7 +281,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 	}
 
 	/**
-	 * Set the failure channel name. After a send failure, an {@link ErrorMessage} will be
+	 * Set the failure channel name. After a send failure, an
+	 * {@link org.springframework.messaging.support.ErrorMessage} will be
 	 * sent to this channel name with a payload of a {@link KafkaSendFailureException}
 	 * with the failed message and cause.
 	 * @param sendFailureChannelName the failure channel name.
@@ -392,10 +393,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	@Override
 	public void stop() {
-		if (this.running.compareAndSet(true, false)) {
+		if (this.running.compareAndSet(true, false) && (!this.transactional || this.allowNonTransactional)) {
-			if (!this.transactional || this.allowNonTransactional) {
+			this.kafkaTemplate.flush();
-				this.kafkaTemplate.flush();
-			}
 		}
 	}
 
@@ -404,11 +403,12 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		return this.running.get();
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings("unchecked") // NOSONAR - complexity
 	@Override
 	protected Object handleRequestMessage(final Message<?> message) {
 		final ProducerRecord<K, V> producerRecord;
-		boolean flush = this.flushExpression.getValue(this.evaluationContext, message, Boolean.class);
+		boolean flush =
+				Boolean.TRUE.equals(this.flushExpression.getValue(this.evaluationContext, message, Boolean.class));
 		boolean preBuilt = message.getPayload() instanceof ProducerRecord;
 		if (preBuilt) {
 			producerRecord = (ProducerRecord<K, V>) message.getPayload();
@@ -430,9 +430,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 			if (this.transactional
 					&& TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null
 					&& !this.allowNonTransactional) {
-				sendFuture = this.kafkaTemplate.executeInTransaction(template -> {
+				sendFuture = this.kafkaTemplate.executeInTransaction(template -> template.send(producerRecord));
-					return template.send(producerRecord);
-				});
 			}
 			else {
 				sendFuture = this.kafkaTemplate.send(producerRecord);
@@ -446,7 +444,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 			throw new MessageHandlingException(message, e);
 		}
 		catch (ExecutionException e) {
-			throw new MessageHandlingException(message, e.getCause());
+			throw new MessageHandlingException(message, e.getCause()); // NOSONAR
 		}
 		if (flush) {
 			this.kafkaTemplate.flush();
@@ -488,12 +486,11 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 			headers = new RecordHeaders();
 			this.headerMapper.fromHeaders(messageHeaders, headers);
 		}
-		final ProducerRecord<K, V> producerRecord = this.producerRecordCreator.create(message, topic, partitionId,
+		return this.producerRecordCreator.create(message, topic, partitionId, timestamp, (K) messageKey, payload,
-				timestamp, (K) messageKey, payload, headers);
+				headers);
-		return producerRecord;
 	}
 
-	private byte[] getReplyTopic(final Message<?> message) {
+	private byte[] getReplyTopic(Message<?> message) { // NOSONAR
 		if (this.replyTopicsAndPartitions.isEmpty()) {
 			determineValidReplyTopicsAndPartitions();
 		}
@@ -569,8 +566,9 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 			ListenableFuture<SendResult<K, V>> future, MessageChannel metadataChannel)
 			throws InterruptedException, ExecutionException {
 
-		if (getSendFailureChannel() != null || metadataChannel != null) {
+		final MessageChannel sendFailureChannel = getSendFailureChannel();
-			future.addCallback(new ListenableFutureCallback<SendResult<K, V>>() {
+		if (sendFailureChannel != null || metadataChannel != null) {
+			future.addCallback(new ListenableFutureCallback<SendResult<K, V>>() { // NOSONAR
 
 				@Override
 				public void onSuccess(SendResult<K, V> result) {
@@ -583,8 +581,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 				@Override
 				public void onFailure(Throwable ex) {
-					if (getSendFailureChannel() != null) {
+					if (sendFailureChannel != null) {
-						KafkaProducerMessageHandler.this.messagingTemplate.send(getSendFailureChannel(),
+						KafkaProducerMessageHandler.this.messagingTemplate.send(sendFailureChannel,
 								KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage(
 										new KafkaSendFailureException(message, producerRecord, ex), null));
 					}
@@ -623,7 +621,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		}
 
 		private void addCallback(final RequestReplyFuture<?, ?, Object> future) {
-			future.addCallback(new ListenableFutureCallback<ConsumerRecord<?, Object>>() {
+			future.addCallback(new ListenableFutureCallback<ConsumerRecord<?, Object>>() { // NOSONAR
 
 				@Override
 				public void onSuccess(ConsumerRecord<?, Object> result) {

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -251,13 +251,13 @@ class MessageSourceTests {
 			return null;
 		}).given(consumer).commitAsync(any(), any());
 		Map<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
-		records1.put(topicPartition, Arrays.asList(
+		records1.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
 		Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
-		records2.put(topicPartition, Arrays.asList(
+		records2.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "bar")));
 		Map<TopicPartition, List<ConsumerRecord>> records3 = new LinkedHashMap<>();
-		records3.put(topicPartition, Arrays.asList(
+		records3.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 2L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "baz")));
 		Map<TopicPartition, List<ConsumerRecord>> records4 = new LinkedHashMap<>();
 		records4.put(topicPartition, Collections.singletonList(

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java

@@ -23,8 +23,6 @@ import org.reactivestreams.Publisher;
 
 import org.springframework.data.r2dbc.core.DatabaseClient;
 import org.springframework.data.r2dbc.core.FetchSpec;
-import org.springframework.data.r2dbc.core.R2dbcEntityOperations;
-import org.springframework.data.relational.core.query.Query;
 import org.springframework.expression.Expression;
 import org.springframework.expression.TypeLocator;
 import org.springframework.expression.common.LiteralExpression;
@@ -39,15 +37,13 @@ import reactor.core.publisher.Mono;
 /**
  * An instance of {@link org.springframework.integration.core.MessageSource} which returns
  * a {@link org.springframework.messaging.Message} with a payload which is the result of
- * execution of a {@link Query}. When {@code expectSingleResult} is false (default), the R2dbc
+ * execution of query. When {@code expectSingleResult} is false (default), the R2DBC
- * {@link Query} is executed using {@link R2dbcEntityOperations#select(Query, Class)} method which
+ * query is executed returning a {@link reactor.core.publisher.Flux}.
- * returns a {@link reactor.core.publisher.Flux}.
  * The returned {@link reactor.core.publisher.Flux} will be used as the payload of the
  * {@link org.springframework.messaging.Message} returned by the {@link #receive()}
  * method.
  * <p>
- * When {@code expectSingleResult} is true, the {@link R2dbcEntityOperations#selectOne(Query, Class)} is
+ * When {@code expectSingleResult} is true, the query is executed returning a {@link reactor.core.publisher.Mono}
- * used instead, and the message payload will be a {@link reactor.core.publisher.Mono}
  * for the single object returned from the query.
  *
  * @author Rohan Mukesh
@@ -108,7 +104,7 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 	}
 
 	/**
-	 * Provide a way to manage which find* method to invoke on {@link R2dbcEntityOperations}.
+	 * Provide a way to return all the records matching criteria or only and only a one otherwise.
 	 * Default is 'false', which means the {@link #receive()} method will use
 	 * the {@link org.springframework.data.r2dbc.core.DatabaseClient#execute(String)} method and will fetch all. If set
 	 * to 'true'{@link #receive()} will use {@link org.springframework.data.r2dbc.core.DatabaseClient#execute(String)}
@@ -140,7 +136,7 @@ public class R2dbcMessageSource extends AbstractMessageSource<Publisher<?>> {
 	}
 
 	/**
-	 * Execute a {@link Query} returning its results as the Message payload.
+	 * Execute a query returning its results as the Message payload.
 	 * The payload can be either {@link reactor.core.publisher.Flux} or
 	 * {@link reactor.core.publisher.Mono} of objects of type identified by {@link #payloadType},
 	 * or a single element of type identified by {@link #payloadType}

spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java

@@ -204,13 +204,16 @@ public class R2dbcMessageHandler extends AbstractReactiveMessageHandler {
 	}
 
 	private String evaluateTableNameExpression(Message<?> message) {
-		String tableName = this.tableNameExpression.getValue(this.evaluationContext, message, String.class);
+		String tableName =
+				this.tableNameExpression.getValue(this.evaluationContext, message, String.class); // NOSONAR
 		Assert.notNull(tableName, "'tableNameExpression' must not evaluate to null");
 		return tableName;
 	}
 
 	@SuppressWarnings("unchecked")
 	private Map<String, Object> evaluateValuesExpression(Message<?> message) {
+		Assert.notNull(this.valuesExpression,
+				"'this.valuesExpression' must not be null when 'tableNameExpression' mode is used");
 		Map<String, Object> fieldValues =
 				(Map<String, Object>) this.valuesExpression.getValue(this.evaluationContext, message, Map.class);
 		Assert.notNull(fieldValues, "'valuesExpression' must not evaluate to null");
@@ -218,6 +221,8 @@ public class R2dbcMessageHandler extends AbstractReactiveMessageHandler {
 	}
 
 	private Criteria evaluateCriteriaExpression(Message<?> message) {
+		Assert.notNull(this.criteriaExpression,
+				"'this.criteriaExpression' must not be null when 'tableNameExpression' mode is used");
 		Criteria criteria =
 				this.criteriaExpression.getValue(this.evaluationContext, message, Criteria.class);
 		Assert.notNull(criteria, "'criteriaExpression' must not evaluate to null");