GH-8573: Fix KafkaMessageSource samples in docs (#8575)

Fixes #8573

* Also add a Kotlin DSL sample

**Cherry-pick to `6.0.x` & `5.5.x`**

Author: artembilan

src/reference/asciidoc/kafka.adoc

@@ -360,22 +360,37 @@ The `KafkaMessageSource` provides a pollable channel adapter implementation.
 ----
 @Bean
 public IntegrationFlow flow(ConsumerFactory<String, String> cf)  {
-    return IntegrationFlow.from(Kafka.inboundChannelAdapter(cf, "myTopic")
-                .groupId("myDslGroupId"), e -> e.poller(Pollers.fixedDelay(5000)))
+    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);
 }
 ----
 [source, xml, role="secondary"]
@@ -384,18 +399,29 @@ public KafkaMessageSource<String, String> source(ConsumerFactory<String, String>
 <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>
 ----
 ====
 
@@ -421,7 +447,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.
 
 [[kafka-outbound-gateway-configuration]]