BAEL-7288 Reactive kafka stream

pom.xml

@@ -605,6 +605,7 @@
                 <!--<module>tablesaw</module>--><!-- failing after upgrading to jdk17-->
                 <!--<module>spring-swagger-codegen-modules/openapi-custom-generator</module>
                 <module>spring-swagger-codegen-modules/openapi-custom-generator-api-client</module> --> <!-- cannot add the individual modules here because the order is not guaranteed -->
+                <module>spring-reactive-modules/spring-reactive-kafka-stream-binder</module>
             </modules>
 
             <properties>
@@ -1000,6 +1001,7 @@
                 <!--<module>tablesaw</module>--><!-- failing after upgrading to jdk17-->
                 <!--<module>spring-swagger-codegen-modules/openapi-custom-generator</module>
                 <module>spring-swagger-codegen-modules/openapi-custom-generator-api-client</module> --> <!-- cannot add the individual modules here because the order is not guaranteed -->
+                <module>spring-reactive-modules/spring-reactive-kafka-stream-binder</module>
             </modules>
 
             <properties>

spring-reactive-modules/pom.xml

@@ -31,6 +31,7 @@
         <module>spring-reactive-exceptions</module>
         <module>spring-reactor</module>
         <module>spring-webflux-amqp</module>
+        <module>spring-reactive-kafka-stream-binder</module>
 
         <!--    the following submodules are commented out as a workaround in order to use java 19+ and SB 3.2.x -->
         <!--    <module>spring-reactive-performance</module>-->

spring-reactive-modules/spring-reactive-kafka-stream-binder/pom.xml

