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

Properly mention Apache Kafka trademark in docs (#3540)

Browse Source
This commit is contained in:
Artem Bilan
2021-04-07 16:59:57 -04:00
committed by GitHub
parent e7d99e994d
commit c57634f085
23 changed files with 78 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java
View File

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

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

@@ -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<ChannelInterceptor> 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];
}
}

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java
View File

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

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

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/package-info.java
View File

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java
View File

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java
View File

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/package-info.java
View File

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

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

@@ -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 <S> the spec type.
* @param <C> the channel type.

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

@@ -143,7 +143,8 @@ public final class Kafka {
KafkaAckCallbackFactory<K, V> ackCallbackFactory,
boolean allowMultiFetch) {
return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory, allowMultiFetch);
return new KafkaInboundChannelAdapterSpec<>(consumerFactory, consumerProperties, ackCallbackFactory,
allowMultiFetch);
}
/**

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

@@ -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 <K> the key type.
* @param <V> the value type.

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

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

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

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

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

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

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

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

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/package-info.java
View File

@@ -1,4 +1,4 @@
/**
* Provides inbound Spring Integration Kafka components.
* Provides Spring Integration inbound components for Apache Kafka.
*/
package org.springframework.integration.kafka.inbound;

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

@@ -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.
* <p>
* 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> extends AbstractReplyProducingMes
/**
* Set a {@link ProducerRecordCreator} to create the {@link ProducerRecord}.
* @param producerRecordCreator the creator.
* @since 3.2.1
*/
public void setProducerRecordCreator(ProducerRecordCreator<K, V> producerRecordCreator) {
Assert.notNull(producerRecordCreator, "'producerRecordCreator' cannot be null");
@@ -413,7 +397,6 @@ public class KafkaProducerMessageHandler<K, V> 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<K, V> 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");

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/package-info.java
View File

@@ -1,4 +1,4 @@
/**
* Provides outbound Spring Integration Kafka components.
* Provides Spring Integration outbound components for Apache Kafka.
*/
package org.springframework.integration.kafka.outbound;

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/package-info.java
View File

@@ -1,4 +1,4 @@
/**
* Root package of the Kafka Module.
* Root package of the Apache Kafka Module.
*/
package org.springframework.integration.kafka;

41
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd
View File

@@ -12,16 +12,15 @@
schemaLocation="https://www.springframework.org/schema/integration/spring-integration.xsd"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for the Spring Integration
Kafka Adapters.
]]></xsd:documentation>
<xsd:documentation>
Defines the configuration elements for the Spring Integration Adapters for Apache Kafka.
</xsd:documentation>
</xsd:annotation>
<xsd:element name="channel">
<xsd:annotation>
<xsd:documentation>
Creates a subscribable channel that is backed by a Kafka topic.
Creates a subscribable channel that is backed by an Apache Kafka topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -35,7 +34,7 @@
<xsd:element name="pollable-channel">
<xsd:annotation>
<xsd:documentation>
Creates a pollable channel that is backed by a Kafka topic.
Creates a pollable channel that is backed by an Apache Kafka topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -64,7 +63,7 @@
<xsd:element name="publish-subscribe-channel">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -98,7 +97,7 @@
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
@@ -129,7 +128,7 @@
<xsd:attribute name="request-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -151,7 +150,7 @@
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -195,7 +194,7 @@
<xsd:documentation>
Defines a Polling Channel Adapter for the
'org.springframework.integration.kafka.inbound.KafkaMessageSource'
for polling a Kafka topic.
for polling an Apache Kafka topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -253,7 +252,7 @@
<xsd:annotation>
<xsd:documentation>
[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'.
</xsd:documentation>
</xsd:annotation>
@@ -262,7 +261,7 @@
<xsd:annotation>
<xsd:documentation>
[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'.
</xsd:documentation>
</xsd:annotation>
@@ -512,7 +511,7 @@
<xsd:attribute name="topic" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the name of the Kafka topic.
Specifies the name of an Apache Kafka topic.
This attribute is mutually exclusive with 'topic-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
@@ -520,7 +519,7 @@
<xsd:attribute name="topic-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the name of the Kafka topic
Specifies the expression to determine the name of an Apache Kafka topic
against the Message at runtime.
This attribute is mutually exclusive with 'topic' attribute.
]]></xsd:documentation>
@@ -529,7 +528,7 @@
<xsd:attribute name="message-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the Key for the Kafka message.
Specifies the Key for the Kafka record.
This attribute is mutually exclusive with 'message-key-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
@@ -537,7 +536,7 @@
<xsd:attribute name="message-key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the Key for Kafka message
Specifies the expression to determine the Key for Kafka record
against the Message at runtime.
This attribute is mutually exclusive with 'message-key' attribute.
]]></xsd:documentation>
@@ -546,7 +545,7 @@
<xsd:attribute name="partition-id" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the target partition for the Kafka message.
Specifies the target partition for the Kafka record.
This attribute is mutually exclusive with 'partition-id-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
@@ -554,7 +553,7 @@
<xsd:attribute name="partition-id-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the partition for Kafka message
Specifies the expression to determine the partition for Kafka record
against the Message at runtime.
This attribute is mutually exclusive with 'partition-id' attribute.
]]></xsd:documentation>
@@ -760,7 +759,7 @@
<xsd:attribute name="group-id">
<xsd:annotation>
<xsd:documentation>
Set the 'group.id' Kafka consumer property.
Set the 'group.id' KafkaConsumer property.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -770,7 +769,7 @@
<xsd:attribute name="header-mapper" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the HeaderMapper used map Spring message headers to or from Kafka message headers.
Specifies the HeaderMapper used map Spring message headers to or from Kafka record headers.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">

4
src/reference/asciidoc/changes-5.3-5.4.adoc
View File

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

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

@@ -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 `<int-kafka:outbound-channel-adapter>` 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<String, String> 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` (`<int-kafka:message-driven-channel-adapter>`) 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<String, String> outGateway(
Refer to the javadocs for available properties.
Notice that the same class as the <<kafka-outbound,outbound channel adapter>> is used, the only difference being that the Kafka template passed into the constructor is a `ReplyingKafkaTemplate`.
Notice that the same class as the <<kafka-outbound,outbound channel adapter>> 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 <<kafka-inbound>> 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<String, String>
=== 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:

2
src/reference/asciidoc/polling-consumer.adoc
View File

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