diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java index 21c9ffa311..e16028cc5e 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java @@ -31,6 +31,8 @@ import org.springframework.messaging.Message; import org.springframework.util.Assert; /** + * The base {@link AbstractMessageChannel} implementation for AMQP. + * * @author Mark Fisher * @author Artem Bilan * @author Gary Russell diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java index 038d9914a3..e400278523 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java @@ -45,6 +45,8 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; /** + * The base {@link AbstractAmqpChannel} extension for a {@link SubscribableChannel} contract. + * * @author Mark Fisher * @author Gary Russell * @author Artem Bilan diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java index b41262efa1..85f2689308 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java @@ -26,6 +26,12 @@ import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrateg import org.springframework.integration.dispatcher.UnicastingDispatcher; /** + * The {@link AbstractSubscribableAmqpChannel} implementation for one-to-one subscription + * over AMQP queue. + *

+ * If queue name is not provided, the channel bean name is used internally to declare + * a queue via provided {@link AmqpAdmin} (if any). + * * @author Mark Fisher * @author Artem Bilan * @@ -45,6 +51,7 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq */ public PointToPointSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container, AmqpTemplate amqpTemplate) { + super(channelName, container, amqpTemplate); } @@ -62,6 +69,7 @@ public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmq */ public PointToPointSubscribableAmqpChannel(String channelName, AbstractMessageListenerContainer container, AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + super(channelName, container, amqpTemplate, outboundMapper, inboundMapper); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 2e8d6bda89..c34afa1981 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -87,6 +87,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel */ public PollableAmqpChannel(String channelName, AmqpTemplate amqpTemplate, AmqpHeaderMapper outboundMapper, AmqpHeaderMapper inboundMapper) { + super(amqpTemplate, outboundMapper, inboundMapper); Assert.hasText(channelName, "channel name must not be empty"); this.channelName = channelName; diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java index 79f164dc47..29216f9bff 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java @@ -25,6 +25,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; +import org.springframework.integration.channel.DirectChannel; import org.springframework.util.StringUtils; /** @@ -32,6 +33,7 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan * * @since 2.1 */ @@ -42,7 +44,9 @@ public class AmqpInboundChannelAdapterParser extends AbstractAmqpInboundAdapterP } @Override - protected final String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) throws BeanDefinitionStoreException { + protected final String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) + throws BeanDefinitionStoreException { + String id = element.getAttribute("id"); if (!element.hasAttribute("channel")) { // the created channel will get the 'id', so the adapter's bean name includes a suffix @@ -69,8 +73,7 @@ public class AmqpInboundChannelAdapterParser extends AbstractAmqpInboundAdapterP parserContext.getReaderContext().error("The channel-adapter's 'id' attribute is required when no 'channel' " + "reference has been provided, because that 'id' would be used for the created channel.", element); } - BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.channel.DirectChannel"); + BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class); BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId); BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry()); return channelId; diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java index 007e64d505..2da2a1020c 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java @@ -23,19 +23,20 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa * * @author Mark Fisher * @author Gary Russell + * * @since 2.1 */ public class AmqpNamespaceHandler extends AbstractIntegrationNamespaceHandler { @Override public void init() { - this.registerBeanDefinitionParser("channel", new AmqpChannelParser()); - this.registerBeanDefinitionParser("publish-subscribe-channel", new AmqpChannelParser()); - this.registerBeanDefinitionParser("inbound-channel-adapter", new AmqpInboundChannelAdapterParser()); - this.registerBeanDefinitionParser("inbound-gateway", new AmqpInboundGatewayParser()); - this.registerBeanDefinitionParser("outbound-channel-adapter", new AmqpOutboundChannelAdapterParser()); - this.registerBeanDefinitionParser("outbound-gateway", new AmqpOutboundGatewayParser()); - this.registerBeanDefinitionParser("outbound-async-gateway", new AmqpOutboundGatewayParser()); + registerBeanDefinitionParser("channel", new AmqpChannelParser()); + registerBeanDefinitionParser("publish-subscribe-channel", new AmqpChannelParser()); + registerBeanDefinitionParser("inbound-channel-adapter", new AmqpInboundChannelAdapterParser()); + registerBeanDefinitionParser("inbound-gateway", new AmqpInboundGatewayParser()); + registerBeanDefinitionParser("outbound-channel-adapter", new AmqpOutboundChannelAdapterParser()); + registerBeanDefinitionParser("outbound-gateway", new AmqpOutboundGatewayParser()); + registerBeanDefinitionParser("outbound-async-gateway", new AmqpOutboundGatewayParser()); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java index 8d8982176b..3bc3e68399 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -35,6 +35,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author Gary Russell * @author Artem Bilan + * * @since 2.1 */ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java index ce5558c8ff..2e60c34f59 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java @@ -191,7 +191,7 @@ public abstract class AbstractMessageListenerContainerSpec When false, the condition is not considered fatal and the container will @@ -316,7 +316,7 @@ public abstract class AbstractMessageListenerContainerSpec pollableChannel(ConnectionFactory connectionFactory) { + public static AmqpPollableMessageChannelSpec pollableChannel( + ConnectionFactory connectionFactory) { + return pollableChannel(null, connectionFactory); } @@ -284,7 +286,8 @@ public final class Amqp { public static AmqpPollableMessageChannelSpec pollableChannel(@Nullable String id, ConnectionFactory connectionFactory) { - AmqpPollableMessageChannelSpec spec = new AmqpPollableMessageChannelSpec<>(connectionFactory); + AmqpPollableMessageChannelSpec spec = + new AmqpPollableMessageChannelSpec<>(connectionFactory); return spec.id(id); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index c1f0e1ff98..c05211b949 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -54,6 +54,8 @@ import org.springframework.util.StringUtils; import org.springframework.util.concurrent.SettableListenableFuture; /** + * A base {@link AbstractReplyProducingMessageHandler} extension for AMQP message handlers. + * * @author Gary Russell * @author Artem Bilan * diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java index 58d7e63526..c73ed68c73 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java @@ -49,7 +49,7 @@ public class AmqpMessageHeaderErrorMessageStrategy implements ErrorMessageStrate public ErrorMessage buildErrorMessage(Throwable throwable, @Nullable AttributeAccessor context) { Object inputMessage = context == null ? null : context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY); - Map headers = new HashMap(); + Map headers = new HashMap<>(); if (context != null) { headers.put(AMQP_RAW_MESSAGE, context.getAttribute(AMQP_RAW_MESSAGE)); headers.put(IntegrationMessageHeaderAccessor.SOURCE_DATA, context.getAttribute(AMQP_RAW_MESSAGE)); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdvice.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdvice.java index ed7d0189cf..8cd42a0703 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdvice.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdvice.java @@ -35,7 +35,7 @@ import com.rabbitmq.client.ConfirmCallback; * An advice that causes all downstream {@link RabbitOperations} operations to be executed * on the same channel, as long as there are no thread handoffs, since the channel is * bound to the thread. The same RabbitOperations must be used in this and all downstream - * components. Typically used with a splitter or some other mechanism that would cause + * components. Typically, used with a splitter or some other mechanism that would cause * multiple messages to be sent. Optionally waits for publisher confirms if the channel is * so configured. * 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 91a3a978e1..df5a238e0d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -79,7 +79,6 @@ public final class Kafka { * @param the Kafka message key type. * @param the Kafka message value type. * @return the spec. - * @since 3.2 */ public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, ConsumerProperties consumerProperties) { @@ -96,7 +95,6 @@ public final class Kafka { * @param the Kafka message key type. * @param the Kafka message value type. * @return the spec. - * @since 3.2 */ public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, @@ -115,7 +113,6 @@ public final class Kafka { * @param the Kafka message key type. * @param the Kafka message value type. * @return the spec. - * @since 3.2 */ public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, @@ -135,7 +132,6 @@ public final class Kafka { * @param the Kafka message key type. * @param the Kafka message value type. * @return the spec. - * @since 3.2 */ public static KafkaInboundChannelAdapterSpec inboundChannelAdapter( ConsumerFactory consumerFactory, @@ -341,7 +337,6 @@ public final class Kafka { * @param the Kafka message value type (request). * @param the Kafka message value type (reply). * @return the KafkaGatewayMessageHandlerSpec. - * @since 3.0.2 */ public static KafkaOutboundGatewaySpec outboundGateway( ReplyingKafkaTemplate kafkaTemplate) { @@ -357,7 +352,6 @@ public final class Kafka { * @param the Kafka message value type (request). * @param the Kafka message value type (reply). * @return the KafkaGatewayMessageHandlerSpec. - * @since 3.0.2 */ public static KafkaOutboundGatewaySpec.KafkaGatewayMessageHandlerTemplateSpec outboundGateway( ProducerFactory producerFactory, GenericMessageListenerContainer replyContainer) { @@ -378,7 +372,6 @@ public final class Kafka { * @param the Kafka message value type (request). * @param the Kafka message value type (reply). * @return the spec. - * @since 3.0.2 */ public static KafkaInboundGatewaySpec inboundGateway( AbstractMessageListenerContainer container, KafkaTemplate template) { @@ -396,7 +389,6 @@ public final class Kafka { * @param the Kafka message value type (request). * @param the Kafka message value type (reply). * @return the spec. - * @since 3.0.2 */ public static KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec inboundGateway( ConsumerFactory consumerFactory, ContainerProperties containerProperties, @@ -416,7 +408,6 @@ public final class Kafka { * @param the Kafka message value type (request). * @param the Kafka message value type (reply). * @return the spec. - * @since 3.0.2 */ public static KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec inboundGateway( KafkaMessageListenerContainerSpec containerSpec, KafkaTemplateSpec templateSpec) { @@ -430,7 +421,6 @@ public final class Kafka { * @param containerFactory the container factory. * @param topic the topic. * @return the spec. - * @since 3.3 */ public static KafkaPointToPointChannelSpec channel(KafkaTemplate template, KafkaListenerContainerFactory containerFactory, String topic) { @@ -439,12 +429,11 @@ public final class Kafka { } /** - * Create a spec for a publish/subscribe channel with the provided parameters. + * Create a spec for a publish-subscribe channel with the provided parameters. * @param template the template. * @param containerFactory the container factory. * @param topic the topic. * @return the spec. - * @since 3.3 */ public static KafkaPublishSubscribeChannelSpec publishSubscribeChannel(KafkaTemplate template, KafkaListenerContainerFactory containerFactory, String topic) { 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 f02536eab2..b266f56924 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-2020 the original author or authors. + * Copyright 2018-2021 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. @@ -90,7 +90,7 @@ public class KafkaInboundGatewaySpec recoveryCallback) { + public S recoveryCallback(RecoveryCallback recoveryCallback) { this.target.setRecoveryCallback(recoveryCallback); return _this(); } @@ -101,7 +101,6 @@ public class KafkaInboundGatewaySpec, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { 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 c963363b51..144e06a1d1 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-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -136,7 +136,7 @@ public class KafkaMessageDrivenChannelAdapterSpec recoveryCallback) { + public S recoveryCallback(RecoveryCallback recoveryCallback) { this.target.setRecoveryCallback(recoveryCallback); return _this(); } @@ -146,7 +146,6 @@ public class KafkaMessageDrivenChannelAdapterSpec payloadType) { this.target.setPayloadType(payloadType); @@ -178,7 +177,6 @@ public class KafkaMessageDrivenChannelAdapterSpec, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { 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 73e41fdf28..f10a168e47 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2021 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. @@ -204,7 +204,7 @@ public class KafkaMessageListenerContainerSpec } /** - * Set whether or not to call consumer.commitSync() or commitAsync() when the + * Set whether to call consumer.commitSync() or commitAsync() when the * container is responsible for commits. Default true. See * https://github.com/spring-projects/spring-kafka/issues/62 At the time of * writing, async commits are not entirely reliable. 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 1add5f11b0..c243b1a82d 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-2020 the original author or authors. + * Copyright 2018-2021 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. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index cda980f045..9127d51895 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -251,7 +251,7 @@ public class KafkaProducerMessageHandlerSpec * {@code * .flush(m -> m.getPayload().shouldFlush()) @@ -314,7 +314,6 @@ public class KafkaProducerMessageHandlerSpec extends MessagingGatewaySupport implem } /** - * When using a type-aware message converter (such as {@code StringJsonMessageConverter}, + * When using a type-aware message converter such as {@code StringJsonMessageConverter}, * set the payload type the converter should create. Defaults to {@link Object}. * @param payloadType the type. */ @@ -154,7 +154,6 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem * This is called from the internal * {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation. * @param onPartitionsAssignedCallback the {@link BiConsumer} to use - * @since 3.0.4 * @see ConsumerSeekAware#onPartitionsAssigned */ public void setOnPartitionsAssignedSeekCallback( @@ -166,14 +165,12 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem * Set to true to bind the source consumer record in the header named * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. * @param bindSourceRecord true to bind. - * @since 3.1.4 */ public void setBindSourceRecord(boolean bindSourceRecord) { this.bindSourceRecord = bindSourceRecord; } @Override - @SuppressWarnings("deprecation") protected void onInit() { super.onInit(); MessageListener kafkaListener = this.listener; 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 45f9401004..03feaaef95 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 @@ -117,7 +117,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * Construct an instance with the provided mode. * @param messageListenerContainer the container. * @param mode the mode. - * @since 1.2 */ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer messageListenerContainer, ListenerMode mode) { @@ -153,7 +152,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo /** * Set the message converter to use with a record-based consumer. * @param messageConverter the converter. - * @since 2.1 */ public void setRecordMessageConverter(RecordMessageConverter messageConverter) { this.recordListener.setMessageConverter(messageConverter); @@ -162,7 +160,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo /** * Set the message converter to use with a batch-based consumer. * @param messageConverter the converter. - * @since 2.1 */ public void setBatchMessageConverter(BatchMessageConverter messageConverter) { this.batchListener.setBatchMessageConverter(messageConverter); @@ -173,7 +170,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * {@link KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into * {@link FilteringMessageListenerAdapter}. * @param recordFilterStrategy the {@link RecordFilterStrategy} to use. - * @since 2.0.1 */ public void setRecordFilterStrategy(RecordFilterStrategy recordFilterStrategy) { this.recordFilterStrategy = recordFilterStrategy; @@ -184,7 +180,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * should acknowledge discarded records or not. * Does not make sense if {@link #setRecordFilterStrategy(RecordFilterStrategy)} isn't specified. * @param ackDiscarded true to ack (commit offset for) discarded messages. - * @since 2.0.1 */ public void setAckDiscarded(boolean ackDiscarded) { this.ackDiscarded = ackDiscarded; @@ -195,7 +190,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * {@link KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into * {@code RetryingMessageListenerAdapter}. * @param retryTemplate the {@link RetryTemplate} to use. - * @since 2.0.1 */ public void setRetryTemplate(RetryTemplate retryTemplate) { Assert.isTrue(retryTemplate == null || this.mode.equals(ListenerMode.record), @@ -209,7 +203,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * (unless an error channel is configured). * Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified. * @param recoveryCallback the recovery callback. - * @since 2.0.1 */ public void setRecoveryCallback(RecoveryCallback recoveryCallback) { this.recoveryCallback = recoveryCallback; @@ -224,17 +217,15 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * {@link RecordFilterStrategy} is present, or any. * @param filterInRetry the order for {@code RetryingMessageListenerAdapter} and * {@link FilteringMessageListenerAdapter} wrapping. Defaults to {@code false}. - * @since 2.0.1 */ public void setFilterInRetry(boolean filterInRetry) { this.filterInRetry = filterInRetry; } /** - * When using a type-aware message converter (such as {@code StringJsonMessageConverter}, + * When using a type-aware message converter such as {@code StringJsonMessageConverter}, * set the payload type the converter should create. Defaults to {@link Object}. * @param payloadType the type. - * @since 2.1.1 */ public void setPayloadType(Class payloadType) { this.recordListener.setFallbackType(payloadType); @@ -248,7 +239,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * This is called from the internal * {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation. * @param onPartitionsAssignedCallback the {@link BiConsumer} to use - * @since 3.0.4 * @see ConsumerSeekAware#onPartitionsAssigned */ public void setOnPartitionsAssignedSeekCallback( @@ -261,7 +251,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * {@link IntegrationMessageHeaderAccessor#SOURCE_DATA}. * Does not apply to batch listeners. * @param bindSourceRecord true to bind. - * @since 3.1.4 */ public void setBindSourceRecord(boolean bindSourceRecord) { this.bindSourceRecord = bindSourceRecord; @@ -273,7 +262,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } @Override - @SuppressWarnings("deprecation") protected void onInit() { super.onInit(); @@ -364,7 +352,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo * attributes for use by the {@link org.springframework.integration.support.ErrorMessageStrategy}. * @param record the record. * @param message the message. - * @since 2.1.1 */ private void setAttributesIfNecessary(Object record, Message message) { boolean needHolder = getErrorChannel() != null && this.retryTemplate == null; @@ -411,8 +398,6 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo /** * The listener mode for the container, record or batch. - * @since 1.2 - * */ public enum ListenerMode { 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 e706b28f96..ea63edbe8c 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 @@ -155,7 +155,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl * records per poll will be disabled. * @param consumerFactory the consumer factory. * @param consumerProperties the consumer properties. - * @since 3.2 * @see #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean) */ public KafkaMessageSource(ConsumerFactory consumerFactory, ConsumerProperties consumerProperties) { @@ -175,7 +174,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl * @param consumerFactory the consumer factory. * @param consumerProperties the consumer properties. * @param allowMultiFetch true to allow {@code max.poll.records > 1}. - * @since 3.2 */ public KafkaMessageSource(ConsumerFactory consumerFactory, ConsumerProperties consumerProperties, @@ -190,7 +188,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl * @param consumerFactory the consumer factory. * @param consumerProperties the consumer properties. * @param ackCallbackFactory the ack callback factory. - * @since 3.2 * @see #KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean) */ public KafkaMessageSource(ConsumerFactory consumerFactory, @@ -210,12 +207,10 @@ public class KafkaMessageSource extends AbstractMessageSource impl * within {@code max.poll.interval.ms}. When false, you must call {@link #receive()} * within {@code max.poll.interval.ms}. {@link #pause()} will not take effect until * the records from the previous poll are consumed. - * * @param consumerFactory the consumer factory. * @param consumerProperties the consumer properties. * @param ackCallbackFactory the ack callback factory. * @param allowMultiFetch true to allow {@code max.poll.records > 1}. - * @since 3.2 */ public KafkaMessageSource(ConsumerFactory consumerFactory, ConsumerProperties consumerProperties, @@ -243,7 +238,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl /** * Return the currently assigned partitions. * @return the partitions. - * @since 3.2.2 */ public Collection getAssignedPartitions() { return Collections.unmodifiableCollection(this.assignedPartitions); @@ -260,7 +254,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl * Get a reference to the configured consumer properties; allows further * customization of the properties before the source is started. * @return the properties. - * @since 3.2 */ public ConsumerProperties getConsumerProperties() { return this.consumerProperties; 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 f3a8e9d07e..e0d36b12fa 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 @@ -273,7 +273,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Specify a timeout in milliseconds for how long this - * {@link KafkaProducerMessageHandler} should wait wait for send operation results. + * {@link KafkaProducerMessageHandler} should wait for send operation results. * Defaults to the kafka {@code delivery.timeout.ms} property + 5 seconds. The timeout * is applied Also applies when sending to the success or failure channels. * @param sendTimeout the timeout to wait for result for a send operation. @@ -286,7 +286,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Specify a SpEL expression to evaluate a timeout in milliseconds for how long this - * {@link KafkaProducerMessageHandler} should wait wait for send operation results. + * {@link KafkaProducerMessageHandler} should wait for send operation results. * Defaults to the kafka {@code delivery.timeout.ms} property + 5 seconds. The timeout * is applied only in {@link #sync} mode. If this expression yields a result that is * less than that value, the higher value is used. @@ -618,7 +618,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes if (topicToCheck == null) { topicToCheck = new String(replyTopic, StandardCharsets.UTF_8); } - if (!this.replyTopicsAndPartitions.keySet().contains(topicToCheck)) { + if (!this.replyTopicsAndPartitions.containsKey(topicToCheck)) { throw new IllegalStateException("The reply topic header [" + topicToCheck + "] does not match any reply container topic: " + this.replyTopicsAndPartitions.keySet()); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index e00c922449..a806ad89b5 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -58,15 +58,11 @@ These let you specify `topic`, `message-key`, and `partition-id`, respectively, IMPORTANT: The `KafkaHeaders` interface (provided by `spring-kafka`) contains constants used for interacting with headers. The `messageKey` and `topic` default headers now require a `kafka_` prefix. -When migrating from an earlier version that used the old headers, you need to specify -`message-key-expression="headers['messageKey']"` and `topic-expression="headers['topic']"` on the -``. -Alternatively, you can change the headers upstream to -the new headers from `KafkaHeaders` by using a `` or a `MessageBuilder`. +When migrating from an earlier version that used the old headers, you need to specify `message-key-expression="headers['messageKey']"` and `topic-expression="headers['topic']"` on the ``. +Alternatively, you can change the headers upstream to the new headers from `KafkaHeaders` by using a `` or a `MessageBuilder`. If you use constant values, you can also configure them on the adapter by using `topic` and `message-key`. -NOTE : If the adapter is configured with a topic or message key (either with a constant or expression), those are used -and the corresponding header is ignored. +NOTE : If the adapter is configured with a topic or message key (either with a constant or expression), those are used and the corresponding header is ignored. If you wish the header to override the configuration, you need to configure it in an expression, such as the following: ==== @@ -88,12 +84,12 @@ IMPORTANT: If your application uses transactions and the same channel adapter is The prefix used by container-initiated transactions (the producer factory or transaction manager property) must be the same on all application instances. The prefix used for producer-only transactions must be unique on all application instances. -Starting with version 3.3, you can configure a `flushExpression` which must resolve to a boolean value. +You can configure a `flushExpression` which must resolve to a boolean value. Flushing after sending several messages might be useful if you are using the `linger.ms` and `batch.size` Kafka producer properties; the expression should evaluate to `Boolean.TRUE` on the last message and an incomplete batch will be sent immediately. By default, the expression looks for a `Boolean` value in the `KafkaIntegrationHeaders.FLUSH` header (`kafka_flush`). The flush will occur if the value is `true` and not if it's `false` or the header is absent. -Starting with version 5.4, the `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed from 10 seconds to the `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework. +The `KafkaProducerMessageHandler.sendTimeoutExpression` default has changed from 10 seconds to the `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework. This has been changed for consistency because you may get unexpected behavior (Spring may timeout the send, while it is actually, eventually, successful). IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it to get more timely failures. @@ -302,7 +298,7 @@ public IntegrationFlow topic1ListenerFromKafkaFlow() { ---- ==== -Starting with Spring for Apache Kafka version 2.2, you can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes. +You can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes. See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] for an example. With the Java DSL, the container does not have to be configured as a `@Bean`, because the DSL registers the container as a bean. @@ -605,7 +601,7 @@ public IntegrationFlow serverGateway() { ---- ==== -Starting with Spring for Apache Kafka version 2.2, you can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes. +You can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes. See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] and <> for examples. ==== XML Configuration @@ -877,7 +873,7 @@ This functionality is supported by the underlying message listener container, to However, in order to support this, we need to block the listener thread until the success (or failure) of the write operation so that any exceptions can be thrown to the container. When consuming single records, this is achieved by setting the `sync` property on the outbound adapter. However, when consuming batches, using `sync` causes a significant performance degradation because the application would wait for the result of each send before sending the next message. -Starting with version 5.4, you can now perform multiple sends and then wait for the results of those sends afterwards. +You also can perform multiple sends and then wait for the results of those sends afterwards. This is achieved by adding a `futuresChannel` to the message handler. To enable the feature add `KafkaIntegrationHeaders.FUTURE_TOKEN` to the outbound messages; this can then be used to correlate a `Future` to a particular sent message. Here is an example of how you might use this feature: