Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Some docs clean up

Browse Source
This commit is contained in:
Artem Bilan
2021-11-15 10:00:14 -05:00
parent db611028da
commit 6f3fdc76c1
23 changed files with 64 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java
View File

@@ -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

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java
View File

@@ -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

8
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java
View File

@@ -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.
* <p>
* 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);
}

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
View File

@@ -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;

9
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java
View File

@@ -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;

15
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java
View File

@@ -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());
}
}

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -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 {

8
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java
View File

@@ -191,7 +191,7 @@ public abstract class AbstractMessageListenerContainerSpec<S extends AbstractMes
}
/**
* Determine whether or not the container should de-batch batched
* Determine whether the container should de-batch batched
* messages (true) or call the listener with the batch (false). Default: true.
* @param deBatchingEnabled the deBatchingEnabled to set.
* @return the spec.
@@ -300,7 +300,7 @@ public abstract class AbstractMessageListenerContainerSpec<S extends AbstractMes
}
/**
* If all of the configured queue(s) are not available on the broker, this setting
* If all the configured queue(s) are not available on the broker, this setting
* determines whether the condition is fatal. When true, and
* the queues are missing during startup, the context refresh() will fail.
* <p> When false, the condition is not considered fatal and the container will
@@ -316,7 +316,7 @@ public abstract class AbstractMessageListenerContainerSpec<S extends AbstractMes
/**
* Prevent the container from starting if any of the queues defined in the context have
* mismatched arguments (TTL etc). Default false.
* mismatched arguments (TTL etc.). Default false.
* @param mismatchedQueuesFatal true to fail initialization when this condition occurs.
* @return the spec.
* @see AbstractMessageListenerContainer#setMismatchedQueuesFatal(boolean)
@@ -329,7 +329,7 @@ public abstract class AbstractMessageListenerContainerSpec<S extends AbstractMes
/**
* Set to true to automatically declare elements (queues, exchanges, bindings)
* in the application context during container start().
* @param autoDeclare the boolean flag to indicate an declaration operation.
* @param autoDeclare the boolean flag to indicate a declaration operation.
* @return the spec.
* @see AbstractMessageListenerContainer#setAutoDeclare(boolean)
*/

7
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java
View File

