diff --git a/build.gradle b/build.gradle index 3f234324e7..ae334510a7 100644 --- a/build.gradle +++ b/build.gradle @@ -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 { diff --git a/spring-integration-kafka/src/api/overview.html b/spring-integration-kafka/src/api/overview.html deleted file mode 100644 index 8b6c313b36..0000000000 --- a/spring-integration-kafka/src/api/overview.html +++ /dev/null @@ -1,22 +0,0 @@ - - -This document is the API specification for Spring Integration Kafka Extension -
-
-

- For further API reference and developer documentation, see the - Spring - Integration reference documentation. - That documentation contains more detailed, developer-targeted - descriptions, with conceptual overviews, definitions of terms, - workarounds, and working code examples. -

- -

- If you are interested in commercial training, consultancy, and - support for Spring Integration, please visit - https://www.springsource.com -

-
- - diff --git a/spring-integration-kafka/src/checkstyle/checkstyle-header.txt b/spring-integration-kafka/src/checkstyle/checkstyle-header.txt deleted file mode 100644 index 7cd20a6770..0000000000 --- a/spring-integration-kafka/src/checkstyle/checkstyle-header.txt +++ /dev/null @@ -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$ -^$ -^.*$ diff --git a/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml b/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml deleted file mode 100644 index a6301059dd..0000000000 --- a/spring-integration-kafka/src/checkstyle/checkstyle-suppressions.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - diff --git a/spring-integration-kafka/src/checkstyle/checkstyle.xml b/spring-integration-kafka/src/checkstyle/checkstyle.xml deleted file mode 100644 index e8e51d9979..0000000000 --- a/spring-integration-kafka/src/checkstyle/checkstyle.xml +++ /dev/null @@ -1,180 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-kafka/src/dist/license.txt b/spring-integration-kafka/src/dist/license.txt deleted file mode 100644 index 7c50aab067..0000000000 --- a/spring-integration-kafka/src/dist/license.txt +++ /dev/null @@ -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. diff --git a/spring-integration-kafka/src/dist/notice.txt b/spring-integration-kafka/src/dist/notice.txt deleted file mode 100644 index 607cea5e0a..0000000000 --- a/spring-integration-kafka/src/dist/notice.txt +++ /dev/null @@ -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. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java index d397685e40..233829fc4c 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/AbstractKafkaChannel.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java index 2ebf72e82a..85d4513ea2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java @@ -38,7 +38,7 @@ import org.springframework.util.Assert; * * @author Gary Russell * - * @since 3.3 + * @since 5.4 * */ @SuppressWarnings("deprecation") diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java index 60c8dab9ba..8c1ae3989c 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PublishSubscribeKafkaChannel.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java index 166586b936..a3adbe2b1d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/SubscribableKafkaChannel.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java index fa071f5745..7fce490fc0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaChannelParser.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java index 1bf4ddf505..8f94a5a072 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java index 1f835a192b..23d8833abc 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java index c7c3cba283..20560c1de6 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java @@ -32,6 +32,8 @@ import org.springframework.util.StringUtils; * * @author Artem Bilan * @author Gary Russell + * + * @since 5.4 */ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapterParser { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java index d3fc6b01f1..d17d047ad5 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java index 3e4a774301..85f006d714 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java index 58d7d3e830..e360594ca7 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParser.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java index c8d6a2f3c2..7aff34ac58 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaParsingUtils.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java index 2684dcdfa9..0ea8ccc865 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/AbstractKafkaChannelSpec.java @@ -27,7 +27,8 @@ import org.springframework.integration.kafka.channel.AbstractKafkaChannel; * @param the channel type. * * @author Gary Russell - * @since 3.3 + * + * @since 5.4 * */ public abstract class AbstractKafkaChannelSpec, C extends AbstractKafkaChannel> diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java index 34aa81e362..4006994878 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java index a6c2943d58..a3e37696bb 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java @@ -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 diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java index 5976df3b9d..3a80bfa31e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java @@ -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> extends MessagingGatewaySpec> diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java index 368e60c55d..4196e6789a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java @@ -48,7 +48,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Cameron Mayfield * - * @since 3.0 + * @since 5.4 */ public class KafkaMessageDrivenChannelAdapterSpec> extends MessageProducerSpec> diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java index de596ece25..13e57521cc 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java @@ -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 extends IntegrationComponentSpec, ConcurrentMessageListenerContainer> { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java index 0b01d01ca5..bdf32545e2 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java @@ -43,7 +43,7 @@ import org.springframework.util.Assert; * * @author Gary Russell * - * @since 3.0.2 + * @since 5.4 * */ public class KafkaOutboundGatewaySpec> diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java index 65856b87f8..43e4d4ca9f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPointToPointChannelSpec.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java index 18ddd3d0ec..72b2444a8c 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPollableChannelSpec.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index 13fcddd19a..08aef2d721 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -46,7 +46,7 @@ import org.springframework.util.Assert; * @author Biju Kunjummen * @author Gary Russell * - * @since 3.0 + * @since 5.4 */ public class KafkaProducerMessageHandlerSpec> extends MessageHandlerSpec> { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java index 3af04029d0..82c925c773 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaPublishSubscribeChannelSpec.java @@ -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 diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java index 7d4cf11528..ee2d29362d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaSubscribableChannelSpec.java @@ -24,7 +24,8 @@ import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; * @param the channel type. * * @author Gary Russell - * @since 3.3 + * + * @since 5.4 * */ public abstract class KafkaSubscribableChannelSpec diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java index d96bb71152..9d6bbd15a1 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java @@ -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 extends IntegrationComponentSpec, KafkaTemplate> { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index 9d51c1ef1d..0dedba8223 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -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 extends MessagingGatewaySupport implements OrderlyShutdownCapable { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 7d2255dfa4..d992775979 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -73,6 +73,7 @@ import org.springframework.util.Assert; * @author Artem Bilan * @author Urs Keller * + * @since 5.4 */ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSupport implements OrderlyShutdownCapable, Pausable { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index d238582c85..09a4244887 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -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 extends AbstractMessageSource implements Pausable { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index c31bdee2f6..0eac0b95b0 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -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 extends AbstractReplyProducingMessageHandler implements Lifecycle { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java index 557705f7d8..39bb3ccaaa 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java @@ -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 { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java index c448070304..0f223d1431 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaSendFailureException.java @@ -26,7 +26,7 @@ import org.springframework.messaging.MessagingException; * * @author Gary Russell * - * @since 2.1.2 + * @since 5.4 * */ public class KafkaSendFailureException extends MessagingException { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java index 93f970691c..552fb6ba5f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/RawRecordHeaderErrorMessageStrategy.java @@ -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 { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java index c20c1c22ba..16f475dce2 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java @@ -53,7 +53,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell - * @since 3.3 + * + * @since 5.4 * */ @SpringJUnitConfig diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml index 9cec89bea7..581f84cee6 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml @@ -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"> diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java index f5fe1a059f..1660205ebb 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java @@ -30,7 +30,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell - * @since 3.2 + * + * @since 5.4 * */ @SpringJUnitConfig diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java index 45fe258413..5e5c63357e 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java @@ -38,7 +38,8 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell - * @since 3.3 + * + * @since 5.4 * */ @SpringJUnitConfig diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml index df661cb319..c43d635507 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml @@ -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"> + 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"> + 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"> + 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"> diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index fc7786bed5..9ad8aec0ea 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -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 diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index 44b47d9236..1702ad0f51 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -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 { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 208122f0af..953dd27ba4 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -100,7 +100,7 @@ import org.springframework.retry.support.RetryTemplate; * @author Cameron Mayfield * @author Urs Keller * - * @since 2.0 + * @since 5.4 * */ class MessageDrivenAdapterTests { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java index 5d27ec5a2d..bd7b542b73 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -43,7 +43,7 @@ import org.springframework.messaging.Message; * @author Artem Bilan * @author Anshul Mehra * - * @since 3.0.1 + * @since 5.4 * */ class MessageSourceIntegrationTests { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index 764d487265..0e97bb0da6 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -91,7 +91,7 @@ import org.springframework.messaging.Message; * @author Anshul Mehra * @author Artem Bilan * - * @since 3.0.1 + * @since 5.4 * */ class MessageSourceTests { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 058a31f28a..9ae0b49aec 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -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 { diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt index b12837cf6e..45f87141c6 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt @@ -82,7 +82,7 @@ import java.util.stream.Stream * @author Artem Bilan * @author Gary Russell * - * @since 3.0.3 + * @since 5.4 */ @SpringJUnitConfig diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index f98f6f3c35..6a12e3b499 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -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.*, diff --git a/src/reference/asciidoc/endpoint-summary.adoc b/src/reference/asciidoc/endpoint-summary.adoc index 021bc6ef65..ee6323a254 100644 --- a/src/reference/asciidoc/endpoint-summary.adoc +++ b/src/reference/asciidoc/endpoint-summary.adoc @@ -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>> diff --git a/src/reference/asciidoc/index-single.adoc b/src/reference/asciidoc/index-single.adoc index 6b87044bf8..47ee42c766 100644 --- a/src/reference/asciidoc/index-single.adoc +++ b/src/reference/asciidoc/index-single.adoc @@ -49,6 +49,8 @@ include::./jpa.adoc[] include::./jms.adoc[] +include::./kafka.adoc[] + include::./mail.adoc[] include::./mongodb.adoc[] diff --git a/src/reference/asciidoc/index.adoc b/src/reference/asciidoc/index.adoc index 66f170f585..dbd17d66ce 100644 --- a/src/reference/asciidoc/index.adoc +++ b/src/reference/asciidoc/index.adoc @@ -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>> :: diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc new file mode 100644 index 0000000000..a1022e5c06 --- /dev/null +++ b/src/reference/asciidoc/kafka.adoc @@ -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]] +=== 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 `` 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 +``. +Alternatively, you can change the headers upstream to +the new headers from `KafkaHeaders` by using a `` or a `MessageBuilder`. +If you use constant values, you can also configure them on the adapter by using `topic` and `message-key`. + +NOTE : If the adapter is configured with a topic or message key (either with a constant or expression), those are used +and the corresponding header is ignored. +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 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 kafkaTemplate() { + return new KafkaTemplate<>(producerFactory()); +} + +@Bean +public ProducerFactory producerFactory() { + Map 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 producerFactory() { + return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka)); +} + +@Bean +public IntegrationFlow sendToKafkaFlow() { + return f -> f + .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 kafkaMessageHandler( + ProducerFactory 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] +---- + + + + + + + + + + ... + + + + + +---- +==== + +[[kafka-inbound]] +=== Message-driven Channel Adapter + +The `KafkaMessageDrivenChannelAdapter` (``) 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 + adapter(KafkaMessageListenerContainer container) { + KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = + new KafkaMessageDrivenChannelAdapter<>(container, ListenerMode.record); + kafkaMessageDrivenChannelAdapter.setOutputChannel(received()); + return kafkaMessageDrivenChannelAdapter; +} + +@Bean +public KafkaMessageListenerContainer container() throws Exception { + ContainerProperties properties = new ContainerProperties(this.topic); + // set more properties + return new KafkaMessageListenerContainer<>(consumerFactory(), properties); +} + +@Bean +public ConsumerFactory consumerFactory() { + Map 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)) + .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] +---- + + + + + + + + + ... + + + + + + + + + + + +---- +==== + +[[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 source(ConsumerFactory cf) { + KafkaMessageSource 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 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] +---- + + + +---- +==== + +[[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 outGateway( + ReplyingKafkaTemplate kafkaTemplate) { + return new KafkaProducerMessageHandler<>(kafkaTemplate); +} +---- +==== + +Refer to the javadocs for available properties. + +Notice that the same class as the <> is used, the only difference being that the Kafka template passed into the constructor is a `ReplyingKafkaTemplate`. +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 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] +---- + +---- +==== + +[[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 inboundGateway( + AbstractMessageListenerContainercontainer, + KafkaTemplate replyTemplate) { + + KafkaInboundGateway 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 container, + KafkaTemplate replyTemplate) { + return IntegrationFlows + .from(Kafka.inboundGateway(container, template) + .replyTimeout(30_000)) + .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)) + .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 <> for examples. + +==== XML Configuration + +==== +[source, xml] +---- + +---- +==== + +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 template, + ConcurrentKafkaListenerContainerFactory containerFactory) { + + return IntegrationFlows.from(...) + ... + .channel(Kafka.channel(template, containerFactory, "someTopic1").groupId("group1")) + ... + .get(); +} + +@Bean +public IntegrationFlow flowWithPubSub(KafkaTemplate template, + ConcurrentKafkaListenerContainerFactory containerFactory) { + + return IntegrationFlows.from(...) + ... + .publishSubscribeChannel(pubSub(template, containerFactory), + pubsub -> pubsub + .subscribe(subflow -> ...) + .subscribe(subflow -> ...)) + .get(); +} + +@Bean +public BroadcastCapableChannel pubSub(KafkaTemplate template, + ConcurrentKafkaListenerContainerFactory containerFactory) { + + return Kafka.publishSubscribeChannel(template, containerFactory, "someTopic2") + .groupId("group2") + .get(); +} + +@Bean +public IntegrationFlow flowWithPollable(KafkaTemplate template, + KafkaMessageSource 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 template, + KafkaListenerContainerFactory factory) + + SubscribableKafkaChannel channel = + new SubscribableKafkaChannel(template, factory, "topicA"); + channel.setGroupId("group1"); + return channel; +} + +/** + * Channel for multiple subscribers. + **/ +@Bean +SubscribableKafkaChannel pubsub(KafkaTemplate template, + KafkaListenerContainerFactory 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 template, + KafkaMessageSource source) + + PollableKafkaChannel channel = + new PollableKafkaChannel(template, source); + channel.setGroupId("group3"); + return channel; +} +---- +==== + +==== XML Configuration + +==== +[source, xml] +---- + + + + + +---- +==== + +[[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] +---- + + + + +---- +==== + +The following example shows how to set the `payload-type` attribute (`payloadType` property) on the adapter in Java configuration: + +==== +[source, java] +---- +@Bean +public KafkaMessageDrivenChannelAdapter + adapter(KafkaMessageListenerContainer container) { + KafkaMessageDrivenChannelAdapter 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 kStream(StreamsBuilder kStreamBuilder, + MessagingTransformer transformer) transformer) { + KStream stream = kStreamBuilder.stream(STREAMING_TOPIC1); + stream.mapValues((ValueMapper) String::toUpperCase) + ... + .transform(() -> transformer) + .to(streamingTopic2); + + stream.print(Printed.toSysOut()); + + return stream; +} + +@Bean +@DependsOn("flow") +public MessagingTransformer 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. diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index e6303e194e..303f5b7b64 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -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: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 9c68547617..e523d9ba9b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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