From dfae1a2e7dea89276c5a186d600cb560cea2935a Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Mon, 20 Mar 2023 10:59:25 -0400
Subject: [PATCH] GH-8573: Fix `KafkaMessageSource` samples in docs (#8575)

Fixes https://github.com/spring-projects/spring-integration/issues/8573

* Also add a Kotlin DSL sample

**Cherry-pick to `6.0.x` & `5.5.x`**
---
 src/reference/asciidoc/kafka.adoc | 60 +++++++++++++++++++++++++------
 1 file changed, 49 insertions(+), 11 deletions(-)

diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc
index a806ad89b5..315282db77 100644
--- a/src/reference/asciidoc/kafka.adoc
+++ b/src/reference/asciidoc/kafka.adoc
@@ -373,15 +373,42 @@ The `KafkaMessageSource` provides a pollable channel adapter implementation.
 ==== Java Configuration
 
 ====
-[source, java]
+[source, java, role="primary"]
+.Java DSL
+----
+@Bean
+public IntegrationFlow flow(ConsumerFactory<String, String> cf)  {
+    return IntegrationFlow.from(Kafka.inboundChannelAdapter(cf, new ConsumerProperties("myTopic")),
+                          e -> e.poller(Pollers.fixedDelay(5000)))
+            .handle(System.out::println)
+            .get();
+}
+----
+[source, kotlin, role="secondary"]
+.Kotlin
+----
+@Bean
+fun sourceFlow(cf: ConsumerFactory<String, String>) =
+    integrationFlow(Kafka.inboundChannelAdapter(cf,
+        ConsumerProperties(TEST_TOPIC3).also {
+            it.groupId = "kotlinMessageSourceGroup"
+        }),
+        { poller(Pollers.fixedDelay(100)) }) {
+        handle { m ->
+
+        }
+    }
+----
+[source, java, role="secondary"]
+.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;
+    ConsumerProperties consumerProperties = new ConsumerProperties("myTopic");
+	consumerProperties.setGroupId("myGroupId");
+	consumerProperties.setClientId("myClientId");
+    retunr new KafkaMessageSource<>(cf, consumerProperties);
 }
 ----
 ====
@@ -420,18 +447,29 @@ public IntegrationFlow flow(ConsumerFactory<String, String> cf)  {
 <int-kafka:inbound-channel-adapter
         id="adapter1"
         consumer-factory="consumerFactory"
+        consumer-properties="consumerProperties1"
         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">
+        auto-startup="false">
     <int:poller fixed-delay="5000"/>
 </int-kafka:inbound-channel-adapter>
+
+<bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
+    <constructor-arg>
+        <map>
+            <entry key="max.poll.records" value="1"/>
+        </map>
+    </constructor-arg>
+</bean>
+
+<bean id="consumerProperties1" class="org.springframework.kafka.listener.ConsumerProperties">
+    <constructor-arg name="topics" value="topic1"/>
+    <property name="groupId" value="group"/>
+    <property name="clientId" value="client"/>
+</bean>
 ----
 ====
 
@@ -446,7 +484,7 @@ IMPORTANT: The gateway does not accept requests until the reply container has be
 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.
 
 The `KafkaProducerMessageHandler` `sendTimeoutExpression` default is `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework.
-This has been changed for consistency because you may get unexpected behavior (Spring may timeout the send, while it is actually, eventually, successful).
+This has been changed for consistency because you may get unexpected behavior (Spring may time out the `send()`, while it is actually, eventually, successful).
 IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it to get more timely failures.
 
 ==== Java Configuration