Fix reactive chain

Author: mfvanek

.gitignore

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

.kotlin/errors/errors-1753873052588.log

@@ -36,19 +36,18 @@ public class TimeController {
     @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);
-        final Mono<LocalDateTime> response = publicApiService.getZonedTime()
-            .switchIfEmpty(Mono.just(LocalDateTime.now(clock)));
-        response
-            .subscribe(it -> {
-                assert traceId != null;
-                kafkaSendingService.sendNotification("Current time = " + it, traceId);
-                log.info("Awaiting acknowledgement from Kafka");
-            });
-        return response;
+
+        return publicApiService.getZonedTime()
+            .defaultIfEmpty(LocalDateTime.now(clock))
+            .flatMap(now -> kafkaSendingService.sendNotification("Current time = " + now)
+                .doOnSuccess(v -> log.info("Awaiting acknowledgement from Kafka"))
+                .thenReturn(now)
+            );
     }
 }

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

@@ -9,17 +9,13 @@
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.internals.RecordHeader;
-import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.slf4j.MDC;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
-import java.nio.charset.StandardCharsets;
 import java.util.UUID;
 import javax.annotation.Nonnull;
 
@@ -31,37 +27,13 @@ public class KafkaSendingService {
     private final KafkaTemplate<UUID, String> kafkaTemplate;
     @Value("${app.tenant.name}")
     private String tenantName;
-    //private final Propagator propagator;
-    //private final Tracer tracer;
 
-    public void sendNotification(@Nonnull final String message, @Nonnull final String traceId) {
-        Mono.fromFuture(() -> {
+    public Mono<SendResult<UUID, String>> sendNotification(@Nonnull final String message) {
+        return Mono.deferContextual(contextView -> {
             try (MDC.MDCCloseable ignored = MDC.putCloseable("tenant.name", tenantName)) {
                 log.info("Sending message \"{}\" to Kafka", message);
-                // это тоже удаляет
-                //propagator.inject(
-                //    Objects.requireNonNull(tracer.currentSpan()).context(),
-                //    kafkaTemplate.getProducerFactory().getConfigurationProperties(),
-                //    (stringObjectMap, key, value) -> stringObjectMap.put("traceparent", traceId)
-                //);
-                // это удаляет все
-                // kafkaTemplate.getProducerFactory().getConfigurationProperties().remove("traceparent");
-                //  kafkaTemplate.getProducerFactory().getConfigurationProperties().remove("b3");
-                //   kafkaTemplate.getProducerFactory().getConfigurationProperties().putIfAbsent("traceparent", traceId);
-                //   kafkaTemplate.getProducerFactory().getConfigurationProperties().putIfAbsent("b3", traceId);
-                // так можно добавить хедер с другим названием, но существующие не заменяются
-                final ProducerRecord<UUID, String> producerRecord = new ProducerRecord<>(
-                    kafkaTemplate.getDefaultTopic(),
-                    kafkaTemplate.partitionsFor(kafkaTemplate.getDefaultTopic()).get(0).partition(),
-                    UUID.randomUUID(),
-                    message,
-                    new RecordHeaders(new Header[]{
-                        new RecordHeader("traceparent", traceId.getBytes(StandardCharsets.UTF_8)),
-                        new RecordHeader("b3", traceId.getBytes(StandardCharsets.UTF_8))})
-                );
-                return kafkaTemplate.send(producerRecord);
-                //return kafkaTemplate.sendDefault(UUID.randomUUID(), message);
+                return Mono.fromFuture(() -> kafkaTemplate.sendDefault(UUID.randomUUID(), message));
             }
-        }).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

@@ -19,6 +19,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.dao.QueryTimeoutException;
 
 import java.util.Locale;
 
@@ -65,7 +66,7 @@ void jdbcQueryTimeoutFromProperties() {
     @DisplayName("Throws exception when query exceeds timeout")
     void exceptionWithLongQuery() {
         assertThatThrownBy(() -> jdbcTemplate.execute("select pg_sleep(1.1);"))
-            .isInstanceOf(DataAccessResourceFailureException.class)
+            .isInstanceOf(QueryTimeoutException.class)
             .hasMessageContaining("ERROR: canceling statement due to user request");
     }