Add reactive application with database and Kafka (#290)

* add tests

* add autowired filter

* fix tests and flows

* fix mono code

* lower branch coverage for reactive app to 0.5

* fix after merge master

* delete some log details

* done with Mono.fromCallable()

* 3 versions to change headers but not working as expected

* Fix reactive chain

* Fix tests

* Add BlockHound

* Polish

---------

Co-authored-by: m.zharinova <[email protected]>
Co-authored-by: Ivan Vakhrushev <[email protected]>

Author: 3 people

.gitignore

@@ -8,3 +8,5 @@ build/
 *.iml
 *.ipr
 out/
+
+.kotlin

buildSrc/src/main/kotlin/sb-ot-demo.jacoco-rules.gradle.kts

@@ -34,13 +34,6 @@ tasks {
                     minimum = "0.93".toBigDecimal()
                 }
             }
-            rule {
-                limit {
-                    counter = "BRANCH"
-                    value = "COVEREDRATIO"
-                    minimum = "0.66".toBigDecimal()
-                }
-            }
         }
     }
 }

settings.gradle.kts

@@ -4,3 +4,4 @@ include("spring-boot-3-demo-app")
 include("common-internal-bom")
 include("db-migrations")
 include("spring-boot-3-demo-app-kotlin")
+include("spring-boot-3-demo-app-reactive")

spring-boot-3-demo-app-kotlin/src/test/kotlin/io/github/mfvanek/spring/boot3/kotlin/test/IndexesMaintenanceTest.kt renamed to spring-boot-3-demo-app-kotlin/src/test/kotlin/io/github/mfvanek/spring/boot3/kotlin/test/DatabaseStructureStaticAnalysisTest.kt

@@ -11,7 +11,8 @@ import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 import org.springframework.beans.factory.annotation.Autowired
 