@@ -0,0 +1,145 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.baeldung.webflux.kafka.stream</groupId>
+    <artifactId>spring-reactive-kafka-stream-binder</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <packaging>jar</packaging>
+    <name>spring-reactive-kafka-stream-binder</name>
+    <description>WebFlux and Spring Cloud Stream Reactive Kafka Binder</description>
+
+    <parent>
+        <groupId>com.baeldung</groupId>
+        <artifactId>parent-boot-3</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+        <relativePath>../../parent-boot-3</relativePath>
+    </parent>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.springframework.cloud</groupId>
+                <artifactId>spring-cloud-dependencies</artifactId>
+                <version>${spring-cloud.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.testcontainers</groupId>
+                <artifactId>testcontainers-bom</artifactId>
+                <version>${testcontainers.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>${lombok.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- Spring Cloud Stream Reactive Kafka Binder -->
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream-binder-kafka-reactive</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>clickhouse-r2dbc</artifactId>
+            <version>${clickhouse.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>${clickhouse.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>junit</groupId>
+                    <artifactId>junit</artifactId>
+                </exclusion>
+            </exclusions>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>clickhouse</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>${maven-surefire-plugin.version}</version>
+                <configuration>
+                    <source>${java.version}</source>
+                    <target>${java.version}</target>
+                    <encoding>${project.build.sourceEncoding}</encoding>
+                    <testFailureIgnore>true</testFailureIgnore>
+                    <excludes>
+                        <exclude>**/*IntegrationTest.java</exclude>
+                        <exclude>**/*LiveTest.java</exclude>
+                    </excludes>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <properties>
+        <spring-cloud.version>2023.0.4</spring-cloud.version>
+        <start-class>com.baeldung.webflux.kafka.stream.Application</start-class>
+        <java.version>17</java.version>
+        <testcontainers.version>1.20.3</testcontainers.version>
+        <clickhouse.version>0.7.1</clickhouse.version>
+        <maven.compiler.release>17</maven.compiler.release>
+        <maven.compiler.release>17</maven.compiler.release>
+    </properties>
+
+</project>

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/Application.java

@@ -0,0 +1,12 @@
+package com.baeldung.reactive.kafka.stream.binder;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class Application {
+    public static void main(String[] args) {
+        SpringApplication.run(Application.class, args);
+    }
+}
+

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/api/StocksApi.java

@@ -0,0 +1,35 @@
+package com.baeldung.reactive.kafka.stream.binder.api;
+
+import com.baeldung.reactive.kafka.stream.binder.domain.StockUpdate;
+import com.baeldung.reactive.kafka.stream.binder.repository.ClickHouseRepository;
+import jakarta.validation.constraints.NotNull;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ResponseStatusException;
+import reactor.core.publisher.Flux;
+
+import java.time.Instant;
+
+@RestController
+public class StocksApi {
+
+    private final ClickHouseRepository repository;
+
+    @Autowired
+    public StocksApi(ClickHouseRepository repository) {
+        this.repository = repository;
+    }
+
+    @GetMapping("/stock-prices-out")
+    public Flux<StockUpdate> getAvgStockPrices(@RequestParam("from") @NotNull Instant from,  @RequestParam("to") @NotNull Instant to) {
+
+        if (from.isAfter(to)) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "'from' must come before 'to'");
+        }
+
+        return repository.findMinuteAvgStockPrices(from, to);
+    }
+}

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/domain/StockUpdate.java

@@ -0,0 +1,7 @@
+package com.baeldung.reactive.kafka.stream.binder.domain;
+
+import java.time.Instant;
+
+public record StockUpdate(String symbol, double price, String currency, Instant timestamp) {
+
+}

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/domain/currency/CurrencyRate.java

@@ -0,0 +1,7 @@
+package com.baeldung.reactive.kafka.stream.binder.domain.currency;
+
+import reactor.core.publisher.Mono;
+
+public interface CurrencyRate {
+    Mono<Double> convertRate(String from, String to, double amount);
+}

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/kafka/TopicConfig.java

@@ -0,0 +1,38 @@
+package com.baeldung.reactive.kafka.stream.binder.kafka;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.cloud.stream.binder.kafka.BinderHeaderMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.support.KafkaHeaderMapper;
+
+@Configuration
+public class TopicConfig {
+
+    public static final String STOCK_PRICES_OUT = "stock-prices-out";
+    public static final String STOCK_PRICES_IN = "stock-prices-in";
+
+    @Bean(STOCK_PRICES_IN)
+    public NewTopic stockPricesIn() {
+        return TopicBuilder.name(STOCK_PRICES_IN)
+                .partitions(2)
+                .replicas(1)
+                .build();
+    }
+
+    @Bean(STOCK_PRICES_OUT)
+    public NewTopic stockPricesOut() {
+        return TopicBuilder.name(STOCK_PRICES_OUT)
+                .partitions(2)
+                .replicas(1)
+                .build();
+    }
+
+    @Bean
+    public KafkaHeaderMapper kafkaBinderHeaderMapper() {
+        BinderHeaderMapper headerMapper = new BinderHeaderMapper();
+        headerMapper.setMapAllStringsOut(true);
+        return headerMapper;
+    }
+}

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/kafka/consumer/StockPriceConsumer.java

@@ -0,0 +1,66 @@
+package com.baeldung.reactive.kafka.stream.binder.kafka.consumer;
+
+import com.baeldung.reactive.kafka.stream.binder.domain.StockUpdate;
+import com.baeldung.reactive.kafka.stream.binder.kafka.TopicConfig;
+import jakarta.annotation.PostConstruct;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.TopicPartition;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
+import org.springframework.stereotype.Component;
+import reactor.core.scheduler.Schedulers;
+import reactor.kafka.receiver.ReceiverOptions;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
+
+@Component
+@Slf4j
+public class StockPriceConsumer {
+
+    private final ReactiveKafkaConsumerTemplate<String, StockUpdate> kafkaConsumerTemplate;
+    private final AtomicInteger count = new AtomicInteger();
+
+    @SuppressWarnings("all")
+    public StockPriceConsumer(@NonNull KafkaProperties properties, @Qualifier(TopicConfig.STOCK_PRICES_OUT) NewTopic topic) {
+        var receiverOptions = ReceiverOptions.<String, StockUpdate>create(properties.buildConsumerProperties())
+                .subscription(List.of(topic.name()))
+                .assignment(IntStream.range(0, topic.numPartitions()).mapToObj(i -> new TopicPartition(topic.name(), i)).toList())
+                .addAssignListener(partitions -> log.info("************** onPartitionsAssigned: {}", partitions));
+
+        this.kafkaConsumerTemplate = new ReactiveKafkaConsumerTemplate<>(receiverOptions);;
+    }
+
+    @PostConstruct
+    public void consume() {
+      Schedulers.boundedElastic().schedule(() -> kafkaConsumerTemplate
+              .receiveAutoAck()
+              .doOnNext(consumerRecord -> {
+                  // simulate processing
+                  count.incrementAndGet();
+
+                  log.info(
+                          "received key={}, value={} from topic={}, offset={}, partition={}", consumerRecord.key(),
+                          consumerRecord.value(),
+                          consumerRecord.topic(),
+                          consumerRecord.offset(),
+                          consumerRecord.partition());
+              })
+              .doOnError(e -> log.error("Consumer error",  e))
+              .doOnComplete(() -> log.info("Consumed all messages"))
+              .subscribe());
+    }
+
+    public int getCount() {
+        return count.get();
+    }
+
+    public void resetCount() {
+        count.set(0);
+    }
+
+}

spring-reactive-modules/spring-reactive-kafka-stream-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/kafka/processor/StockPriceProcessor.java

@@ -0,0 +1,41 @@
+package com.baeldung.reactive.kafka.stream.binder.kafka.processor;
+
+import com.baeldung.reactive.kafka.stream.binder.domain.StockUpdate;
+import com.baeldung.reactive.kafka.stream.binder.domain.currency.CurrencyRate;
+import com.baeldung.reactive.kafka.stream.binder.repository.ClickHouseRepository;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Function;
+
+@Configuration
+public class StockPriceProcessor {
+
+    private static final String USD = "USD";
+    private static final String EUR = "EUR";
+
+    @Bean
+    public Function<Flux<Message<StockUpdate>>, Flux<Message<StockUpdate>>> processStockPrices(ClickHouseRepository repository, CurrencyRate currencyRate) {
+        return stockPrices -> stockPrices.flatMapSequential(message -> {
+            StockUpdate stockUpdate = message.getPayload();
+            return repository.saveStockPrice(stockUpdate)
+                    .flatMap(success -> Boolean.TRUE.equals(success) ? Mono.just(stockUpdate) : Mono.empty())
+                    .flatMap(stock -> currencyRate.convertRate(USD, EUR, stock.price()))
+                            .map(newPrice -> convertPrice(stockUpdate, newPrice))
+                            .map(priceInEuro -> MessageBuilder.withPayload(priceInEuro)
+                                    .setHeader(KafkaHeaders.KEY, stockUpdate.symbol())
+                                    .copyHeaders(message.getHeaders())
+                                    .build());
+        });
+    }
+
+    private StockUpdate convertPrice(StockUpdate stockUpdate, double newPrice) {
+        return new StockUpdate(stockUpdate.symbol(), newPrice, EUR, stockUpdate.timestamp());
+    }
+
+}