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

Merge Spring Integration Kafka into Main Repo

Browse Source 
- merge master branch from old repo (`git-filter-repo`) for src branch
- update build.gradle
- copy docs from Spring for Apache Kafka Chapter

Doc polishing.

Fix http: references in tests.

* Add `What's New` note and fix `polling-consumer.adoc`
not linking to external project any more
This commit is contained in:
Gary Russell
2020-06-24 16:03:13 -04:00
committed by Artem Bilan
parent 3ec74f380f
commit b860a2e818
66 changed files with 961 additions and 508 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
build.gradle
View File

@@ -98,6 +98,7 @@ ext {
soapVersion = '1.4.0'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.0-SNAPSHOT'
springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2020.0.0-SNAPSHOT'
springKafkaVersion = '2.5.3.BUILD-SNAPSHOT'
springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.4.0-SNAPSHOT'
springRetryVersion = '1.3.0'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.0-SNAPSHOT'
@@ -592,6 +593,24 @@ project('spring-integration-jpa') {
}
}
project('spring-integration-kafka') {
description = 'Spring Integration for Apache Kafka'
dependencies {
api project(':spring-integration-core')
api ("org.springframework.kafka:spring-kafka:$springKafkaVersion") {
exclude group: 'org.springframework'
}
testImplementation "org.springframework.kafka:spring-kafka-test:$springKafkaVersion"
testImplementation "com.willowtreeapps.assertk:assertk-jvm:$assertkVersion"
testImplementation 'org.jetbrains.kotlin:kotlin-reflect'
testImplementation 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core'
testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'
}
}
project('spring-integration-mail') {
description = 'Spring Integration Mail Support'
dependencies {

22
spring-integration-kafka/src/api/overview.html
View File

@@ -1,22 +0,0 @@
<html>
<body>
This document is the API specification for Spring Integration Kafka Extension
<hr/>
<div id="overviewBody">
<p>
For further API reference and developer documentation, see the
<a href="https://docs.spring.io/spring-integration/reference" target="_top">Spring
Integration reference documentation</a>.
That documentation contains more detailed, developer-targeted
descriptions, with conceptual overviews, definitions of terms,
workarounds, and working code examples.
</p>
<p>
If you are interested in commercial training, consultancy, and
support for Spring Integration, please visit <a href="https://www.springsource.com" target="_top">
https://www.springsource.com</a>
</p>
</div>
</body>
</html>

17
spring-integration-kafka/src/checkstyle/checkstyle-header.txt
View File

@@ -1,17 +0,0 @@
^\Q/*\E$
^\Q * Copyright \E20\d\d(\-20\d\d)?\Q the original author or authors.\E$
^\Q *\E$
^\Q * Licensed under the Apache License, Version 2.0 (the "License");\E$
^\Q * you may not use this file except in compliance with the License.\E$
^\Q * You may obtain a copy of the License at\E$
^\Q *\E$
^\Q * https://www.apache.org/licenses/LICENSE-2.0\E$
^\Q *\E$
^\Q * Unless required by applicable law or agreed to in writing, software\E$
^\Q * distributed under the License is distributed on an "AS IS" BASIS,\E$
^\Q * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\E$
^\Q * See the License for the specific language governing permissions and\E$
^\Q * limitations under the License.\E$
^\Q */\E$
^$
^.*$

8
spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml
View File

@@ -1,8 +0,0 @@
<?xml version="1.0"?>
<!DOCTYPE suppressions PUBLIC
"-//Checkstyle//DTD SuppressionFilter Configuration 1.2//EN"
"https://checkstyle.org/dtds/suppressions_1_2.dtd">
<suppressions>
<suppress files="package-info\.java" checks=".*"/>
<suppress files="[\\/]test[\\/]" checks="Javadoc*"/>
</suppressions>

180
spring-integration-kafka/src/checkstyle/checkstyle.xml
View File

@@ -1,180 +0,0 @@
<?xml version="1.0"?>
<!DOCTYPE module PUBLIC
"-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
"https://checkstyle.org/dtds/configuration_1_3.dtd">
<module name="Checker">
<module name="SuppressionFilter">
<property name="file" value="src/checkstyle/checkstyle-suppressions.xml"/>
</module>
<!-- Root Checks -->
<module name="RegexpHeader">
<property name="headerFile" value="src/checkstyle/checkstyle-header.txt"/>
<property name="fileExtensions" value="java"/>
</module>
<module name="NewlineAtEndOfFile"/>
<!-- TreeWalker Checks -->
<module name="TreeWalker">
<!-- Annotations -->
<module name="AnnotationUseStyle">
<property name="elementStyle" value="compact"/>
</module>
<module name="MissingOverride"/>
<module name="PackageAnnotation"/>
<module name="AnnotationLocation">
<property name="allowSamelineSingleParameterlessAnnotation"
value="false"/>
</module>
<!-- Block Checks -->
<module name="EmptyBlock">
<property name="option" value="text"/>
</module>
<module name="LeftCurly"/>
<module name="RightCurly">
<property name="option" value="alone"/>
</module>
<module name="NeedBraces"/>
<module name="AvoidNestedBlocks"/>
<!-- Class Design -->
<module name="FinalClass"/>
<module name="InterfaceIsType"/>
<module name="HideUtilityClassConstructor"/>
<module name="MutableException"/>
<module name="InnerTypeLast"/>
<module name="OneTopLevelClass"/>
<!-- Coding -->
<module name="CovariantEquals"/>
<module name="EmptyStatement"/>
<module name="EqualsHashCode"/>
<module name="InnerAssignment"/>
<module name="SimplifyBooleanExpression"/>
<module name="SimplifyBooleanReturn"/>
<module name="StringLiteralEquality"/>
<module name="NestedForDepth">
<property name="max" value="3"/>
</module>
<module name="NestedIfDepth">
<property name="max" value="4"/>
</module>
<module name="NestedTryDepth">
<property name="max" value="3"/>
</module>
<module name="MultipleVariableDeclarations"/>
<module name="RequireThis">
<property name="validateOnlyOverlapping" value="false"/>
<property name="checkMethods" value="false"/>
</module>
<module name="OneStatementPerLine"/>
<!-- Imports -->
<module name="AvoidStarImport"/>
<module name="AvoidStaticImport">
<property name="excludes"
value="org.assertj.core.api.Assertions.*,
org.junit.Assert.*,
org.junit.Assume.*,
org.junit.internal.matchers.ThrowableMessageMatcher.*,
org.hamcrest.CoreMatchers.*,
org.hamcrest.Matchers.*,
org.mockito.Mockito.*,
org.mockito.BDDMockito.*,
org.mockito.ArgumentMatchers.*,
org.springframework.kafka.test.hamcrest.KafkaMatchers.*,
org.springframework.kafka.test.assertj.KafkaConditions.*"/>
</module>
<module name="IllegalImport"/>
<module name="RedundantImport"/>
<module name="UnusedImports">
<property name="processJavadoc" value="true"/>
</module>
<module name="ImportOrder">
<property name="groups" value="java,/^javax?\./,org,org.springframework,*"/>
<property name="ordered" value="true"/>
<property name="separated" value="true"/>
<property name="option" value="top"/>
<property name="sortStaticImportsAlphabetically" value="true"/>
</module>
<!-- Javadoc Comments -->
<module name="JavadocType">
<property name="scope" value="package"/>
<property name="authorFormat" value=".+\s.+"/>
</module>
<module name="JavadocMethod"/>
<module name="JavadocVariable">
<property name="scope" value="public"/>
</module>
<module name="JavadocStyle">
<property name="checkEmptyJavadoc" value="true"/>
</module>
<module name="NonEmptyAtclauseDescription"/>
<module name="JavadocTagContinuationIndentation">
<property name="offset" value="0"/>
</module>
<module name="AtclauseOrder">
<property name="target" value="CLASS_DEF, INTERFACE_DEF, ENUM_DEF"/>
<property name="tagOrder" value="@param, @author, @since, @see, @version, @serial, @deprecated"/>
</module>
<module name="AtclauseOrder">
<property name="target" value="METHOD_DEF, CTOR_DEF, VARIABLE_DEF"/>
<property name="tagOrder" value="@param, @return, @throws, @since, @deprecated, @see"/>
</module>
<!-- Miscellaneous -->
<module name="CommentsIndentation">
<property name="tokens" value="BLOCK_COMMENT_BEGIN"/>
</module>
<module name="UpperEll"/>
<module name="ArrayTypeStyle"/>
<module name="OuterTypeFilename"/>
<!-- Modifiers -->
<module name="RedundantModifier"/>
<!-- Regexp -->
<module name="RegexpSinglelineJava">
<property name="format" value="^\t* +\t*\S"/>
<property name="message"
value="Line has leading space characters; indentation should be performed with tabs only."/>
<property name="ignoreComments" value="true"/>
</module>
<module name="RegexpSinglelineJava">
<property name="maximum" value="0"/>
<property name="format" value="org\.mockito\..*Mockito\.(when|doThrow|doAnswer)"/>
<property name="message"
value="Please use BDDMockito instead of Mockito.(when|doThrow|doAnswer)."/>
<property name="ignoreComments" value="true"/>
</module>
<module name="RegexpSinglelineJava">
<property name="maximum" value="0"/>
<property name="format" value="org\.junit\.Assert\.assert"/>
<property name="message"
value="Please use AssertJ imports."/>
<property name="ignoreComments" value="true"/>
</module>
<module name="Regexp">
<property name="format" value="[ \t]+$"/>
<property name="illegalPattern" value="true"/>
<property name="message" value="Trailing whitespace"/>
</module>
<!-- Whitespace -->
<module name="GenericWhitespace"/>
<module name="MethodParamPad"/>
<module name="NoWhitespaceAfter">
<property name="tokens" value="BNOT, DEC, DOT, INC, LNOT, UNARY_MINUS, UNARY_PLUS, ARRAY_DECLARATOR"/>
</module>
<module name="NoWhitespaceBefore"/>
<module name="ParenPad"/>
<module name="TypecastParenPad"/>
<module name="WhitespaceAfter"/>
<module name="WhitespaceAround"/>
</module>
</module>

201
spring-integration-kafka/src/dist/license.txt vendored
View File

@@ -1,201 +0,0 @@
Apache License
Version 2.0, January 2004
https://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by testData1) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class testData1 and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [testData1 of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

21
spring-integration-kafka/src/dist/notice.txt vendored
View File

@@ -1,21 +0,0 @@
========================================================================
== NOTICE file corresponding to section 4 d of the Apache License, ==
== Version 2.0, in this case for the Spring Integration distribution. ==
========================================================================
This product includes software developed by
the Apache Software Foundation (https://www.apache.org).
The end-user documentation included with a redistribution, if any,
must include the following acknowledgement:
"This product includes software developed by the Spring Framework
Project (https://www.springframework.org)."
Alternatively, this acknowledgement may appear in the software itself,
if and wherever such third-party acknowledgements normally appear.
The names "Spring", "Spring Framework", and "Spring Integration" must
not be used to endorse or promote products derived from this software
without prior written permission. For written permission, please contact
enquiries@springsource.com.

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

@@ -32,7 +32,8 @@ import org.springframework.util.Assert;
* Abstract MessageChannel backed by a Kafka topic.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public abstract class AbstractKafkaChannel extends AbstractMessageChannel {

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

@@ -38,7 +38,7 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
*
* @since 3.3
* @since 5.4
*
*/
@SuppressWarnings("deprecation")

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

@@ -26,7 +26,8 @@ import org.springframework.kafka.core.KafkaOperations;
* Publish/subscribe channel backed by a Kafka topic.
*
* @author Gary Russell
* @since 3.3
*
* @since 4.4
*
*/
public class PublishSubscribeKafkaChannel extends SubscribableKafkaChannel implements BroadcastCapableChannel {

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

@@ -37,7 +37,8 @@ import org.springframework.util.Assert;
* Subscribable channel backed by a Kafka topic.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public class SubscribableKafkaChannel extends AbstractKafkaChannel implements SubscribableChannel, SmartLifecycle {

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

@@ -31,7 +31,8 @@ import org.springframework.util.StringUtils;
* Parser for a channel backed by a Kafka topic.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public class KafkaChannelParser extends AbstractChannelParser {

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java
View File

@@ -31,7 +31,8 @@ import org.springframework.util.StringUtils;
*
* @author Gary Russell
* @author Anshul Mehra
* @since 3.2
*
* @since 5.4
*
*/
public class KafkaInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java
View File

@@ -27,7 +27,8 @@ import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
* Inbound gateway parser.
*
* @author Gary Russell
* @since 3.2
*
* @since 5.4
*
*/
public class KafkaInboundGatewayParser extends AbstractInboundGatewayParser {

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

@@ -32,6 +32,8 @@ import org.springframework.util.StringUtils;
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.4
*/
public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapterParser {

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

@@ -23,7 +23,8 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
*
* @author Soby Chacko
* @author Gary Russell
* @since 0.5
*
* @since 5.4
*
*/
public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler {

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java
View File

@@ -31,7 +31,8 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle
* @author Artem Bilan
* @author Gary Russell
* @author Biju Kunjummen
* @since 0.5
*
* @since 5.4
*
*/
public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java
View File

@@ -27,7 +27,8 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle
* Parser for the outbound gateway.
*
* @author Gary Russell
* @since 3.2
*
* @since 5.4
*
*/
public class KafkaOutboundGatewayParser extends AbstractConsumerEndpointParser {

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java
View File

@@ -28,7 +28,8 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
*
* @author Gary Russell
* @author Tom van den Berge
* @since 3.2
*
* @since 5.4
*
*/
public final class KafkaParsingUtils {

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

@@ -27,7 +27,8 @@ import org.springframework.integration.kafka.channel.AbstractKafkaChannel;
* @param <C> the channel type.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public abstract class AbstractKafkaChannelSpec<S extends AbstractKafkaChannelSpec<S, C>, C extends AbstractKafkaChannel>

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

@@ -42,7 +42,7 @@ import org.springframework.kafka.support.TopicPartitionOffset;
* @author Gary Russell
* @author Anshul Mehra
*
* @since 3.0
* @since 5.4
*/
public final class Kafka {

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

@@ -37,7 +37,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter;
* @author Gary Russell
* @author Anshul Mehra
*
* @since 3.0.1
* @since 5.4
*
*/
public class KafkaInboundChannelAdapterSpec<K, V>

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

@@ -47,7 +47,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0.2
* @since 5.4
*/
public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<K, V, R, S>>
extends MessagingGatewaySpec<S, KafkaInboundGateway<K, V, R>>

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

@@ -48,7 +48,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Cameron Mayfield
*
* @since 3.0
* @since 5.4
*/
public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDrivenChannelAdapterSpec<K, V, S>>
extends MessageProducerSpec<S, KafkaMessageDrivenChannelAdapter<K, V>>

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

@@ -42,7 +42,7 @@ import org.springframework.kafka.support.TopicPartitionOffset;
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0
* @since 5.4
*/
public class KafkaMessageListenerContainerSpec<K, V>
extends IntegrationComponentSpec<KafkaMessageListenerContainerSpec<K, V>, ConcurrentMessageListenerContainer<K, V>> {

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

@@ -43,7 +43,7 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
*
* @since 3.0.2
* @since 5.4
*
*/
public class KafkaOutboundGatewaySpec<K, V, R, S extends KafkaOutboundGatewaySpec<K, V, R, S>>

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

@@ -24,7 +24,8 @@ import org.springframework.kafka.core.KafkaTemplate;
* Spec for a point to point channel backed by a Kafka topic.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public class KafkaPointToPointChannelSpec extends KafkaSubscribableChannelSpec<SubscribableKafkaChannel> {

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

@@ -24,7 +24,8 @@ import org.springframework.kafka.core.KafkaTemplate;
* Spec for a pollable channel.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public class KafkaPollableChannelSpec extends AbstractKafkaChannelSpec<KafkaPollableChannelSpec, PollableKafkaChannel> {

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

@@ -46,7 +46,7 @@ import org.springframework.util.Assert;
* @author Biju Kunjummen
* @author Gary Russell
*
* @since 3.0
* @since 5.4
*/
public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessageHandlerSpec<K, V, S>>
extends MessageHandlerSpec<S, KafkaProducerMessageHandler<K, V>> {

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

@@ -24,7 +24,8 @@ import org.springframework.kafka.core.KafkaTemplate;
* Spec for a publish/subscribe channel backed by a Kafka topic.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public class KafkaPublishSubscribeChannelSpec

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

@@ -24,7 +24,8 @@ import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
* @param <C> the channel type.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public abstract class KafkaSubscribableChannelSpec<C extends SubscribableKafkaChannel>

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

@@ -32,7 +32,7 @@ import org.springframework.kafka.support.converter.RecordMessageConverter;
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0
* @since 5.4
*/
public class KafkaTemplateSpec<K, V>
extends IntegrationComponentSpec<KafkaTemplateSpec<K, V>, KafkaTemplate<K, V>> {

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

@@ -67,7 +67,7 @@ import org.springframework.util.Assert;
* @author Artem Bilan
* @author Urs Keller
*
* @since 3.0.2
* @since 5.4
*
*/
public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implements OrderlyShutdownCapable {

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

@@ -73,6 +73,7 @@ import org.springframework.util.Assert;
* @author Artem Bilan
* @author Urs Keller
*
* @since 5.4
*/
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable,
Pausable {

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

@@ -93,7 +93,7 @@ import org.springframework.util.StringUtils;
* @author Artem Bilan
* @author Anshul Mehra
*
* @since 3.0.1
* @since 5.4
*
*/
public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> implements Pausable {

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

@@ -97,7 +97,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* @author Biju Kunjummen
* @author Tom van den Berge
*
* @since 0.5
* @since 5.4
*/
public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMessageHandler
implements Lifecycle {

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java
View File

@@ -22,7 +22,8 @@ import org.springframework.kafka.support.KafkaHeaders;
* Headers specifically for Spring Integration components.
*
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
public final class KafkaIntegrationHeaders {

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java
View File

@@ -26,7 +26,7 @@ import org.springframework.messaging.MessagingException;
*
* @author Gary Russell
*
* @since 2.1.2
* @since 5.4
*
*/
public class KafkaSendFailureException extends MessagingException {

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java
View File

@@ -35,7 +35,7 @@ import org.springframework.messaging.support.ErrorMessage;
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1.1
* @since 5.4
*
*/
public class RawRecordHeaderErrorMessageStrategy implements ErrorMessageStrategy {

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java
View File

@@ -53,7 +53,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
@SpringJUnitConfig

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml
View File

@@ -3,9 +3,9 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
<bean id="cf" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java
View File

@@ -30,7 +30,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @since 3.2
*
* @since 5.4
*
*/
@SpringJUnitConfig

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java
View File

@@ -38,7 +38,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @since 3.3
*
* @since 5.4
*
*/
@SpringJUnitConfig

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml
View File

@@ -3,9 +3,9 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
<int-kafka:inbound-channel-adapter
id="adapter1"

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests.java
View File

@@ -32,7 +32,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @since 3.2
*
* @since 5.4
*
*/
@SpringJUnitConfig

4
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml
View File

@@ -2,8 +2,8 @@
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
<int-kafka:inbound-gateway
id="gateway1"

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java
View File

@@ -30,7 +30,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @since 3.2
*
* @since 5.4
*
*/
@SpringJUnitConfig

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
View File

@@ -45,6 +45,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Artem Bilan.
* @author Gary Russell
*
* @since 5.4
*/
@SpringJUnitConfig
@DirtiesContext

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java
View File

@@ -52,7 +52,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
* @author Biju Kunjummen
* @author Tom van den Berge
*
* @since 0.5
* @since 5.4
*/
@SpringJUnitConfig
@DirtiesContext

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml
View File

@@ -3,9 +3,9 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
<int-kafka:outbound-gateway
id="allProps"

3
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java
View File

@@ -32,7 +32,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @author Tom van den Berge
* @since 3.2
*
* @since 5.4
*
*/
@SpringJUnitConfig

6
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/channels-context.xml
View File

@@ -3,9 +3,9 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
<int-kafka:channel kafka-template="template" id="ptp" topic="ptpTopic" group-id="ptpGroup"
auto-startup="false" container-factory="containerFactory" />

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -89,7 +89,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
* @author Gary Russell
* @author Anshul Mehra
*
* @since 3.0
* @since 5.4
*/
@SpringJUnitConfig
@DirtiesContext

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java
View File

@@ -71,7 +71,7 @@ import org.springframework.retry.support.RetryTemplate;
* @author Artem Bilan
* @author Urs Keller
*
* @since 3.0.2
* @since 5.4
*
*/
class InboundGatewayTests {

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -100,7 +100,7 @@ import org.springframework.retry.support.RetryTemplate;
* @author Cameron Mayfield
* @author Urs Keller
*
* @since 2.0
* @since 5.4
*
*/
class MessageDrivenAdapterTests {

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java
View File

@@ -43,7 +43,7 @@ import org.springframework.messaging.Message;
* @author Artem Bilan
* @author Anshul Mehra
*
* @since 3.0.1
* @since 5.4
*
*/
class MessageSourceIntegrationTests {

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java
View File

@@ -91,7 +91,7 @@ import org.springframework.messaging.Message;
* @author Anshul Mehra
* @author Artem Bilan
*
* @since 3.0.1
* @since 5.4
*
*/
class MessageSourceTests {

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -107,7 +107,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* @author Artem Bilan
* @author Tom van den Berge
*
* @since 2.0
* @since 5.4
*/
class KafkaProducerMessageHandlerTests {

2
spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt
View File

@@ -82,7 +82,7 @@ import java.util.stream.Stream
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0.3
* @since 5.4
*/
@SpringJUnitConfig

1
src/checkstyle/checkstyle.xml
View File

@@ -86,6 +86,7 @@
org.mockito.ArgumentMatchers.*,
org.mockito.AdditionalMatchers.*,
org.springframework.integration.gemfire.config.xml.ParserTestUtil.*,
org.springframework.kafka.test.assertj.KafkaConditions.*,
org.springframework.integration.test.mock.MockIntegration.*,
org.springframework.integration.test.util.TestUtils.*,
org.springframework.test.web.client.match.MockRestRequestMatchers.*,

6
src/reference/asciidoc/endpoint-summary.adoc
View File

@@ -114,6 +114,12 @@ The following table summarizes the various endpoints with quick links to the app
| N
| <<./jpa.adoc#jpa-updating-outbound-gateway,Updating Outbound Gateway>> and <<./jpa.adoc#jpa-retrieving-outbound-gateway,Retrieving Outbound Gateway>>
| *Apache Kafka*
| <<./kafka.adoc#kafka-inbound,Message Driven Channel Adapter>> and <<./kafka.adoc#kafka-inbound-pollable,Inbound Channel Adapter>>
| <<./kafka.adoc#kafka-outbound,Outbound Channel Adapter>>
| <<./kafka.adoc#kafka-inbound-gateway,Inbound Gateway>>
| <<./kafka.adoc#kafka-outbound-gateway,Outbound Gateway>>
| *Mail*
| <<./mail.adoc#mail-inbound,Mail-receiving Channel Adapter>>
| <<./mail.adoc#mail-outbound,Mail-sending Channel Adapter>>

2
src/reference/asciidoc/index-single.adoc
View File

@@ -49,6 +49,8 @@ include::./jpa.adoc[]
include::./jms.adoc[]
include::./kafka.adoc[]
include::./mail.adoc[]
include::./mongodb.adoc[]

1
src/reference/asciidoc/index.adoc
View File

@@ -34,6 +34,7 @@ This documentation is also available as single searchable link:index-single.html
<<./jdbc.adoc#jdbc,JDBC Support>> ::
<<./jpa.adoc#jpa,JPA Support>> ::
<<./jms.adoc#jms,JMS Support>> ::
<<./kafka.adoc#kafka,Apache Kafka Support>> ::
<<./mail.adoc#mail,Mail Support>> ::
<<./mongodb.adoc#mongodb,MongoDb Support>> ::
<<./mqtt.adoc#mqtt,MQTT Support>> ::

841
src/reference/asciidoc/kafka.adoc Normal file
View File

@@ -0,0 +1,841 @@
[[kafka]]
== Spring for Apache Kafka Support
=== Overview
Spring Integration for Apache Kafka is based on the https://projects.spring.io/spring-kafka/[Spring for Apache Kafka project].
It provides the following components:
* <<kafka-outbound>>
* <<kafka-inbound>>
* <<kafka-inbound-pollable>>
* <<kafka-outbound-gateway>>
* <<kafka-inbound-gateway>>
* <<kafka-channels>>
[[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 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.
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:
* `topic` and `topic-expression`
* `message-key` and `message-key-expression`
* `partition-id` and `partition-id-expression`
These let you specify `topic`, `message-key`, and `partition-id`, respectively, as static values on the adapter or to dynamically evaluate their values at runtime against the request message.
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`.
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.
If you wish the header to override the configuration, you need to configure it in an expression, such as the following:
====
[source,java]
----
topic-expression="headers['topic'] != null ? headers['topic'] : 'myTopic'"
----
====
The adapter requires a `KafkaTemplate`, which, in turn, requires a suitably configured `KafkaProducerFactory`.
If a `send-failure-channel` (`sendFailureChannel`) is provided and a send failure (sync or async) is received, an `ErrorMessage` is sent to the channel.
The payload is a `KafkaSendFailureException` with `failedMessage`, `record` (the `ProducerRecord`) and `cause` properties.
You can override the `DefaultErrorMessageStrategy` by setting the `error-message-strategy` property.
If a `send-success-channel` (`sendSuccessChannel`) is provided, a message with a payload of type `org.apache.kafka.clients.producer.RecordMetadata` is sent after a successful send.
IMPORTANT: If your application uses transactions and the same channel adapter is used to publish messages where the transaction is started by a listener container, as well as publishing where there is no existing transaction, you must configure a `transactionIdPrefix` on the `KafkaTemplate` to override the prefix used by the container or transaction manager.
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.
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.
==== Java Configuration
The following example shows how to configure the Kafka outbound channel adapter with Java:
====
[source, java]
----
@Bean
@ServiceActivator(inputChannel = "toKafka")
public MessageHandler handler() throws Exception {
KafkaProducerMessageHandler<String, String> handler =
new KafkaProducerMessageHandler<>(kafkaTemplate());
handler.setTopicExpression(new LiteralExpression("someTopic"));
handler.setMessageKeyExpression(new LiteralExpression("someKey"));
handler.setSuccessChannel(successes());
handler.setFailureChannel(failures());
return handler;
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
// set more properties
return new DefaultKafkaProducerFactory<>(props);
}
----
====
==== Java DSL Configuration
The following example shows how to configure the Kafka outbound channel adapter Spring Integration Java DSL:
====
[source, java]
----
@Bean
public ProducerFactory<Integer, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka));
}
@Bean
public IntegrationFlow sendToKafkaFlow() {
return f -> f
.<String>split(p -> Stream.generate(() -> p).limit(101).iterator(), null)
.publishSubscribeChannel(c -> c
.subscribe(sf -> sf.handle(
kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
.timestampExpression("T(Long).valueOf('1487694048633')"),
e -> e.id("kafkaProducer1")))
.subscribe(sf -> sf.handle(
kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
.timestamp(m -> 1487694048644L),
e -> e.id("kafkaProducer2")))
);
}
@Bean
public DefaultKafkaHeaderMapper mapper() {
return new DefaultKafkaHeaderMapper();
}
private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
ProducerFactory<Integer, String> producerFactory, String topic) {
return Kafka
.outboundChannelAdapter(producerFactory)
.messageKey(m -> m
.getHeaders()
.get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
.headerMapper(mapper())
.partitionId(m -> 10)
.topicExpression("headers[kafka_topic] ?: '" + topic + "'")
.configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
}
----
====
==== XML Configuration
The following example shows how to configure the Kafka outbound channel adapter with XML:
====
[source, xml]
----
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapter"
kafka-template="template"
auto-startup="false"
channel="inputToKafka"
topic="foo"
sync="false"
message-key-expression="'bar'"
send-failure-channel="failures"
send-success-channel="successes"
error-message-strategy="ems"
partition-id-expression="2">
</int-kafka:outbound-channel-adapter>
<bean id="template" class="org.springframework.kafka.core.KafkaTemplate">
<constructor-arg>
<bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
<constructor-arg>
<map>
<entry key="bootstrap.servers" value="localhost:9092" />
... <!-- more producer properties -->
</map>
</constructor-arg>
</bean>
</constructor-arg>
</bean>
----
====
[[kafka-inbound]]
=== Message-driven 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.
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.
As with the batched `@KafkaListener`, the `KafkaHeaders.RECEIVED_MESSAGE_KEY`, `KafkaHeaders.RECEIVED_PARTITION_ID`, `KafkaHeaders.RECEIVED_TOPIC`, and `KafkaHeaders.OFFSET` headers are also lists, with positions corresponding to the position in the payload.
Received messages have certain headers populated.
See the https://docs.spring.io/spring-kafka/api/org/springframework/kafka/support/KafkaHeaders.html[`KafkaHeaders` class] for more information.
IMPORTANT: The `Consumer` object (in the `kafka_consumer` header) is not thread-safe.
You must invoke its methods only on the thread that calls the listener within the adapter.
If you hand off the message to another thread, you must not call its methods.
When a `retry-template` is provided, delivery failures are retried according to its retry policy.
An `error-channel` is not allowed in this case.
You can use the `recovery-callback` to handle the error when retries are exhausted.
In most cases, this is an `ErrorMessageSendingRecoverer` that sends the `ErrorMessage` to a channel.
When building an `ErrorMessage` (for use in the `error-channel` or `recovery-callback`), you can customize the error message by setting the `error-message-strategy` property.
By default, a `RawRecordHeaderErrorMessageStrategy` is used, to provide access to the converted message as well as the raw `ConsumerRecord`.
==== Java Configuration
The following example shows how to configure a message-driven channel adapter with Java:
====
[source, java]
----
@Bean
public KafkaMessageDrivenChannelAdapter<String, String>
adapter(KafkaMessageListenerContainer<String, String> container) {
KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(container, ListenerMode.record);
kafkaMessageDrivenChannelAdapter.setOutputChannel(received());
return kafkaMessageDrivenChannelAdapter;
}
@Bean
public KafkaMessageListenerContainer<String, String> container() throws Exception {
ContainerProperties properties = new ContainerProperties(this.topic);
// set more properties
return new KafkaMessageListenerContainer<>(consumerFactory(), properties);
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
// set more properties
return new DefaultKafkaConsumerFactory<>(props);
}
----
====
==== Java DSL Configuration
The following example shows how to configure a message-driven channel adapter with the Spring Integration Java DSL:
====
[source, java]
----
@Bean
public IntegrationFlow topic1ListenerFromKafkaFlow() {
return IntegrationFlows
.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
.configureListenerContainer(c ->
c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)
.id("topic1ListenerContainer"))
.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()))
.retryTemplate(new RetryTemplate())
.filterInRetry(true))
.filter(Message.class, m ->
m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
f -> f.throwExceptionOnRejection(true))
.<String, String>transform(String::toUpperCase)
.channel(c -> c.queue("listeningFromKafkaResults1"))
.get();
}
----
====
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.
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.
The following example shows how to do so:
====
[source, java]
----
@Bean
public IntegrationFlow topic2ListenerFromKafkaFlow() {
return IntegrationFlows
.from(Kafka.messageDrivenChannelAdapter(kafkaListenerContainerFactory().createContainer(TEST_TOPIC2),
KafkaMessageDrivenChannelAdapter.ListenerMode.record)
.id("topic2Adapter"))
...
get();
}
----
====
Notice that, in this case, the adapter is given an `id` (`topic2Adapter`).
The container is registered in the application context with a name of `topic2Adapter.container`.
If the adapter does not have an `id` property, the container's bean name is the container's fully qualified class name plus `#n`, where `n` is incremented for each container.
==== XML Configuration
The following example shows how to configure a message-driven channel adapter with XML:
====
[source, xml]
----
<int-kafka:message-driven-channel-adapter
id="kafkaListener"
listener-container="container1"
auto-startup="false"
phase="100"
send-timeout="5000"
mode="record"
retry-template="template"
recovery-callback="callback"
error-message-strategy="ems"
channel="someChannel"
error-channel="errorChannel" />
<bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
<constructor-arg>
<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<map>
<entry key="bootstrap.servers" value="localhost:9092" />
...
</map>
</constructor-arg>
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.kafka.listener.config.ContainerProperties">
<constructor-arg name="topics" value="foo" />
</bean>
</constructor-arg>
</bean>
----
====
[[kafka-inbound-pollable]]
=== Inbound Channel Adapter
Introduced in version 3.0.1, the `KafkaMessageSource` provides a pollable channel adapter implementation.
==== Java Configuration
====
[source, java]
----
@InboundChannelAdapter(channel = "fromKafka", poller = @Poller(fixedDelay = "5000"))
@Bean
public KafkaMessageSource<String, String> source(ConsumerFactory<String, String> cf) {
KafkaMessageSource<String, String> source = new KafkaMessageSource<>(cf, "myTopic");
source.setGroupId("myGroupId");
source.setClientId("myClientId");
return source;
}
----
====
Refer to the javadocs for available properties.
[[max-poll-records]]
By default, `max.poll.records` must be either explicitly set in the consumer factory, or it will be forced to 1 if the consumer factory is a `DefaultKafkaConsumerFactory`.
Starting with version 3.2, you can set the property `allowMultiFetch` to `true` to override this behavior.
IMPORTANT: You must poll the consumer within `max.poll.interval.ms` to avoid a rebalance.
If you set `allowMultiFetch` to `true` you must process all the retrieved records, and poll again, within `max.poll.interval.ms`.
Messages emitted by this adapter contain a header `kafka_remainingRecords` with a count of records remaining from the previous poll.
==== Java DSL Configuration
====
[source, java]
----
@Bean
public IntegrationFlow flow(ConsumerFactory<String, String> cf) {
return IntegrationFlows.from(Kafka.inboundChannelAdapter(cf, "myTopic")
.groupId("myDslGroupId"), e -> e.poller(Pollers.fixedDelay(5000)))
.handle(System.out::println)
.get();
}
----
====
==== XML Configuration
====
[source, xml]
----
<int-kafka:inbound-channel-adapter
id="adapter1"
consumer-factory="consumerFactory"
ack-factory="ackFactory"
topics="topic1"
channel="inbound"
client-id="client"
group-id="group"
message-converter="converter"
payload-type="java.lang.String"
raw-header="true"
auto-startup="false"
rebalance-listener="rebal">
<int:poller fixed-delay="5000"/>
</int-kafka:inbound-channel-adapter>
----
====
[[kafka-outbound-gateway]]
=== Outbound Gateway
The outbound gateway is for request/reply operations.
It differs from most Spring Integration gateways in that the sending thread does not block in the gateway and the reply is processed on the reply listener container thread.
If your code invokes the gateway behind a synchronous https://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#gateway[Messaging Gateway], the user thread blocks there until the reply is received (or a timeout occurs).
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.
==== Java Configuration
The following example shows how to configure a gateway with Java:
====
[source, java]
----
@Bean
@ServiceActivator(inputChannel = "kafkaRequests", outputChannel = "kafkaReplies")
public KafkaProducerMessageHandler<String, String> outGateway(
ReplyingKafkaTemplate<String, String, String> kafkaTemplate) {
return new KafkaProducerMessageHandler<>(kafkaTemplate);
}
----
====
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`.
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.
The reply topic is determined as follows:
. A message header named `KafkaHeaders.REPLY_TOPIC` (if present, it must have a `String` or `byte[]` value) is validated against the template's reply container's subscribed topics.
. If the template's `replyContainer` is subscribed to only one topic, it is used.
You can also specify a `KafkaHeaders.REPLY_PARTITION` header to determine a specific partition to be used for replies.
Again, this is validated against the template's reply container's subscriptions.
==== Java DSL Configuration
The following example shows how to configure an outbound gateway with the Java DSL:
====
[source, java]
----
@Bean
public IntegrationFlow outboundGateFlow(
ReplyingKafkaTemplate<String, String, String> kafkaTemplate) {
return IntegrationFlows.from("kafkaRequests")
.handle(Kafka.outboundGateway(kafkaTemplate))
.channel("kafkaReplies")
.get();
}
----
====
Alternatively, you can also use a configuration similar to the following bean:
====
[source, java]
----
@Bean
public IntegrationFlow outboundGateFlow() {
return IntegrationFlows.from("kafkaRequests")
.handle(Kafka.outboundGateway(producerFactory(), replyContainer())
.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
.channel("kafkaReplies")
.get();
}
----
====
==== XML Configuration
====
[source, xml]
----
<int-kafka:outbound-gateway
id="allProps"
error-message-strategy="ems"
kafka-template="template"
message-key-expression="'key'"
order="23"
partition-id-expression="2"
reply-channel="replies"
reply-timeout="43"
request-channel="requests"
requires-reply="false"
send-success-channel="successes"
send-failure-channel="failures"
send-timeout-expression="44"
sync="true"
timestamp-expression="T(System).currentTimeMillis()"
topic-expression="'topic'"/>
----
====
[[kafka-inbound-gateway]]
=== Inbound Gateway
The inbound gateway is for request/reply operations.
The following example shows how to configure an inbound gateway with Java:
====
[source, java]
----
@Bean
public KafkaInboundGateway<Integer, String, String> inboundGateway(
AbstractMessageListenerContainer<Integer, String>container,
KafkaTemplate<Integer, String> replyTemplate) {
KafkaInboundGateway<Integer, String, String> gateway =
new KafkaInboundGateway<>(container, replyTemplate);
gateway.setRequestChannel(requests);
gateway.setReplyChannel(replies);
gateway.setReplyTimeout(30_000);
return gateway;
}
----
====
Refer to the javadocs for available properties.
The following example shows how to configure a simple upper case converter with the Java DSL:
====
[source, java]
----
@Bean
public IntegrationFlow serverGateway(
ConcurrentMessageListenerContainer<Integer, String> container,
KafkaTemplate<Integer, String> replyTemplate) {
return IntegrationFlows
.from(Kafka.inboundGateway(container, template)
.replyTimeout(30_000))
.<String, String>transform(String::toUpperCase)
.get();
}
----
====
Alternatively, you could configure an upper-case converter by using code similar to the following:
====
[source, java]
----
@Bean
public IntegrationFlow serverGateway() {
return IntegrationFlows
.from(Kafka.inboundGateway(consumerFactory(), containerProperties(),
producerFactory())
.replyTimeout(30_000))
.<String, String>transform(String::toUpperCase)
.get();
}
----
====
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.
See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] and <<kafka-inbound>> for examples.
==== XML Configuration
====
[source, xml]
----
<int-kafka:inbound-gateway
id="gateway1"
listener-container="container1"
kafka-template="template"
auto-startup="false"
phase="100"
request-timeout="5000"
request-channel="nullChannel"
reply-channel="errorChannel"
reply-timeout="43"
message-converter="messageConverter"
payload-type="java.lang.String"
error-message-strategy="ems"
retry-template="retryTemplate"
recovery-callback="recoveryCallback"/>
----
====
See the XML schema for a description of each property.
[[kafka-channels]]
=== Channels Backed by Kafka Topics
Spring Integration for Apache Kafka version 3.3 (still under development) introduces channels backed by a 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.
==== Java DSL Configuration
====
[source, java]
----
@Bean
public IntegrationFlow flowWithSubscribable(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return IntegrationFlows.from(...)
...
.channel(Kafka.channel(template, containerFactory, "someTopic1").groupId("group1"))
...
.get();
}
@Bean
public IntegrationFlow flowWithPubSub(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return IntegrationFlows.from(...)
...
.publishSubscribeChannel(pubSub(template, containerFactory),
pubsub -> pubsub
.subscribe(subflow -> ...)
.subscribe(subflow -> ...))
.get();
}
@Bean
public BroadcastCapableChannel pubSub(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return Kafka.publishSubscribeChannel(template, containerFactory, "someTopic2")
.groupId("group2")
.get();
}
@Bean
public IntegrationFlow flowWithPollable(KafkaTemplate<Integer, String> template,
KafkaMessageSource<Integer, String> source) {
return IntegrationFlows.from(...)
...
.channel(Kafka.pollableChannel(template, source, "someTopic3").greoupId("group3"))
.handle(..., e -> e.poller(...))
...
.get();
}
----
====
==== Java Configuration
====
[source, java]
----
/**
* Channel for a single subscriber.
**/
@Bean
SubscribableKafkaChannel pointToPoint(KafkaTemplate<String, String> template,
KafkaListenerContainerFactory<String, String> factory)
SubscribableKafkaChannel channel =
new SubscribableKafkaChannel(template, factory, "topicA");
channel.setGroupId("group1");
return channel;
}
/**
* Channel for multiple subscribers.
**/
@Bean
SubscribableKafkaChannel pubsub(KafkaTemplate<String, String> template,
KafkaListenerContainerFactory<String, String> factory)
SubscribableKafkaChannel channel =
new SubscribableKafkaChannel(template, factory, "topicB", true);
channel.setGroupId("group2");
return channel;
}
/**
* Pollable channel (topic is configured on the source)
**/
@Bean
PollableKafkaChannel pollable(KafkaTemplate<String, String> template,
KafkaMessageSource<String, String> source)
PollableKafkaChannel channel =
new PollableKafkaChannel(template, source);
channel.setGroupId("group3");
return channel;
}
----
====
==== XML Configuration
====
[source, xml]
----
<int-kafka:channel kafka-template="template" id="ptp" topic="ptpTopic" group-id="ptpGroup"
container-factory="containerFactory" />
<int-kafka:pollable-channel kafka-template="template" id="pollable" message-source="source"
group-id = "pollableGroup"/>
<int-kafka:publish-subscribe-channel kafka-template="template" id="pubSub" topic="pubSubTopic"
group-id="pubSubGroup" container-factory="containerFactory" />
----
====
[[message-conversion]]
=== Message Conversion
A `StringJsonMessageConverter` is provided.
See https://docs.spring.io/spring-kafka/docs/current/reference/html/[the Spring for Apache Kafka documentation] for more information.
When using this converter with a message-driven channel adapter, you can specify the type to which you want the incoming payload to be converted.
This is achieved by setting the `payload-type` attribute (`payloadType` property) on the adapter.
The following example shows how to do so in XML configuration:
====
[source, xml]
----
<int-kafka:message-driven-channel-adapter
id="kafkaListener"
listener-container="container1"
auto-startup="false"
phase="100"
send-timeout="5000"
channel="nullChannel"
message-converter="messageConverter"
payload-type="com.example.Foo"
error-channel="errorChannel" />
<bean id="messageConverter"
class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
----
====
The following example shows how to set the `payload-type` attribute (`payloadType` property) on the adapter in Java configuration:
====
[source, java]
----
@Bean
public KafkaMessageDrivenChannelAdapter<String, String>
adapter(KafkaMessageListenerContainer<String, String> container) {
KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter =
new KafkaMessageDrivenChannelAdapter<>(container, ListenerMode.record);
kafkaMessageDrivenChannelAdapter.setOutputChannel(received());
kafkaMessageDrivenChannelAdapter.setMessageConverter(converter());
kafkaMessageDrivenChannelAdapter.setPayloadType(Foo.class);
return kafkaMessageDrivenChannelAdapter;
}
----
====
[[kafka-tombstones]]
=== 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`.
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.
To do so, mark the parameter with `@Payload(required = false)`.
The following example shows how to do so:
====
[source, java]
----
@ServiceActivator(inputChannel = "fromSomeKafkaInboundEndpoint")
public void in(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
@Payload(required = false) Customer customer) {
// customer is null if a tombstone record
...
}
----
====
[[streams-integration]]
=== Calling a Spring Integration flow from a `KStream`
You can use a `MessagingTransformer` to invoke an integration flow from a `KStream`:
====
[source, java]
----
@Bean
public KStream<byte[], byte[]> kStream(StreamsBuilder kStreamBuilder,
MessagingTransformer<byte[], byte[], byte[]> transformer) transformer) {
KStream<byte[], byte[]> stream = kStreamBuilder.stream(STREAMING_TOPIC1);
stream.mapValues((ValueMapper<byte[], byte[]>) String::toUpperCase)
...
.transform(() -> transformer)
.to(streamingTopic2);
stream.print(Printed.toSysOut());
return stream;
}
@Bean
@DependsOn("flow")
public MessagingTransformer<byte[], byte[], String> transformer(
MessagingFunction function) {
MessagingMessageConverter converter = new MessagingMessageConverter();
converter.setHeaderMapper(new SimpleKafkaHeaderMapper("*"));
return new MessagingTransformer<>(function, converter);
}
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from(MessagingFunction.class)
...
.get();
}
----
====
When an integration flow starts with an interface, the proxy that is created has the name of the flow bean, appended with ".gateway" so this bean name can be used a a `@Qualifier` if needed.

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

@@ -57,7 +57,7 @@ For more information regarding messaging endpoints in general and polling consum
==== Deferred Acknowledgment Pollable Message Source
Starting with version 5.0.1, certain modules provide `MessageSource` implementations that support deferring acknowledgment until the downstream flow completes (or hands off the message to another thread).
This is currently limited to the `AmqpMessageSource` and the `KafkaMessageSource` provided by the `spring-integration-kafka` https://github.com/spring-projects/spring-integration-kafka[extension project].
This is currently limited to the `AmqpMessageSource` and the `KafkaMessageSource`.
With these message sources, the `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header (see <<./message.adoc#message-header-accessor,`MessageHeaderAccessor` API>>) is added to the message.
When used with pollable message sources, the value of the header is an instance of `AcknowledgmentCallback`, as the following example shows:

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -15,6 +15,11 @@ If you are interested in more details, see the Issue Tracker tickets that were r
[[x5.4-new-components]]
=== 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.
See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information.
[[x5.4-general]]
=== General Changes