From 6e11d4cdf97732c3c07fb749ae4fe89caff052d4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 13 Jul 2020 14:07:07 -0400 Subject: [PATCH] Fix new and some old Sonar smells --- .../filter/ExpressionEvaluatingSelector.java | 8 +- .../GatewayMethodInboundMessageMapper.java | 11 +- .../json/SimpleJsonSerializer.java | 10 +- .../router/RecipientListRouter.java | 14 +- .../scheduling/PollSkipAdvice.java | 8 +- .../store/PersistentMessageGroup.java | 16 +- .../channel/BeanFactoryChannelResolver.java | 8 +- .../json/AbstractJacksonJsonObjectMapper.java | 4 +- .../json/Jackson2JsonObjectMapper.java | 4 +- .../support/json/JacksonPresent.java | 18 +- .../leader/LockRegistryLeaderInitiator.java | 30 +-- .../kafka/channel/PollableKafkaChannel.java | 6 +- .../channel/SubscribableKafkaChannel.java | 4 +- .../integration/kafka/dsl/Kafka.java | 6 +- .../dsl/KafkaInboundChannelAdapterSpec.java | 16 +- .../kafka/dsl/KafkaInboundGatewaySpec.java | 7 +- .../KafkaMessageDrivenChannelAdapterSpec.java | 7 +- .../KafkaMessageListenerContainerSpec.java | 17 +- .../kafka/dsl/KafkaOutboundGatewaySpec.java | 14 +- .../kafka/dsl/KafkaTemplateSpec.java | 5 +- .../kafka/inbound/KafkaInboundGateway.java | 22 +- .../KafkaMessageDrivenChannelAdapter.java | 31 +-- .../kafka/inbound/KafkaMessageSource.java | 251 +++++++++--------- .../outbound/KafkaProducerMessageHandler.java | 46 ++-- .../kafka/inbound/MessageSourceTests.java | 8 +- .../r2dbc/inbound/R2dbcMessageSource.java | 14 +- .../r2dbc/outbound/R2dbcMessageHandler.java | 7 +- 27 files changed, 287 insertions(+), 305 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/ExpressionEvaluatingSelector.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/ExpressionEvaluatingSelector.java index a298c41c18..afca2436f2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/ExpressionEvaluatingSelector.java +++ b/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(expressionParser.parseExpression(expressionString), + super(new ExpressionEvaluatingMessageProcessor(EXPRESSION_PARSER.parseExpression(expressionString), Boolean.class)); this.expressionString = expressionString; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java index 6765e714c3..83ed90a9a3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java +++ b/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 headersToPopulate, Map argumentValue) { - if (messageOrPayload instanceof Map && !foundPayloadAnnotation) { - if (GatewayMethodInboundMessageMapper.this.payloadExpression == null) { - throw new MessagingException("Ambiguous method parameters; found more than one " + - "Map-typed parameter and neither one contains a @Payload annotation"); - } + if (messageOrPayload instanceof Map && !foundPayloadAnnotation + && GatewayMethodInboundMessageMapper.this.payloadExpression == null) { + throw new MessagingException("Ambiguous method parameters; found more than one " + + "Map-typed parameter and neither one contains a @Payload annotation"); } copyHeaders(argumentValue, headersToPopulate); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java b/spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java index 8547b7ed8d..1266de8b03 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/json/SimpleJsonSerializer.java +++ b/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()) { - logger.debug("Failed to serialize property " + propertyName, exception); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Failed to serialize property " + propertyName, exception); } result = diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java index bccac4c44e..2ecb7efc48 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java +++ b/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 (this.channelResolver != null) { - this.channel = this.channelResolver.resolveDestination(channelNameForInitialization); - this.channelName = null; - } + if (channelNameForInitialization != null && this.channelResolver != null) { + this.channel = this.channelResolver.resolveDestination(channelNameForInitialization); + this.channelName = null; } } return this.channel; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java index 51927f04de..94f9247025 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java +++ b/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()) { - logger.debug("Skipping poll because " + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Skipping poll because " + this.pollSkipStrategy.getClass().getName() + ".skipPoll() returned true"); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java index 533d029a50..a9827abf55 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java +++ b/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()) { - logger.debug("Lazy loading of one message for messageGroup: " + this.original.getGroupId()); + if (LOGGER.isDebugEnabled()) { + 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()) { - logger.debug("Lazy loading of group size for messageGroup: " + this.original.getGroupId()); + if (LOGGER.isDebugEnabled()) { + 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()) { - logger.debug("Lazy loading of messages for messageGroup: " + groupId); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Lazy loading of messages for messageGroup: " + groupId); } this.collection = PersistentMessageGroup.this.messageGroupStore.getMessagesForGroup(groupId); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java index 563b2b6319..cb83ff7836 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java +++ b/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, 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 implements JsonObjectMapper, BeanClassLoaderAware { - protected static final Collection> supportedJsonTypes = + protected static final Collection> SUPPORTED_JSON_TYPES = Arrays.asList(String.class, byte[].class, File.class, URL.class, InputStream.class, Reader.class); private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java b/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java index 7889629a57..0519008f2b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/json/Jackson2JsonObjectMapper.java +++ b/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 { @@ -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()) { - logger.debug("Yielding leadership from " + this); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Yielding leadership from " + this); } if (LockRegistryLeaderInitiator.this.future != null) { LockRegistryLeaderInitiator.this.future.cancel(true); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java index 85d4513ea2..342391f702 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java +++ b/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"); - return source.getConsumerProperties().getTopics()[0]; + String[] topics = source.getConsumerProperties().getTopics(); + Assert.isTrue(topics != null && topics.length == 1, "Only one topic is allowed"); + return topics[0]; } @Override diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java index a3adbe2b1d..92935aab3b 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java +++ b/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; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index 4006994878..6b432c2ac0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/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 the Kafka message key type. * @param 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 the Kafka message key type. * @param the Kafka message value type. * @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java index a3e37696bb..798d0ade52 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java +++ b/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 /** * 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 } /** - * Set to true to include the raw {@link ConsumerRecord} as headers with keys - * {@link KafkaHeaders#RAW_DATA} and - * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. enabling callers to have + * Set to true to include the raw {@link org.apache.kafka.clients.consumer.ConsumerRecord} as headers with keys + * {@link org.springframework.kafka.support.KafkaHeaders#RAW_DATA} and + * {@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. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java index 3a80bfa31e..f02536eab2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java +++ b/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 the key type. * @param the request value type. * @param the reply value type. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java index 4196e6789a..c963363b51 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java +++ b/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 the key type. * @param the value type. */ diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java index 13e57521cc..bd7b735285 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java +++ b/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 } @Override - public KafkaMessageListenerContainerSpec id(String id) { + public KafkaMessageListenerContainerSpec 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 } /** - * Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}. - * @param errorHandler the {@link ErrorHandler}. + * Specify an {@link org.springframework.kafka.listener.ErrorHandler} for the + * {@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 errorHandler(GenericErrorHandler errorHandler) { this.target.setGenericErrorHandler(errorHandler); @@ -104,7 +103,7 @@ public class KafkaMessageListenerContainerSpec * {@code #setPollTimeout(long) pollTimeout}. *
  • COUNT: Ack after at least this number of records have been received
    • *
  • MANUAL: Listener is responsible for acking - use a - * {@link AcknowledgingMessageListener}. + * {@link org.springframework.kafka.listener.AcknowledgingMessageListener}. * * @param ackMode the {@link ContainerProperties.AckMode}; default BATCH. * @return the spec. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java index bdf32545e2..1add5f11b0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java +++ b/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 the key type. * @param the outbound value type. @@ -64,7 +61,7 @@ public class KafkaOutboundGatewaySpec the key type. * @param the outbound value type. @@ -107,7 +104,8 @@ public class KafkaOutboundGatewaySpec the key type. * @param the request value type. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java index 9d6bbd15a1..2e40d31d1f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java +++ b/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 /** * 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 diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index 0dedba8223..b4b34d3063 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/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 extends MessagingGatewaySupport implements OrderlyShutdownCapable { - private static final ThreadLocal attributesHolder = new ThreadLocal<>(); + private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener(); @@ -82,7 +82,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem private RetryTemplate retryTemplate; - private RecoveryCallback recoveryCallback; + private RecoveryCallback recoveryCallback; private BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback; @@ -143,7 +143,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem * Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified. * @param recoveryCallback the recovery callback. */ - public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { this.recoveryCallback = recoveryCallback; } @@ -224,10 +224,10 @@ public class KafkaInboundGateway 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 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 extends MessagingGatewaySupport implem } finally { if (KafkaInboundGateway.this.retryTemplate == null) { - attributesHolder.remove(); + ATTRIBUTES_HOLDER.remove(); } } } @@ -298,7 +298,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem Map 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 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 extends MessagingGatewaySupport implem @Override public boolean open(RetryContext context, RetryCallback callback) { if (KafkaInboundGateway.this.retryTemplate != null) { - attributesHolder.set(context); + ATTRIBUTES_HOLDER.set(context); } return true; } @@ -371,7 +371,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem public void close(RetryContext context, RetryCallback callback, Throwable throwable) { - attributesHolder.remove(); + ATTRIBUTES_HOLDER.remove(); } @Override diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index d992775979..a59a6d8856 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/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 extends MessageProducerSupport implements OrderlyShutdownCapable, Pausable { - private static final ThreadLocal attributesHolder = new ThreadLocal<>(); + private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); private final AbstractMessageListenerContainer messageListenerContainer; @@ -354,7 +354,7 @@ public class KafkaMessageDrivenChannelAdapter 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 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 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 extends MessageProducerSuppo } finally { if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) { - attributesHolder.remove(); + ATTRIBUTES_HOLDER.remove(); } } } @@ -456,7 +456,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo Map 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 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 extends MessageProducerSuppo @Override public boolean open(RetryContext context, RetryCallback callback) { if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) { - attributesHolder.set(context); + ATTRIBUTES_HOLDER.set(context); } return true; } @@ -500,7 +500,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo public void close(RetryContext context, RetryCallback callback, Throwable throwable) { - attributesHolder.remove(); + ATTRIBUTES_HOLDER.remove(); } @Override @@ -536,8 +536,9 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } catch (RuntimeException e) { Exception exception = new ConversionException("Failed to convert to message for: " + records, e); - if (getErrorChannel() != null) { - getMessagingTemplate().send(getErrorChannel(), new ErrorMessage(exception)); + MessageChannel errorChannel = getErrorChannel(); + if (errorChannel != null) { + getMessagingTemplate().send(errorChannel, new ErrorMessage(exception)); } } @@ -547,7 +548,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo @Override public boolean open(RetryContext context, RetryCallback callback) { if (KafkaMessageDrivenChannelAdapter.this.retryTemplate != null) { - attributesHolder.set(context); + ATTRIBUTES_HOLDER.set(context); } return true; } @@ -556,7 +557,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo public void close(RetryContext context, RetryCallback callback, Throwable throwable) { - attributesHolder.remove(); + ATTRIBUTES_HOLDER.remove(); } @Override diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index f41e2646c3..a56d7f4990 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/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 extends AbstractMessageSource impl */ public static final String REMAINING_RECORDS = KafkaHeaders.PREFIX + "remainingRecords"; - private final Supplier minTimeoutProvider = - () -> Duration.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT)); - private final ConsumerFactory consumerFactory; private final KafkaAckCallbackFactory ackCallbackFactory; @@ -124,7 +121,11 @@ public class KafkaMessageSource extends AbstractMessageSource impl private final Collection 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 extends AbstractMessageSource impl private boolean rawMessageHeader; - private Duration commitTimeout; - private boolean running; - private Duration assignTimeout; - private volatile Consumer consumer; private volatile boolean pausing; @@ -231,7 +228,8 @@ public class KafkaMessageSource extends AbstractMessageSource 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 extends AbstractMessageSource impl this.logger.debug("Consumer is paused; no records will be returned"); } ConsumerRecord record; - TopicPartition topicPartition; if (this.recordsIterator != null) { record = nextRecord(); } @@ -447,125 +444,30 @@ public class KafkaMessageSource extends AbstractMessageSource impl record = nextRecord(); } } - topicPartition = new TopicPartition(record.topic(), record.partition()); - KafkaAckInfo 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 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 nextRecord() { - ConsumerRecord record; - record = this.recordsIterator.next(); - if (!this.recordsIterator.hasNext()) { - this.recordsIterator = null; - } - this.remainingCount.decrementAndGet(); - return record; + return recordToMessage(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 - public void onPartitionsRevoked(Collection partitions) { - 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); - } - } + ConsumerRebalanceListener rebalanceCallback = + new ItegrationConsumerRebalanceListener(this.consumerProperties.getConsumerRebalanceListener()); - } - - @Override - public void onPartitionsLost(Collection partitions) { - if (providedRebalanceListener != null) { - if (isConsumerAware) { - ((ConsumerAwareRebalanceListener) providedRebalanceListener).onPartitionsLost(partitions); - } - else { - providedRebalanceListener.onPartitionsLost(partitions); - } - } - onPartitionsRevoked(partitions); - } - - @Override - public void onPartitionsAssigned(Collection 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); + Pattern topicPattern = this.consumerProperties.getTopicPattern(); + TopicPartitionOffset[] partitions = this.consumerProperties.getTopicPartitions(); + if (topicPattern != null) { + this.consumer.subscribe(topicPattern, rebalanceCallback); } - else if (this.consumerProperties.getTopicPartitions() != null) { - List topicPartitionsToAssign = Arrays - .stream(this.consumerProperties.getTopicPartitions()) - .map(TopicPartitionOffset::getTopicPartition) - .collect(Collectors.toList()); + else if (partitions != null) { + List topicPartitionsToAssign = + Arrays.stream(partitions) + .map(TopicPartitionOffset::getTopicPartition) + .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 extends AbstractMessageSource impl } } else { - this.consumer.subscribe(Arrays.asList(this.consumerProperties.getTopics()), rebalanceCallback); + this.consumer.subscribe(Arrays.asList(this.consumerProperties.getTopics()), // NOSONAR + rebalanceCallback); } } } + private ConsumerRecord nextRecord() { + ConsumerRecord record; + record = this.recordsIterator.next(); + if (!this.recordsIterator.hasNext()) { + this.recordsIterator = null; + } + this.remainingCount.decrementAndGet(); + return record; + } + + private Object recordToMessage(ConsumerRecord record) { + TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition()); + KafkaAckInfo 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 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 extends AbstractMessageSource 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 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 partitions) { + if (this.providedRebalanceListener != null) { + this.providedRebalanceListener.onPartitionsLost(partitions); + } + onPartitionsRevoked(partitions); + } + + @Override + public void onPartitionsAssigned(Collection 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 the key type. @@ -693,7 +698,7 @@ public class KafkaMessageSource extends AbstractMessageSource 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 extends AbstractMessageSource impl } } - private void commitIfPossible(ConsumerRecord record) { + private void commitIfPossible(ConsumerRecord record) { // NOSONAR if (this.ackInfo.isRolledBack()) { if (this.logger.isWarnEnabled()) { this.logger.warn("Cannot commit offset for " @@ -783,7 +788,7 @@ public class KafkaMessageSource extends AbstractMessageSource 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 extends AbstractMessageSource impl /** * Information for building an KafkaAckCallback. */ - public class KafkaAckInfoImpl implements KafkaAckInfo { + public class KafkaAckInfoImpl implements KafkaAckInfo { // NOSONAR - no equals() impl private final ConsumerRecord record; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 0eac0b95b0..a2928c683d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/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 extends AbstractReplyProducingMes private ProducerRecordCreator 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 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 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 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 extends AbstractReplyProducingMes @Override public void stop() { - if (this.running.compareAndSet(true, false)) { - if (!this.transactional || this.allowNonTransactional) { - this.kafkaTemplate.flush(); - } + if (this.running.compareAndSet(true, false) && (!this.transactional || this.allowNonTransactional)) { + this.kafkaTemplate.flush(); } } @@ -404,11 +403,12 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes return this.running.get(); } - @SuppressWarnings("unchecked") + @SuppressWarnings("unchecked") // NOSONAR - complexity @Override protected Object handleRequestMessage(final Message message) { final ProducerRecord 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) message.getPayload(); @@ -430,9 +430,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes if (this.transactional && TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null && !this.allowNonTransactional) { - sendFuture = this.kafkaTemplate.executeInTransaction(template -> { - return template.send(producerRecord); - }); + sendFuture = this.kafkaTemplate.executeInTransaction(template -> template.send(producerRecord)); } else { sendFuture = this.kafkaTemplate.send(producerRecord); @@ -446,7 +444,7 @@ public class KafkaProducerMessageHandler 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 extends AbstractReplyProducingMes headers = new RecordHeaders(); this.headerMapper.fromHeaders(messageHeaders, headers); } - final ProducerRecord producerRecord = this.producerRecordCreator.create(message, topic, partitionId, - timestamp, (K) messageKey, payload, headers); - return producerRecord; + return this.producerRecordCreator.create(message, topic, partitionId, timestamp, (K) messageKey, payload, + headers); } - private byte[] getReplyTopic(final Message message) { + private byte[] getReplyTopic(Message message) { // NOSONAR if (this.replyTopicsAndPartitions.isEmpty()) { determineValidReplyTopicsAndPartitions(); } @@ -569,8 +566,9 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes ListenableFuture> future, MessageChannel metadataChannel) throws InterruptedException, ExecutionException { - if (getSendFailureChannel() != null || metadataChannel != null) { - future.addCallback(new ListenableFutureCallback>() { + final MessageChannel sendFailureChannel = getSendFailureChannel(); + if (sendFailureChannel != null || metadataChannel != null) { + future.addCallback(new ListenableFutureCallback>() { // NOSONAR @Override public void onSuccess(SendResult result) { @@ -583,8 +581,8 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes @Override public void onFailure(Throwable ex) { - if (getSendFailureChannel() != null) { - KafkaProducerMessageHandler.this.messagingTemplate.send(getSendFailureChannel(), + if (sendFailureChannel != null) { + KafkaProducerMessageHandler.this.messagingTemplate.send(sendFailureChannel, KafkaProducerMessageHandler.this.errorMessageStrategy.buildErrorMessage( new KafkaSendFailureException(message, producerRecord, ex), null)); } @@ -623,7 +621,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } private void addCallback(final RequestReplyFuture future) { - future.addCallback(new ListenableFutureCallback>() { + future.addCallback(new ListenableFutureCallback>() { // NOSONAR @Override public void onSuccess(ConsumerRecord result) { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index 0e97bb0da6..d0cba247a2 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/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> 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> 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> 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> records4 = new LinkedHashMap<>(); records4.put(topicPartition, Collections.singletonList( diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java index d003e26d5d..b6a1794c38 100644 --- a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSource.java +++ b/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 - * {@link Query} is executed using {@link R2dbcEntityOperations#select(Query, Class)} method which - * returns a {@link reactor.core.publisher.Flux}. + * execution of query. When {@code expectSingleResult} is false (default), the R2DBC + * query is executed returning 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. *

    - * When {@code expectSingleResult} is true, the {@link R2dbcEntityOperations#selectOne(Query, Class)} is - * used instead, and the message payload will be a {@link reactor.core.publisher.Mono} + * When {@code expectSingleResult} is true, the query is executed returning 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> { } /** - * 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> { } /** - * 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} diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java index 779defbb11..4172b0f454 100644 --- a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java +++ b/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 evaluateValuesExpression(Message message) { + Assert.notNull(this.valuesExpression, + "'this.valuesExpression' must not be null when 'tableNameExpression' mode is used"); Map fieldValues = (Map) 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");