@@ -271,7 +271,9 @@ public final class Amqp {
* @param connectionFactory the connectionFactory.
* @return the AmqpPollableMessageChannelSpec.
*/
public static AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> pollableChannel(ConnectionFactory connectionFactory) {
public static AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> pollableChannel(
ConnectionFactory connectionFactory) {
return pollableChannel(null, connectionFactory);
}
@@ -284,7 +286,8 @@ public final class Amqp {
public static AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> pollableChannel(@Nullable String id,
ConnectionFactory connectionFactory) {
AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> spec = new AmqpPollableMessageChannelSpec<>(connectionFactory);
AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> spec =
new AmqpPollableMessageChannelSpec<>(connectionFactory);
return spec.id(id);
}

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
View File

@@ -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
*

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpMessageHeaderErrorMessageStrategy.java
View File

@@ -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<String, Object> headers = new HashMap<String, Object>();
Map<String, Object> 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));

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/BoundRabbitChannelAdvice.java
View File

@@ -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.
*

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

@@ -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 <K> the Kafka message key type.
* @param <V> the Kafka message value type.
* @return the spec.
* @since 3.2
*/
public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
ConsumerFactory<K, V> consumerFactory, ConsumerProperties consumerProperties) {
@@ -96,7 +95,6 @@ public final class Kafka {
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type.
* @return the spec.
* @since 3.2
*/
public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
@@ -115,7 +113,6 @@ public final class Kafka {
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type.
* @return the spec.
* @since 3.2
*/
public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
@@ -135,7 +132,6 @@ public final class Kafka {
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type.
* @return the spec.
* @since 3.2
*/
public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
@@ -341,7 +337,6 @@ public final class Kafka {
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the KafkaGatewayMessageHandlerSpec.
* @since 3.0.2
*/
public static <K, V, R> KafkaOutboundGatewaySpec<K, V, R, ?> outboundGateway(
ReplyingKafkaTemplate<K, V, R> kafkaTemplate) {
@@ -357,7 +352,6 @@ public final class Kafka {
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the KafkaGatewayMessageHandlerSpec.
* @since 3.0.2
*/
public static <K, V, R> KafkaOutboundGatewaySpec.KafkaGatewayMessageHandlerTemplateSpec<K, V, R> outboundGateway(
ProducerFactory<K, V> producerFactory, GenericMessageListenerContainer<K, R> replyContainer) {
@@ -378,7 +372,6 @@ public final class Kafka {
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the spec.
* @since 3.0.2
*/
public static <K, V, R> KafkaInboundGatewaySpec<K, V, R, ?> inboundGateway(
AbstractMessageListenerContainer<K, V> container, KafkaTemplate<K, R> template) {
@@ -396,7 +389,6 @@ public final class Kafka {
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the spec.
* @since 3.0.2
*/
public static <K, V, R> KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<K, V, R> inboundGateway(
ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
@@ -416,7 +408,6 @@ public final class Kafka {
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the spec.
* @since 3.0.2
*/
public static <K, V, R> KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<K, V, R> inboundGateway(
KafkaMessageListenerContainerSpec<K, V> containerSpec, KafkaTemplateSpec<K, R> 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) {

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

@@ -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<K, V, R, S extends KafkaInboundGatewaySpec<
* @param recoveryCallback the recovery callback.
* @return the spec
*/
public S recoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
public S recoveryCallback(RecoveryCallback<?> recoveryCallback) {
this.target.setRecoveryCallback(recoveryCallback);
return _this();
}
@@ -101,7 +101,6 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
* call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}.
* @param onPartitionsAssignedCallback the {@link BiConsumer} to use
* @return the spec
* @since 3.0.4
*/
public S onPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {

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

@@ -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<K, V, S extends KafkaMessageDr
* @param recoveryCallback the recovery callback.
* @return the spec
*/
public S recoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
public S recoveryCallback(RecoveryCallback<?> recoveryCallback) {
this.target.setRecoveryCallback(recoveryCallback);
return _this();
}
@@ -146,7 +146,6 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
* set the payload type the converter should create. Defaults to {@link Object}.
* @param payloadType the type.
* @return the spec
* @since 3.2.0
*/
public S payloadType(Class<?> payloadType) {
this.target.setPayloadType(payloadType);
@@ -178,7 +177,6 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
* call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}.
* @param onPartitionsAssignedCallback the {@link BiConsumer} to use
* @return the spec
* @since 3.0.4
*/
public S onPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {

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

@@ -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<K, V>
}
/**
* 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.

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

@@ -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.

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java
View File

@@ -251,7 +251,7 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Configure a {@link Function} that will be invoked at runtime to determine whether
* to flush the producer after send. By default, the producer is flushed if a
* header {@code kafka_flush} has a value {@link Boolean#TRUE}. Typically, used with a
* Java 8 Lambda expression:
* Java Lambda expression:
* <pre class="code">
* {@code
* .flush(m -> m.getPayload().shouldFlush())
@@ -314,7 +314,6 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Set the channel to which successful send results are sent.
* @param sendSuccessChannel the channel.
* @return the spec.
* @since 3.0.2
*/
public S sendSuccessChannel(MessageChannel sendSuccessChannel) {
this.target.setSendSuccessChannel(sendSuccessChannel);
@@ -325,7 +324,6 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Set the channel to which successful send results are sent.
* @param sendSuccessChannel the channel name.
* @return the spec.
* @since 3.0.2
*/
public S sendSuccessChannel(String sendSuccessChannel) {
this.target.setSendSuccessChannelName(sendSuccessChannel);
@@ -336,7 +334,6 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Set the channel to which failed send results are sent.
* @param sendFailureChannel the channel.
* @return the spec.
* @since 3.0.2
*/
public S sendFailureChannel(MessageChannel sendFailureChannel) {
this.target.setSendFailureChannel(sendFailureChannel);
@@ -347,7 +344,6 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Set the channel to which failed send results are sent.
* @param sendFailureChannel the channel name.
* @return the spec.
* @since 3.0.2
*/
public S sendFailureChannel(String sendFailureChannel) {
this.target.setSendFailureChannelName(sendFailureChannel);
@@ -358,7 +354,6 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Set the channel to which send futures are sent.
* @param futuresChannel the channel.
* @return the spec.
* @since 5.4
*/
public S futuresChannel(MessageChannel futuresChannel) {
this.target.setFuturesChannel(futuresChannel);
@@ -369,7 +364,6 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
* Set the channel to which send futures are sent.
* @param futuresChannel the channel name.
* @return the spec.
* @since 5.4
*/
public S futuresChannel(String futuresChannel) {
this.target.setFuturesChannelName(futuresChannel);

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

@@ -118,7 +118,7 @@ public class KafkaInboundGateway<K, V, R> 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<K, V, R> 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<K, V, R> 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<K, V> kafkaListener = this.listener;

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

@@ -117,7 +117,6 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
* Construct an instance with the provided mode.
* @param messageListenerContainer the container.
* @param mode the mode.
* @since 1.2
*/
public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer<K, V> messageListenerContainer,
ListenerMode mode) {
@@ -153,7 +152,6 @@ public class KafkaMessageDrivenChannelAdapter<K, V> 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<K, V> 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<K, V> extends MessageProducerSuppo
* {@link KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener} into
* {@link FilteringMessageListenerAdapter}.
* @param recordFilterStrategy the {@link RecordFilterStrategy} to use.
* @since 2.0.1
*/
public void setRecordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrategy) {
this.recordFilterStrategy = recordFilterStrategy;
@@ -184,7 +180,6 @@ public class KafkaMessageDrivenChannelAdapter<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> extends MessageProducerSuppo
}
@Override
@SuppressWarnings("deprecation")
protected void onInit() {
super.onInit();
@@ -364,7 +352,6 @@ public class KafkaMessageDrivenChannelAdapter<K, V> 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<K, V> extends MessageProducerSuppo
/**
* The listener mode for the container, record or batch.
* @since 1.2
*
*/
public enum ListenerMode {

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

@@ -155,7 +155,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> consumerFactory, ConsumerProperties consumerProperties) {
@@ -175,7 +174,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> consumerFactory,
ConsumerProperties consumerProperties,
@@ -190,7 +188,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> consumerFactory,
@@ -210,12 +207,10 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> consumerFactory,
ConsumerProperties consumerProperties,
@@ -243,7 +238,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
/**
* Return the currently assigned partitions.
* @return the partitions.
* @since 3.2.2
*/
public Collection<TopicPartition> getAssignedPartitions() {
return Collections.unmodifiableCollection(this.assignedPartitions);
@@ -260,7 +254,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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;

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

@@ -273,7 +273,7 @@ public class KafkaProducerMessageHandler<K, V> 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<K, V> 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<K, V> 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());

20
src/reference/asciidoc/kafka.adoc
View File

@@ -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
`<int-kafka:outbound-channel-adapter>`.
Alternatively, you can change the headers upstream to
the new headers from `KafkaHeaders` by using a `<header-enricher>` 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 `<int-kafka:outbound-channel-adapter>`.
Alternatively, you can change the headers upstream to the new headers from `KafkaHeaders` by using a `<header-enricher>` 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 <<kafka-inbound>> 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: