feat : process events parallelly

aws-kinesis-project/consumer/README.md

@@ -13,6 +13,7 @@ $ ./mvnw spring-boot:run -Dspring-boot.run.profiles=local
 
  - Starting 4.0.0 kinesis steam binder works with aws v2 of dynamodb, cloudwatch and kinesis
  - The maximum size of a data blob (the data payload before Base64-encoding) is 1 megabyte (MB).
+ - Set DEFAULT_BOUNDED_ELASTIC_ON_VIRTUAL_THREADS=true for using virtual threads
 
 
 ### Useful Links

aws-kinesis-project/consumer/src/main/java/com/learning/aws/spring/consumer/IpConsumer.java

@@ -7,14 +7,15 @@
 import com.learning.aws.spring.model.IpAddressDTO;
 import com.learning.aws.spring.repository.IpAddressEventRepository;
 import java.time.Duration;
-import java.time.LocalDateTime;
 import java.util.List;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import software.amazon.awssdk.services.kinesis.model.Record;
 
 @Configuration(proxyBeanMethods = false)
@@ -38,53 +39,44 @@ public IpConsumer(
     Consumer<Flux<List<Record>>> consumeEvent() {
         return recordFlux ->
                 recordFlux
-                        .flatMap(Flux::fromIterable)
-                        .map(
-                                kinessRecord -> {
+                        .flatMapIterable(list -> list)
+                        .flatMap(
+                                kinesisRecord -> {
                                     log.info(
                                             "Sequence Number :{}, partitionKey :{} and expected ArrivalTime :{}",
-                                            kinessRecord.sequenceNumber(),
-                                            kinessRecord.partitionKey(),
-                                            kinessRecord.approximateArrivalTimestamp());
+                                            kinesisRecord.sequenceNumber(),
+                                            kinesisRecord.partitionKey(),
+                                            kinesisRecord.approximateArrivalTimestamp());
 
                                     String dataAsString =
-                                            new String(kinessRecord.data().asByteArray());
+                                            new String(kinesisRecord.data().asByteArray());
                                     String payload =
                                             dataAsString.substring(dataAsString.indexOf("[{"));
-                                    List<IpAddressDTO> ipAddressDTOS;
+
                                     try {
-                                        ipAddressDTOS =
+                                        List<IpAddressDTO> ipAddressDTOS =
                                                 objectMapper.readValue(
                                                         payload, new TypeReference<>() {});
+                                        return Flux.fromIterable(ipAddressDTOS);
                                     } catch (JsonProcessingException e) {
-                                        throw new RuntimeException(e);
+                                        return Flux.error(e);
                                     }
-                                    return Flux.fromIterable(ipAddressDTOS);
                                 })
-                        .doOnNext(
-                                ipAddressDTOsList -> {
-                                    log.info(
-                                            "IpAddress processed at {} and value is:{}",
-                                            LocalDateTime.now(),
-                                            ipAddressDTOsList);
-                                    this.processEvents(ipAddressDTOsList);
+                        .parallel() // Parallelize processing
+                        .runOn(Schedulers.boundedElastic()) // Run processing on boundedElastic
+                        // Scheduler
+                        .flatMap(
+                                ipAddressDTO -> {
+                                    IpAddressEvent ipAddressEvent =
+                                            new IpAddressEvent(
+                                                    ipAddressDTO.ipAddress(),
+                                                    ipAddressDTO.eventProducedTime());
+                                    return Mono.just(ipAddressEvent)
+                                            .delayElement(
+                                                    Duration.ofSeconds(
+                                                            1)) // Adds artificial latency
+                                            .flatMap(ipAddressEventRepository::save);
                                 })
-                        .subscribe();
-    }
-
-    private void processEvents(Flux<IpAddressDTO> ipAddressDTOFlux) {
-        ipAddressDTOFlux
-                .map(
-                        ipAddressDTO ->
-                                new IpAddressEvent(
-                                        ipAddressDTO.ipAddress(), ipAddressDTO.eventProducedTime()))
-                .delayElements(Duration.ofSeconds(1)) // Adds artificial latency
-                .subscribe(
-                        ipAddressEvent ->
-                                ipAddressEventRepository
-                                        .save(ipAddressEvent)
-                                        .subscribe(
-                                                savedEvent ->
-                                                        log.info("Saved Event :{}", savedEvent)));
+                        .subscribe(savedEvent -> log.info("Saved Event :{}", savedEvent));
     }
 }

aws-kinesis-project/consumer/src/main/resources/application.properties

@@ -1,14 +1,8 @@
 spring.application.name=aws-kinesis-consumer-project
 server.port=8080
 server.shutdown=graceful
-spring.main.allow-bean-definition-overriding=true
 spring.jmx.enabled=false
-
-################ Logging #####################
-logging.file.name=logs/aws-kinesis-consumer-project.log
-logging.level.web=INFO
-logging.level.com.amazonaws.util.EC2MetadataUtils=ERROR
-logging.level.com.amazonaws.internal.InstanceMetadataServiceResourceFetcher=ERROR
+spring.threads.virtual.enabled=true
 
 ################ Actuator #####################
 management.endpoints.web.exposure.include=configprops,env,health,info,logfile,loggers,metrics
@@ -24,17 +18,21 @@ spring.cloud.stream.bindings.consumeEvent-in-0.group=my-test-stream-Consumer-Gro
 spring.cloud.stream.bindings.consumeEvent-in-0.content-type=application/json
 spring.cloud.stream.bindings.consumeEvent-in-0.consumer.header-mode=none
 spring.cloud.stream.bindings.consumeEvent-in-0.consumer.use-native-decoding=true
-#defaults to 1, this will process upto 5 messages concurrently
+#defaults to 1, this will process upto 5 messages concurrently, in reactive mode this is not necessary
 #spring.cloud.stream.bindings.consumeEvent-in-0.consumer.concurrency=5
 spring.cloud.stream.kinesis.bindings.consumeEvent-in-0.consumer.listenerMode=batch
+
 spring.cloud.function.definition=consumeEvent;
 
 #Kinesis-dynamodb-checkpoint
 spring.cloud.stream.kinesis.binder.checkpoint.table=spring-stream-metadata
 spring.cloud.stream.kinesis.binder.checkpoint.billingMode=provisioned
 spring.cloud.stream.kinesis.binder.checkpoint.readCapacity=5
 spring.cloud.stream.kinesis.binder.checkpoint.writeCapacity=5
+
 spring.cloud.stream.kinesis.binder.locks.table=spring-stream-lock-registry
 spring.cloud.stream.kinesis.binder.locks.billingMode=provisioned
 spring.cloud.stream.kinesis.binder.locks.readCapacity=5
 spring.cloud.stream.kinesis.binder.locks.writeCapacity=5
+
+spring.reactor.schedulers.defaultBoundedElasticOnVirtualThreads=true

aws-kinesis-project/consumer/src/main/resources/logback-spring.xml

@@ -1,14 +1,20 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <configuration>
+    <!-- Include default Spring Boot Logback configurations -->
     <include resource="org/springframework/boot/logging/logback/defaults.xml" />
+    <!-- Include default CONSOLE and FILE appenders -->
     <include resource="org/springframework/boot/logging/logback/console-appender.xml" />
 
+    <springProperty scope="context" name="appName" source="spring.application.name"/>
+
     <springProfile name="default">
         <root level="INFO">
             <appender-ref ref="CONSOLE" />
         </root>
     </springProfile>
     <springProfile name="!default">
+        <!-- Define the log file location -->
+        <property name="LOG_FILE" value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}}/logs/${appName}.log}"/>
         <include resource="org/springframework/boot/logging/logback/file-appender.xml" />
         <root level="INFO">
             <appender-ref ref="FILE" />
@@ -17,7 +23,6 @@
     </springProfile>
 
     <logger name="com.learning.aws.spring" level="DEBUG"/>
-    <logger name="com.amazonaws.util.EC2MetadataUtils" level="ERROR"/>
-    <logger name="com.amazonaws.internal.InstanceMetadataServiceResourceFetcher" level="ERROR"/>
+    <logger name="org.springframework" level="INFO"/>
 
 </configuration>