From c57634f085bc0252232b04445133a0b70e4a62a9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 7 Apr 2021 16:59:57 -0400 Subject: [PATCH] Properly mention Apache Kafka trademark in docs (#3540) --- .../kafka/channel/AbstractKafkaChannel.java | 5 ++- .../kafka/channel/PollableKafkaChannel.java | 21 +++++----- .../channel/PublishSubscribeKafkaChannel.java | 2 +- .../channel/SubscribableKafkaChannel.java | 2 +- .../kafka/channel/package-info.java | 2 +- .../kafka/config/xml/KafkaChannelParser.java | 2 +- .../config/xml/KafkaNamespaceHandler.java | 2 +- .../kafka/config/xml/package-info.java | 2 +- .../kafka/dsl/AbstractKafkaChannelSpec.java | 2 +- .../integration/kafka/dsl/Kafka.java | 3 +- .../dsl/KafkaInboundChannelAdapterSpec.java | 2 +- .../dsl/KafkaPointToPointChannelSpec.java | 2 +- .../dsl/KafkaPublishSubscribeChannelSpec.java | 2 +- .../integration/kafka/dsl/package-info.java | 2 +- .../kafka/inbound/KafkaMessageSource.java | 7 ++-- .../kafka/inbound/package-info.java | 2 +- .../outbound/KafkaProducerMessageHandler.java | 25 ++--------- .../kafka/outbound/package-info.java | 2 +- .../integration/kafka/package-info.java | 2 +- .../kafka/config/spring-integration-kafka.xsd | 41 +++++++++---------- src/reference/asciidoc/changes-5.3-5.4.adoc | 4 +- src/reference/asciidoc/kafka.adoc | 39 +++++++++--------- src/reference/asciidoc/polling-consumer.adoc | 2 +- 23 files changed, 78 insertions(+), 97 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java index 3503b5a552..33a00ff4c7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java @@ -28,9 +28,10 @@ import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; /** - * Abstract MessageChannel backed by a Kafka topic. + * Abstract MessageChannel backed by an Apache Kafka topic. * * @author Gary Russell + * @author Artem Bilan * * @since 5.4 * @@ -44,7 +45,7 @@ public abstract class AbstractKafkaChannel extends AbstractMessageChannel { private String groupId; /** - * Construct an instance with the provided paramters. + * Construct an instance with the provided {@link KafkaOperations} and topic. * @param template the template. * @param topic the topic. */ 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 2eedecf966..47b7ba9a6b 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 @@ -33,7 +33,7 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.util.Assert; /** - * Pollable channel backed by a Kafka topic. + * Pollable channel backed by an Apache Kafka topic. * * @author Gary Russell * @author Artem Bilan @@ -64,13 +64,6 @@ public class PollableKafkaChannel extends AbstractKafkaChannel } } - private static String topic(KafkaMessageSource source) { - Assert.notNull(source, "'source' cannot be null"); - String[] topics = source.getConsumerProperties().getTopics(); - Assert.isTrue(topics != null && topics.length == 1, "Only one topic is allowed"); - return topics[0]; - } - @Override @Nullable public Message receive() { @@ -88,10 +81,9 @@ public class PollableKafkaChannel extends AbstractKafkaChannel ChannelInterceptorList interceptorList = getIChannelInterceptorList(); Deque interceptorStack = null; boolean counted = false; - boolean traceEnabled = isLoggingEnabled() && logger.isTraceEnabled(); try { - if (traceEnabled) { - logger.trace("preReceive on channel '" + this + "'"); + if (isLoggingEnabled()) { + logger.trace(() -> "preReceive on channel '" + this + "'"); } if (interceptorList.getInterceptors().size() > 0) { interceptorStack = new ArrayDeque<>(); @@ -197,4 +189,11 @@ public class PollableKafkaChannel extends AbstractKafkaChannel return this.executorInterceptorsSize > 0; } + private static String topic(KafkaMessageSource source) { + Assert.notNull(source, "'source' cannot be null"); + String[] topics = source.getConsumerProperties().getTopics(); + Assert.isTrue(topics != null && topics.length == 1, "Only one topic is allowed"); + return topics[0]; + } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java index 2450bf4b33..8ca744760e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java @@ -23,7 +23,7 @@ import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.KafkaOperations; /** - * Publish/subscribe channel backed by a Kafka topic. + * Publish/subscribe channel backed by an Apache Kafka topic. * * @author Gary Russell * 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 a0d1e28543..7fd35db6b0 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 @@ -34,7 +34,7 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; /** - * Subscribable channel backed by a Kafka topic. + * Subscribable channel backed by an Apache Kafka topic. * * @author Gary Russell * @author Artem Bilan diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java index b42feb5e6e..3816bbfa8f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java @@ -1,4 +1,4 @@ /** - * Provides classes related to message channels. + * Provides classes related to message channel implementations for Apache Kafka. */ package org.springframework.integration.kafka.channel; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java index 7fce490fc0..ba24122338 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java @@ -28,7 +28,7 @@ import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; import org.springframework.util.StringUtils; /** - * Parser for a channel backed by a Kafka topic. + * Parser for a channel backed by an Apache Kafka topic. * * @author Gary Russell * diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java index d17d047ad5..8f3e72b434 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java @@ -19,7 +19,7 @@ package org.springframework.integration.kafka.config.xml; import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; /** - * The namespace handler for the Kafka namespace. + * The namespace handler for the Apache Kafka namespace. * * @author Soby Chacko * @author Gary Russell diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/package-info.java index 3289f40710..d102e5e5ab 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/package-info.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/package-info.java @@ -1,4 +1,4 @@ /** - * Provides parser classes to provide Xml namespace support for the Kafka components. + * Provides parser classes to provide Xml namespace support for the Apache Kafka components. */ package org.springframework.integration.kafka.config.xml; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java index 0ea8ccc865..328aa71c79 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java @@ -21,7 +21,7 @@ import org.springframework.integration.kafka.channel.AbstractKafkaChannel; /** * - * Spec for a message channel backed by a Kafka topic. + * Spec for a message channel backed by an Apache Kafka topic. * * @param the spec type. * @param the channel type. 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 6b432c2ac0..91a3a978e1 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 @@ -143,7 +143,8 @@ public final class Kafka { KafkaAckCallbackFactory ackCallbackFactory, boolean allowMultiFetch) { - return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory, allowMultiFetch); + return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory, + allowMultiFetch); } /** 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 798d0ade52..9d5dec04fd 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 @@ -24,7 +24,7 @@ import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.support.converter.RecordMessageConverter; /** - * Spec for a polled Kafka inbound channel adapter. + * Spec for a polled Apache Kafka inbound channel adapter. * * @param the key type. * @param the value type. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java index 43e4d4ca9f..c3917469a6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java @@ -21,7 +21,7 @@ import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.KafkaTemplate; /** - * Spec for a point to point channel backed by a Kafka topic. + * Spec for a point to point channel backed by an Apache Kafka topic. * * @author Gary Russell * diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java index 82c925c773..1fb085da92 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java @@ -21,7 +21,7 @@ import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.KafkaTemplate; /** - * Spec for a publish/subscribe channel backed by a Kafka topic. + * Spec for a publish/subscribe channel backed by an Apache Kafka topic. * * @author Gary Russell * diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java index adf6553c27..b70ca47e14 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/package-info.java @@ -1,4 +1,4 @@ /** - * Provides Kafka Components support for Spring Integration Java DSL. + * Provides Spring Integration Java DSL Components support for Apache Kafka. */ package org.springframework.integration.kafka.dsl; 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 3a1db60e08..03e9aa57a6 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 @@ -71,15 +71,14 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** - * Polled message source for kafka. Only one thread can poll for data (or - * acknowledge a message) at a time. + * Polled message source for Apache Kafka. + * Only one thread can poll for data (or acknowledge a message) at a time. *

* NOTE: If the application acknowledges messages out of order, the acks * will be deferred until all messages prior to the offset are ack'd. * If multiple records are retrieved and an earlier offset is requeued, records * from the subsequent offsets will be redelivered - even if they were - * processed successfully. Applications should therefore implement - * idempotency. + * processed successfully. Applications should therefore implement idempotency. *

* Starting with version 3.1.2, this source implements {@link Pausable} which * allows you to pause and resume the {@link Consumer}. While the consumer is diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/package-info.java index 2688d7b67d..e75fc2ac77 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/package-info.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/package-info.java @@ -1,4 +1,4 @@ /** - * Provides inbound Spring Integration Kafka components. + * Provides Spring Integration inbound components for Apache Kafka. */ package org.springframework.integration.kafka.inbound; 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 cb9aef46f1..d41d748af0 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 @@ -75,11 +75,11 @@ import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.util.concurrent.SettableListenableFuture; /** - * Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate} it is used as + * A Message Handler for Apache Kafka; when supplied with a {@link ReplyingKafkaTemplate} it is used as * the handler in an outbound gateway. When supplied with a simple {@link KafkaTemplate} * it used as the handler in an outbound channel adapter. *

- * Starting with version 3.2.1 the handler supports receiving a pre-built + * The handler also supports receiving a pre-built * {@link ProducerRecord} payload. In that case, most configuration properties * ({@link #setTopicExpression(Expression)} etc.) are ignored. If the handler is used as * gateway, the {@link ProducerRecord} will have its headers enhanced to add the @@ -103,7 +103,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes implements ManageableLifecycle { /** - * Buffer added to ensure our timeout is longer than Kafka's. + * Buffer added to ensure our timeout is longer than Apache Kafka timeout. */ private static final int DEFAULT_TIMEOUT_BUFFER = 5000; @@ -227,7 +227,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * The resulting value should be a {@link Long} type representing epoch time in milliseconds. * @param timestampExpression the {@link Expression} for timestamp to wait for result * fo send operation. - * @since 2.3 */ public void setTimestampExpression(Expression timestampExpression) { this.timestampExpression = timestampExpression; @@ -239,7 +238,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * {@link Boolean} value in a {@link KafkaIntegrationHeaders#FLUSH} header; false if * absent. * @param flushExpression the {@link Expression}. - * @since 3.3 */ public void setFlushExpression(Expression flushExpression) { Assert.notNull(flushExpression, "'flushExpression' cannot be null"); @@ -249,7 +247,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set the header mapper to use. * @param headerMapper the mapper; can be null to disable header mapping. - * @since 2.3 */ public void setHeaderMapper(KafkaHeaderMapper headerMapper) { this.headerMapper = headerMapper; @@ -268,7 +265,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * should wait for the send operation results or not. Defaults to {@code false}. * In {@code sync} mode a downstream send operation exception will be re-thrown. * @param sync the send mode; async by default. - * @since 2.0.1 */ public void setSync(boolean sync) { this.sync = sync; @@ -280,7 +276,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * 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. - * @since 2.0.1 */ @Override public final void setSendTimeout(long sendTimeout) { @@ -296,7 +291,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * less than that value, the higher value is used. * @param sendTimeoutExpression the {@link Expression} for timeout to wait for result * for a send operation. - * @since 2.1.1 * @see #setTimeoutBuffer(int) */ public void setSendTimeoutExpression(Expression sendTimeoutExpression) { @@ -310,7 +304,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * to this channel with a payload of a {@link KafkaSendFailureException} with the * failed message and cause. * @param sendFailureChannel the failure channel. - * @since 2.1.2 */ public void setSendFailureChannel(MessageChannel sendFailureChannel) { this.sendFailureChannel = sendFailureChannel; @@ -322,7 +315,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * sent to this channel name with a payload of a {@link KafkaSendFailureException} * with the failed message and cause. * @param sendFailureChannelName the failure channel name. - * @since 2.1.2 */ public void setSendFailureChannelName(String sendFailureChannelName) { this.sendFailureChannelName = sendFailureChannelName; @@ -331,7 +323,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set the success channel. * @param sendSuccessChannel the Success channel. - * @since 3.0.2 */ public void setSendSuccessChannel(MessageChannel sendSuccessChannel) { this.sendSuccessChannel = sendSuccessChannel; @@ -340,7 +331,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set the Success channel name. * @param sendSuccessChannelName the Success channel name. - * @since 3.0.2 */ public void setSendSuccessChannelName(String sendSuccessChannelName) { this.sendSuccessChannelName = sendSuccessChannelName; @@ -349,7 +339,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set the futures channel. * @param futuresChannel the futures channel. - * @since 5.4 */ public void setFuturesChannel(MessageChannel futuresChannel) { this.futuresChannel = futuresChannel; @@ -358,7 +347,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set the futures channel name. * @param futuresChannelName the futures channel name. - * @since 5.4 */ public void setFuturesChannelName(String futuresChannelName) { this.futuresChannelName = futuresChannelName; @@ -368,7 +356,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * Set the error message strategy implementation to use when sending error messages after * send failures. Cannot be null. * @param errorMessageStrategy the implementation. - * @since 2.1.2 */ public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) { Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null"); @@ -378,7 +365,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set a message converter for gateway replies. * @param messageConverter the converter. - * @since 3.0.2 * @see #setReplyPayloadType(Type) */ public void setReplyMessageConverter(RecordMessageConverter messageConverter) { @@ -390,7 +376,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * 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 3.0.2 * @see #setReplyMessageConverter(RecordMessageConverter) */ public void setReplyPayloadType(Type payloadType) { @@ -401,7 +386,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Set a {@link ProducerRecordCreator} to create the {@link ProducerRecord}. * @param producerRecordCreator the creator. - * @since 3.2.1 */ public void setProducerRecordCreator(ProducerRecordCreator producerRecordCreator) { Assert.notNull(producerRecordCreator, "'producerRecordCreator' cannot be null"); @@ -413,7 +397,6 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes * to determine the minimum time to wait for the send future completion when * {@link #setSync(boolean) sync} is true. * @param timeoutBuffer the buffer. - * @since 5.4 * @see #setSendTimeoutExpression(Expression) */ public void setTimeoutBuffer(int timeoutBuffer) { @@ -694,7 +677,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes if (this.sync || this.isGateway) { Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class); if (sendTimeout != null && sendTimeout <= this.deliveryTimeoutMsProperty) { - this.logger.debug("'sendTimeout' increased to " + this.logger.debug(() -> "'sendTimeout' increased to " + (this.deliveryTimeoutMsProperty + this.timeoutBuffer) + "ms; it must be greater than the 'delivery.timeout.ms' Kafka producer " + "property to avoid false failures"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java index d5f6b0cd54..8f27e06393 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java @@ -1,4 +1,4 @@ /** - * Provides outbound Spring Integration Kafka components. + * Provides Spring Integration outbound components for Apache Kafka. */ package org.springframework.integration.kafka.outbound; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/package-info.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/package-info.java index 484759047a..d77f8072e7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/package-info.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/package-info.java @@ -1,4 +1,4 @@ /** - * Root package of the Kafka Module. + * Root package of the Apache Kafka Module. */ package org.springframework.integration.kafka; diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd index 699c4003f2..a45de95f31 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd @@ -12,16 +12,15 @@ schemaLocation="https://www.springframework.org/schema/integration/spring-integration.xsd"/> - + + Defines the configuration elements for the Spring Integration Adapters for Apache Kafka. + - Creates a subscribable channel that is backed by a Kafka topic. + Creates a subscribable channel that is backed by an Apache Kafka topic. @@ -35,7 +34,7 @@ - Creates a pollable channel that is backed by a Kafka topic. + Creates a pollable channel that is backed by an Apache Kafka topic. @@ -64,7 +63,7 @@ - Creates a subscribable pub/sub channel that is backed by a Kafka topic. + Creates a subscribable pub/sub channel that is backed by an Apache Kafka topic. @@ -98,7 +97,7 @@ Defines the Consumer Endpoint for the KafkaProducerMessageHandler - that writes the contents of the Message to kafka broker and receives + that writes the contents of the Message to an Apache Kafka broker and receives a reply. @@ -129,7 +128,7 @@ - Message Channel to which Messages should be sent in order to have them sent to Kafka. + Message Channel to which Messages should be sent in order to have them sent to the Apache Kafka. @@ -151,7 +150,7 @@ - Message Channel to which replies should be sent after being received from Kafka. + Message Channel to which replies should be sent after being received from the Apache Kafka. @@ -195,7 +194,7 @@ Defines a Polling Channel Adapter for the 'org.springframework.integration.kafka.inbound.KafkaMessageSource' - for polling a Kafka topic. + for polling an Apache Kafka topic. @@ -253,7 +252,7 @@ [DEPRECATED] - The 'group.id' Kafka Consumer property; overrides the value in the consumer factory. + The 'group.id' KafkaConsumer property; overrides the value in the consumer factory. Deprecated in favor of 'consumer-properties'. @@ -262,7 +261,7 @@ [DEPRECATED] - The 'client.id' Kafka Consumer property; overrides the value in the consumer factory. + The 'client.id' KafkaConsumer property; overrides the value in the consumer factory. Deprecated in favor of 'consumer-properties'. @@ -512,7 +511,7 @@ @@ -520,7 +519,7 @@ @@ -529,7 +528,7 @@ @@ -537,7 +536,7 @@ @@ -546,7 +545,7 @@ @@ -554,7 +553,7 @@ @@ -760,7 +759,7 @@ - Set the 'group.id' Kafka consumer property. + Set the 'group.id' KafkaConsumer property. @@ -770,7 +769,7 @@ diff --git a/src/reference/asciidoc/changes-5.3-5.4.adoc b/src/reference/asciidoc/changes-5.3-5.4.adoc index c1af14ca38..44c7759c90 100644 --- a/src/reference/asciidoc/changes-5.3-5.4.adoc +++ b/src/reference/asciidoc/changes-5.3-5.4.adoc @@ -5,8 +5,8 @@ === New Components [[x5.4-sik]] -==== Apache Kafka Channel Adapters -The standalone https://projects.spring.io/spring-integration-kafka/[Spring Integration Kafka] project has been merged as a `spring-integration-kafka` module to this project. +==== Channel Adapters for Apache Kafka +The standalone https://projects.spring.io/spring-integration-kafka/[Spring Integration for Apache Kafka] project has been merged as a `spring-integration-kafka` module to this project. The `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed. diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 7b3de1cb44..b280eec022 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1,5 +1,5 @@ [[kafka]] -== Spring for Apache Kafka Support +== Apache Kafka Support === Overview @@ -16,15 +16,14 @@ It provides the following components: [[kafka-outbound]] === Outbound Channel Adapter -The Outbound channel adapter is used to publish messages from a Spring Integration channel to Kafka topics. -The channel is defined in the application context and then wired into the application that sends messages to Kafka. -Sender applications can publish to Kafka by using Spring Integration messages, which are internally converted to Kafka messages by the outbound channel adapter, as follows: +The Outbound channel adapter is used to publish messages from a Spring Integration channel to Apache Kafka topics. +The channel is defined in the application context and then wired into the application that sends messages to Apache Kafka. +Sender applications can publish to Apache Kafka by using Spring Integration messages, which are internally converted to Kafka records by the outbound channel adapter, as follows: -* The payload of the Spring Integration message is used to populate the payload of the Kafka message. -* By default, the `kafka_messageKey` header of the Spring Integration message is used to populate the key of the Kafka message. +* The payload of the Spring Integration message is used to populate the payload of the Kafka record. +* By default, the `kafka_messageKey` header of the Spring Integration message is used to populate the key of the Kafka record. -You can customize the target topic and partition for publishing the message through the `kafka_topic` -and `kafka_partitionId` headers, respectively. +You can customize the target topic and partition for publishing the message through the `kafka_topic` and `kafka_partitionId` headers, respectively. In addition, the `` provides the ability to extract the key, target topic, and target partition by applying SpEL expressions on the outbound message. To that end, it supports three mutually exclusive pairs of attributes: @@ -79,7 +78,7 @@ IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it t ==== Java Configuration -The following example shows how to configure the Kafka outbound channel adapter with Java: +The following example shows how to configure the outbound channel adapter for Apache Kafka with Java: ==== [source, java] @@ -113,7 +112,7 @@ public ProducerFactory producerFactory() { ==== Java DSL Configuration -The following example shows how to configure the Kafka outbound channel adapter Spring Integration Java DSL: +The following example shows how to configure the outbound channel adapter for Apache Kafka with Spring Integration Java DSL: ==== [source, java] @@ -199,7 +198,7 @@ The following example shows how to configure the Kafka outbound channel adapter The `KafkaMessageDrivenChannelAdapter` (``) uses a `spring-kafka` `KafkaMessageListenerContainer` or `ConcurrentListenerContainer`. -Starting with spring-integration-kafka version 2.1, the `mode` attribute is available. +Also the `mode` attribute is available. It can accept values of `record` or `batch` (default: `record`). For `record` mode, each message payload is converted from a single `ConsumerRecord`. For `batch` mode, the payload is a list of objects that are converted from all the `ConsumerRecord` instances returned by the consumer poll. @@ -282,7 +281,7 @@ public IntegrationFlow topic1ListenerFromKafkaFlow() { ---- ==== -Starting with Spring for Apache Kafka version 2.2 (Spring Integration Kafka 3.1), you can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes. +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. 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. @@ -352,7 +351,7 @@ The following example shows how to configure a message-driven channel adapter wi [[kafka-inbound-pollable]] === Inbound Channel Adapter -Introduced in version 3.0.1, the `KafkaMessageSource` provides a pollable channel adapter implementation. +The `KafkaMessageSource` provides a pollable channel adapter implementation. ==== Java Configuration @@ -429,7 +428,7 @@ If your code invokes the gateway behind a synchronous https://docs.spring.io/spr IMPORTANT: The gateway does not accept requests until the reply container has been assigned its topics and partitions. It is suggested that you add a `ConsumerRebalanceListener` to the template's reply container properties and wait for the `onPartitionsAssigned` call before sending messages to the gateway. -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 is `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. @@ -451,7 +450,7 @@ public KafkaProducerMessageHandler outGateway( Refer to the javadocs for available properties. -Notice that the same class as the <> is used, the only difference being that the Kafka template passed into the constructor is a `ReplyingKafkaTemplate`. +Notice that the same class as the <> is used, the only difference being that the `KafkaTemplate` passed into the constructor is a `ReplyingKafkaTemplate`. See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] for more information. The outbound topic, partition, key, and so on are determined in the same way as the outbound adapter. @@ -585,7 +584,7 @@ public IntegrationFlow serverGateway() { ---- ==== -Starting with Spring for Apache Kafka version 2.2 (Spring Integration Kafka 3.1), you can also use the container factory that is used for `@KafkaListener` annotations to create `ConcurrentMessageListenerContainer` instances for other purposes. +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. See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] and <> for examples. ==== XML Configuration @@ -614,9 +613,9 @@ See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring See the XML schema for a description of each property. [[kafka-channels]] -=== Channels Backed by Kafka Topics +=== Channels Backed by Apache Kafka Topics -Spring Integration for Apache Kafka version 3.3 (still under development) introduces channels backed by a Kafka topic for persistence. +Spring Integration has `MessageChannel` implementations backed by an Apache Kafka topic for persistence. Each channel requires a `KafkaTemplate` for the sending side and either a listener container factory (for subscribable channels) or a `KafkaMessageSource` for a pollable channel. @@ -786,10 +785,10 @@ public KafkaMessageDrivenChannelAdapter === Null Payloads and Log Compaction 'Tombstone' Records Spring Messaging `Message` objects cannot have `null` payloads. -When you use the Kafka endpoints, `null` payloads (also known as tombstone records) are represented by a payload of type `KafkaNull`. +When you use the endpoints for Apache Kafka, `null` payloads (also known as tombstone records) are represented by a payload of type `KafkaNull`. See See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] for more information. -Starting with version 3.1 of Spring Integration Kafka, such records can now be received by Spring Integration POJO methods with a true `null` value instead. +The POJO methods for Spring Integration endpoints can use a true `null` value instead instead of `KafkaNull`. To do so, mark the parameter with `@Payload(required = false)`. The following example shows how to do so: diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index 303f5b7b64..136b53e6d1 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -97,7 +97,7 @@ public interface AcknowledgmentCallback { } ---- -Not all message sources (for example, Kafka) support the `REJECT` status. +Not all message sources (for example, a `KafkaMessageSource`) support the `REJECT` status. It is treated the same as `ACCEPT`. Applications can acknowledge a message at any time, as the following example shows: