Add reactive application with database and Kafka

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

@@ -1,6 +1,10 @@
+plugins {
+    id("org.gradle.toolchains.foojay-resolver-convention") version "0.8.0"
+}
 rootProject.name = "spring-boot-open-telemetry-demo"
 
 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-reactive/build.gradle.kts

@@ -0,0 +1,83 @@
+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")) {
+        exclude(group = "io.gitlab.arturbosch.detekt")
+    }
+    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:3.8.0-M3")
+}
+tasks {
+    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,19 @@
+/*
+ * 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;
+
+@SpringBootApplication
+public class Application {
+
+    public static void main(String[] args) {
+        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,37 @@
+/*
+ * 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.exporter.otlp.trace.OtlpGrpcSpanExporter;
+import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder;
+import org.springframework.boot.actuate.autoconfigure.tracing.otlp.OtlpTracingAutoConfiguration;
+import org.springframework.boot.actuate.autoconfigure.tracing.otlp.OtlpTracingProperties;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Locale;
+import javax.annotation.Nonnull;
+
+@AutoConfigureBefore(OtlpTracingAutoConfiguration.class)
+@Configuration(proxyBeanMethods = false)
+class OpenTelemetryConfig {
+
+    @Bean
+    @ConditionalOnMissingBean(OtlpGrpcSpanExporter.class)
+    OtlpGrpcSpanExporter otelJaegerGrpcSpanExporter(@Nonnull final OtlpTracingProperties otlpProperties) {
+        final OtlpGrpcSpanExporterBuilder builder = OtlpGrpcSpanExporter.builder()
+            .setEndpoint(otlpProperties.getEndpoint())
+            .setTimeout(otlpProperties.getTimeout())
+            .setConnectTimeout(otlpProperties.getConnectTimeout())
+            .setCompression(String.valueOf(otlpProperties.getCompression()).toLowerCase(Locale.ROOT));
+        otlpProperties.getHeaders().forEach(builder::addHeader);
+        return builder.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();
+    }
+}

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

@@ -0,0 +1,27 @@
+/*
+ * 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.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@Configuration(proxyBeanMethods = false)
+public class WebClientConfig {
+
+    @Value("${app.external-base-url}")
+    private String external;
+
+    @Bean
+    public WebClient webClient(WebClient.Builder builder) {
+        return builder
+            .baseUrl(external)
+            .build();
+    }
+}

spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/controllers/HomeController.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.controllers;
+
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+@RestController
+public class HomeController {
+
+    @GetMapping("/")
+    public Mono<String> home() {
+        return Mono.just("Hello!");
+    }
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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.controllers;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+@RestController
+public class RedirectController {
+
+    // http://localhost:8080/redirect
+    @GetMapping(path = "/redirect")
+    public Mono<ResponseEntity<Object>> redirectToGoogle() throws URISyntaxException {
+        final URI google = new URI("https://www.google.com");
+        final HttpHeaders httpHeaders = new HttpHeaders();
+        httpHeaders.setLocation(google);
+        return Mono.just(new ResponseEntity<>(httpHeaders, HttpStatus.SEE_OTHER));
+    }
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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.controllers;
+
+import io.github.mfvanek.spring.boot3.reactive.service.KafkaSendingService;
+import io.github.mfvanek.spring.boot3.reactive.service.PublicApiService;
+import io.micrometer.tracing.Span;
+import io.micrometer.tracing.TraceContext;
+import io.micrometer.tracing.Tracer;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+
+import java.time.Clock;
+import java.time.LocalDateTime;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+@Slf4j
+@RestController
+@RequiredArgsConstructor
+public class TimeController {
+
+    private final Tracer tracer;
+    private final Clock clock;
+    private final KafkaSendingService kafkaSendingService;
+    private final PublicApiService publicApiService;
+
+    // http://localhost:8080/current-time
+    @GetMapping(path = "/current-time")
+    public Mono<LocalDateTime> getNow() {
+        log.trace("tracer {}", tracer);
+        final String traceId = Optional.ofNullable(tracer.currentSpan())
+            .map(Span::context)
+            .map(TraceContext::traceId)
+            .orElse(null);
+        log.info("Called method getNow. TraceId = {}", traceId);
+        return publicApiService.getZonedTime()
+            .switchIfEmpty(Mono.just(LocalDateTime.now(clock)))
+            .doOnSuccess(this::sendWithKafka);
+    }
+
+    private void sendWithKafka(LocalDateTime localDateTime) {
+        try {
+            kafkaSendingService.sendNotification("Current time = " + localDateTime)
+                .thenRun(() -> log.info("Awaiting acknowledgement from Kafka"))
+                .get();
+        } catch (InterruptedException | ExecutionException e) {
+            log.info("error ", e);
+        }
+    }
+}