From 547ca663c581c22c5fe212aba560552d3cada061 Mon Sep 17 00:00:00 2001
From: Steven PG <steven.p.gantz@gmail.com>
Date: Thu, 9 Feb 2023 17:58:41 -0500
Subject: [PATCH] Add Spring Batch Producer example to existing Kafka batch
 sample (#240)

* Add working batch-produce example in same style as existing sample app

* add documentation and alignment with existing project

* capitalization
---
 kafka-batch-sample/README.adoc                  |  3 ++-
 .../demo/KafkaBatchSampleApplication.java       | 17 +++++++++++++++++
 .../src/main/resources/application.yml          | 15 +++++++++++++++
 3 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/kafka-batch-sample/README.adoc b/kafka-batch-sample/README.adoc
index f76abf3..ba30b63 100644
--- a/kafka-batch-sample/README.adoc
+++ b/kafka-batch-sample/README.adoc
@@ -2,7 +2,8 @@
 
 This is an example of a Spring Cloud Stream demonstrating the processing of record batches.
 
-The application simply upper-cases the input records and sends them to another topic.
+The application simply upper-cases the input records and sends them to another topic using two different
+implementations. The `batch-produce` profile enables batch publishing.
 
 === Running the app:
 
diff --git a/kafka-batch-sample/src/main/java/com/example/demo/KafkaBatchSampleApplication.java b/kafka-batch-sample/src/main/java/com/example/demo/KafkaBatchSampleApplication.java
index 247dbf4..490299b 100644
--- a/kafka-batch-sample/src/main/java/com/example/demo/KafkaBatchSampleApplication.java
+++ b/kafka-batch-sample/src/main/java/com/example/demo/KafkaBatchSampleApplication.java
@@ -20,6 +20,8 @@ import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.List;
 import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -31,6 +33,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Profile;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
 
 @SpringBootApplication
@@ -89,3 +93,16 @@ class Transactions extends Base {
 	}
 
 }
+
+@Component
+@Profile("batch-produce")
+class BatchProduce extends Base {
+	@Bean
+	Function<List<String>, List<Message<String>>> consumer() {
+		return list -> list.stream()
+			.map(string -> string.toUpperCase())
+			.map(uppercasedString -> MessageBuilder.withPayload(uppercasedString).build())
+			.collect(Collectors.toList());
+	}
+
+}
diff --git a/kafka-batch-sample/src/main/resources/application.yml b/kafka-batch-sample/src/main/resources/application.yml
index 97d42e9..8696e7b 100644
--- a/kafka-batch-sample/src/main/resources/application.yml
+++ b/kafka-batch-sample/src/main/resources/application.yml
@@ -28,3 +28,18 @@ spring:
         binder:
           transaction:
             transaction-id-prefix: batch-tx-
+
+---
+
+spring:
+  profiles: batch-produce
+  cloud:
+    stream:
+      bindings:
+        consumer-in-0:
+          destination: batch-in
+          group: batch-in
+          consumer:
+            batch-mode: true
+        consumer-out-0:
+          destination: batch-out