-class IndexesMaintenanceTest : TestBase() {
+internal class DatabaseStructureStaticAnalysisTest : TestBase() {
+
     @Autowired
     private lateinit var checks: List<DatabaseCheckOnHost<out DbObject>>
 

spring-boot-3-demo-app-kotlin/src/test/kotlin/io/github/mfvanek/spring/boot3/kotlin/test/controllers/TimeControllerTest.kt

@@ -4,15 +4,6 @@ import io.github.mfvanek.spring.boot3.kotlin.test.filters.TraceIdInResponseServl
 import io.github.mfvanek.spring.boot3.kotlin.test.service.dto.toParsedDateTime
 import io.github.mfvanek.spring.boot3.kotlin.test.support.KafkaInitializer
 import io.github.mfvanek.spring.boot3.kotlin.test.support.TestBase
-import java.nio.charset.StandardCharsets
-import java.time.Duration
-import java.time.LocalDateTime
-import java.time.temporal.ChronoUnit
-import java.util.Locale
-import java.util.UUID
-import java.util.concurrent.BlockingQueue
-import java.util.concurrent.LinkedBlockingQueue
-import java.util.concurrent.TimeUnit
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.ConsumerRecord
@@ -38,12 +29,20 @@ import org.springframework.kafka.listener.MessageListener
 import org.springframework.kafka.test.utils.ContainerTestUtils
 import org.springframework.kafka.test.utils.KafkaTestUtils
 import org.testcontainers.shaded.org.awaitility.Awaitility
+import java.nio.charset.StandardCharsets
+import java.time.Duration
+import java.time.LocalDateTime
+import java.time.temporal.ChronoUnit
+import java.util.*
+import java.util.concurrent.ArrayBlockingQueue
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.TimeUnit
 
 @ExtendWith(OutputCaptureExtension::class)
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class TimeControllerTest : TestBase() {
     private lateinit var container: KafkaMessageListenerContainer<UUID, String>
-    private val consumerRecords = LinkedBlockingQueue<ConsumerRecord<UUID, String>>()
+    private val consumerRecords = ArrayBlockingQueue<ConsumerRecord<UUID, String>>(4)
 
     @Autowired
     private lateinit var kafkaProperties: KafkaProperties

spring-boot-3-demo-app-reactive/build.gradle.kts

@@ -0,0 +1,87 @@
+plugins {
+    id("sb-ot-demo.java-conventions")
+    id("sb-ot-demo.forbidden-apis")
+    id("sb-ot-demo.docker")
+    alias(libs.plugins.spring.boot.v3)
+    id("io.freefair.lombok")
+}
+
+dependencies {
+    implementation(platform(project(":common-internal-bom")))
+    implementation(platform(libs.springdoc.openapi))
+    implementation(platform(libs.spring.boot.v3.dependencies))
+    implementation(platform(libs.spring.cloud))
+
+    implementation("org.springframework.boot:spring-boot-starter-webflux")
+    implementation("org.springframework.boot:spring-boot-starter-actuator")
+    implementation("org.springframework.boot:spring-boot-starter-security")
+    implementation("io.micrometer:micrometer-registry-prometheus")
+    implementation("io.projectreactor:reactor-tools")
+    implementation("org.springdoc:springdoc-openapi-starter-webflux-ui")
+
+    implementation("org.springframework.kafka:spring-kafka")
+
+    implementation("io.micrometer:micrometer-tracing-bridge-otel")
+    implementation("io.opentelemetry:opentelemetry-exporter-otlp")
+
+    implementation("org.springframework.boot:spring-boot-starter-jdbc")
+    implementation("org.postgresql:postgresql")
+    implementation("com.zaxxer:HikariCP")
+    implementation(project(":db-migrations"))
+    implementation("org.liquibase:liquibase-core")
+    implementation("com.github.blagerweij:liquibase-sessionlock")
+    implementation(libs.datasource.micrometer)
+    implementation(libs.logstash)
+
+    testImplementation("org.springframework.boot:spring-boot-starter-test")
+    testImplementation("org.springframework.boot:spring-boot-starter-webflux")
+    testImplementation("org.testcontainers:postgresql")
+    testImplementation("org.testcontainers:kafka")
+    testImplementation("org.springframework.kafka:spring-kafka-test")
+    testImplementation("org.awaitility:awaitility")
+    testImplementation("io.github.mfvanek:pg-index-health-test-starter")
+    testImplementation("org.springframework.cloud:spring-cloud-starter-contract-stub-runner")
+    testImplementation("io.projectreactor:reactor-test")
+    testImplementation("io.projectreactor.tools:blockhound:1.0.13.RELEASE")
+}
+
+tasks {
+    withType<Test>().all {
+        jvmArgs("-XX:+AllowRedefinitionToAddDeleteMethods")
+    }
+
+    jacocoTestCoverageVerification {
+        dependsOn(jacocoTestReport)
+        violationRules {
+            rule {
+                limit {
+                    counter = "BRANCH"
+                    value = "COVEREDRATIO"
+                    minimum = "0.50".toBigDecimal()
+                }
+            }
+        }
+    }
+}
+
+val coverageExcludeList = listOf("**/*Application.class")
+listOf(JacocoCoverageVerification::class, JacocoReport::class).forEach { taskType ->
+    tasks.withType(taskType) {
+        afterEvaluate {
+            classDirectories.setFrom(
+                files(
+                    classDirectories.files.map { file ->
+                        fileTree(file).apply {
+                            exclude(coverageExcludeList)
+                        }
+                    }
+                )
+            )
+        }
+    }
+}
+
+
+springBoot {
+    buildInfo()
+}

spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/Application.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright (c) 2020-2025. Ivan Vakhrushev and others.
+ * https://github.com/mfvanek/spring-boot-open-telemetry-demo
+ *
+ * Licensed under the Apache License 2.0
+ */
+
+package io.github.mfvanek.spring.boot3.reactive;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import reactor.tools.agent.ReactorDebugAgent;
+
+@SpringBootApplication
+public class Application {
+
+    public static void main(String[] args) {
+        ReactorDebugAgent.init();
+        SpringApplication.run(Application.class, args);
+    }
+}

spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/config/ClockConfig.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2020-2025. Ivan Vakhrushev and others.
+ * https://github.com/mfvanek/spring-boot-open-telemetry-demo
+ *
+ * Licensed under the Apache License 2.0
+ */
+
+package io.github.mfvanek.spring.boot3.reactive.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.time.Clock;
+
+@Configuration(proxyBeanMethods = false)
+public class ClockConfig {
+
+    @Bean
+    public Clock clock() {
+        return Clock.systemUTC();
+    }
+}

spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/config/OpenTelemetryConfig.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2020-2025. Ivan Vakhrushev and others.
+ * https://github.com/mfvanek/spring-boot-open-telemetry-demo
+ *
+ * Licensed under the Apache License 2.0
+ */
+
+package io.github.mfvanek.spring.boot3.reactive.config;
+
+import io.opentelemetry.sdk.common.export.RetryPolicy;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.actuate.autoconfigure.tracing.otlp.OtlpHttpSpanExporterBuilderCustomizer;
+import org.springframework.boot.actuate.autoconfigure.tracing.otlp.OtlpTracingAutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@AutoConfigureBefore(OtlpTracingAutoConfiguration.class)
+@Configuration(proxyBeanMethods = false)
+class OpenTelemetryConfig {
+
+    @Bean
+    OtlpHttpSpanExporterBuilderCustomizer otelJaegerHttpSpanExporterBuilderCustomizer(
+        @Value("${management.otlp.tracing.retry.max-attempts:2}") int maxAttempts
+    ) {
+        return builder -> builder.setRetryPolicy(
+            RetryPolicy.builder()
+                .setMaxAttempts(maxAttempts)
+                .build()
+        );
+    }
+}

spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/config/SecurityConfig.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2020-2025. Ivan Vakhrushev and others.
+ * https://github.com/mfvanek/spring-boot-open-telemetry-demo
+ *
+ * Licensed under the Apache License 2.0
+ */
+
+package io.github.mfvanek.spring.boot3.reactive.config;
+
+import lombok.SneakyThrows;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
+import org.springframework.security.config.web.server.ServerHttpSecurity;
+import org.springframework.security.web.server.SecurityWebFilterChain;
+
+@Configuration(proxyBeanMethods = false)
+@EnableWebFluxSecurity
+public class SecurityConfig {
+
+    @Bean
+    @SneakyThrows
+    public SecurityWebFilterChain securityFilterChain(ServerHttpSecurity http) {
+        http
+            .authorizeExchange(exchanges -> exchanges.anyExchange().permitAll());
+        return http.build();
+    }
+}