GH-233: Add batch consumer for Kinesis sample

Fixes https://github.com/spring-cloud/spring-cloud-stream-samples/issues/233

kinesis-samples/kinesis-produce-consume/src/main/java/demo/security/SecurityConfiguration.java

@@ -0,0 +1,15 @@
+package demo.security;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.web.SecurityFilterChain;
+
+@Configuration
+public class SecurityConfiguration {
+
+	@Bean
+	public SecurityFilterChain filterChain(HttpSecurity httpSecurity) throws Exception {
+		return httpSecurity.csrf().disable().build();
+	}
+}

kinesis-samples/kinesis-produce-consume/src/main/java/demo/stream/OrderStreamConfiguration.java

@@ -16,6 +16,7 @@
 
 package demo.stream;
 
+import java.util.List;
 import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
@@ -27,6 +28,7 @@ import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.context.annotation.Bean;
 
 import demo.repository.OrderRepository;
+import org.springframework.stereotype.Component;
 
 /**
  *
@@ -34,6 +36,7 @@ import demo.repository.OrderRepository;
  * @author Artem Bilan
  *
  */
+@Component
 public class OrderStreamConfiguration {
 
 	private final Log logger = LogFactory.getLog(getClass());
@@ -52,4 +55,20 @@ public class OrderStreamConfiguration {
 		};
 	}
 
+	@Bean
+	public Consumer<List<Event>> processOrders(OrderRepository orders) {
+		return eventList -> {
+			//log the number of orders received and each order
+			logger.info("Received " + eventList.size() + " orders");
+			for(Event event: eventList) {
+				if (!event.getOriginator().equals("KinesisProducer")) {
+					logger.info("An order has been received " + event.toString());
+				}
+				else {
+					logger.info("An order has been placed from this service " + event.toString());
+				}
+			}
+		};
+	}
+
 }

kinesis-samples/kinesis-produce-consume/src/main/resources/application.yml

@@ -20,11 +20,23 @@ spring:
           content-type: application/json
           producer:
             partitionKeyExpression: "1"
+            headerMode: none
         processOrder-in-0:
           destination: test_stream
           content-type: application/json
+        processOrders-in-0:
+          destination: test_stream
+          content-type: application/json
+          consumer:
+            headerMode: none
+      kinesis:
+        bindings:
+          processOrders-in-0:
+            consumer:
+              listenerMode: batch
+              recordsLimit: 5
     function:
-      definition: processOrder;produceOrder
+      definition: processOrder;produceOrder;processOrders
   datasource:
     url: jdbc:h2:mem:testdb
     driverClassName: org.h2.Driver

kotlin-rabbit-functions/pom.xml

@@ -39,7 +39,7 @@
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-function-context</artifactId>
-			<version>3.2.3</version>
+			<version>3.2.2-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>