done with Mono.fromCallable()

Author: m.zharinova

.kotlin/errors/errors-1753873052588.log

@@ -0,0 +1,4 @@
+kotlin version: 2.0.21
+error message: The daemon has terminated unexpectedly on startup attempt #1 with error code: 0. The daemon process output:
+    1. Kotlin compile daemon is ready
+

settings.gradle.kts

@@ -1,6 +1,3 @@
-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")

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

@@ -27,9 +27,7 @@ dependencies {
     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(project(":db-migrations"))
     implementation("org.liquibase:liquibase-core")
     implementation("com.github.blagerweij:liquibase-sessionlock")
     implementation(libs.datasource.micrometer)
@@ -43,8 +41,9 @@ dependencies {
     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")
+    testImplementation("io.projectreactor:reactor-test")
 }
+
 tasks {
     jacocoTestCoverageVerification {
         dependsOn(jacocoTestReport)

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

@@ -7,31 +7,26 @@
 
 package io.github.mfvanek.spring.boot3.reactive.config;
 
-import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
-import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder;
+import io.opentelemetry.sdk.common.export.RetryPolicy;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.actuate.autoconfigure.tracing.otlp.OtlpGrpcSpanExporterBuilderCustomizer;
 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();
+    OtlpGrpcSpanExporterBuilderCustomizer otelJaegerGrpcSpanExporterBuilderCustomizer(
+        @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/controllers/TimeController.java

@@ -21,7 +21,6 @@
 import java.time.Clock;
 import java.time.LocalDateTime;
 import java.util.Optional;
-import java.util.concurrent.ExecutionException;
 
 @Slf4j
 @RestController
@@ -44,16 +43,13 @@ public Mono<LocalDateTime> getNow() {
         log.info("Called method getNow. TraceId = {}", traceId);
         return publicApiService.getZonedTime()
             .switchIfEmpty(Mono.just(LocalDateTime.now(clock)))
-            .doOnSuccess(this::sendWithKafka);
+            .doOnNext(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);
-        }
+        kafkaSendingService.sendNotification("Current time = " + localDateTime)
+            .doOnNext(result -> log.info("Awaiting acknowledgement from Kafka"))
+            .doOnError(e -> log.info("error ", e))
+            .subscribe();
     }
 }

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

@@ -14,9 +14,9 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
 
 import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
 import javax.annotation.Nonnull;
 
 @Slf4j
@@ -28,10 +28,12 @@ public class KafkaSendingService {
     @Value("${app.tenant.name}")
     private String tenantName;
 
-    public CompletableFuture<SendResult<UUID, String>> sendNotification(@Nonnull final String message) {
-        try (MDC.MDCCloseable ignored = MDC.putCloseable("tenant.name", tenantName)) {
-            log.info("Sending message \"{}\" to Kafka", message);
-            return kafkaTemplate.sendDefault(UUID.randomUUID(), message);
-        }
+    public Mono<SendResult<UUID, String>> sendNotification(@Nonnull final String message) {
+        return Mono.fromCallable(() -> {
+            try (MDC.MDCCloseable ignored = MDC.putCloseable("tenant.name", tenantName)) {
+                log.info("Sending message \"{}\" to Kafka", message);
+                return kafkaTemplate.sendDefault(UUID.randomUUID(), message).get();
+            }
+        });
     }
 }

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

@@ -7,10 +7,10 @@
 
 package io.github.mfvanek.spring.boot3.reactive.service;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.github.mfvanek.spring.boot3.reactive.service.dto.CurrentTime;
 import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Value;
@@ -53,14 +53,9 @@ private Mono<CurrentTime> getZonedTimeFromWorldTimeApi() {
             .retrieve()
             .bodyToMono(String.class)
             .retryWhen(prepareRetry(zoneName))
-            .flatMap(string -> {
-                try {
-                    return Mono.just(mapper.readValue(string, CurrentTime.class));
-                } catch (JsonProcessingException e) {
-                    log.info("error from webclient ", e);
-                }
-                return Mono.empty();
-            });
+            .flatMap(this::convert)
+            .onErrorComplete()
+            .flatMap(Mono::justOrEmpty);
     }
 
     private Retry prepareRetry(final String zoneName) {
@@ -77,4 +72,8 @@ private Retry prepareRetry(final String zoneName) {
             });
     }
 
+    @SneakyThrows
+    private Mono<? extends CurrentTime> convert(String string) {
+        return Mono.just(mapper.readValue(string, CurrentTime.class));
+    }
 }

spring-boot-3-demo-app-reactive/src/main/resources/application.yml

@@ -4,15 +4,8 @@ app:
     tenant.name: ru-a1-private
 
 server:
-    port: 8080
+    port: 8081
     # See also https://docs.spring.io/spring-boot/docs/2.7.9/reference/html/application-properties.html#appendix.application-properties.server
-    tomcat:
-        accept-count: 10
-        max-connections: 400
-        threads:
-            max: 10
-            min-spare: 5 # actuator port uses the same configuration
-    shutdown: graceful
 
 demo:
     kafka:
@@ -57,7 +50,7 @@ spring:
 
 management:
     server:
-        port: 8085
+        port: 8087
     endpoints:
         web:
             exposure.include: '*'

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

@@ -36,8 +36,8 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
 
@@ -47,7 +47,7 @@
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class TimeControllerTest extends TestBase {
 
-    private final BlockingQueue<ConsumerRecord<UUID, String>> consumerRecords = new LinkedBlockingQueue<>();
+    private final BlockingQueue<ConsumerRecord<UUID, String>> consumerRecords = new ArrayBlockingQueue<>(4);
     private KafkaMessageListenerContainer<UUID, String> container;
     @Autowired
     private KafkaProperties kafkaProperties;