3 versions to change headers but not working as expected

m.zharinova · m.zharinova · commit bacffa9a0744 · 2025-08-10T15:47:28.000+05:00
diff --git a/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/config/OpenTelemetryConfig.java b/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/config/OpenTelemetryConfig.java
@@ -9,7 +9,7 @@
 
 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.OtlpHttpSpanExporterBuilderCustomizer;
 import org.springframework.boot.actuate.autoconfigure.tracing.otlp.OtlpTracingAutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.context.annotation.Bean;
@@ -20,7 +20,7 @@
 class OpenTelemetryConfig {
 
     @Bean
-    OtlpGrpcSpanExporterBuilderCustomizer otelJaegerGrpcSpanExporterBuilderCustomizer(
+    OtlpHttpSpanExporterBuilderCustomizer otelJaegerHttpSpanExporterBuilderCustomizer(
         @Value("${management.otlp.tracing.retry.max-attempts:2}") int maxAttempts
     ) {
         return builder -> builder.setRetryPolicy(
diff --git a/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/controllers/TimeController.java b/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/controllers/TimeController.java
@@ -41,15 +41,14 @@ public Mono<LocalDateTime> getNow() {
             .map(TraceContext::traceId)
             .orElse(null);
         log.info("Called method getNow. TraceId = {}", traceId);
-        return publicApiService.getZonedTime()
-            .switchIfEmpty(Mono.just(LocalDateTime.now(clock)))
-            .doOnNext(this::sendWithKafka);
-    }
-
-    private void sendWithKafka(LocalDateTime localDateTime) {
-        kafkaSendingService.sendNotification("Current time = " + localDateTime)
-            .doOnNext(result -> log.info("Awaiting acknowledgement from Kafka"))
-            .doOnError(e -> log.info("error ", e))
-            .subscribe();
+        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;
     }
 }
diff --git a/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/service/KafkaReadingService.java b/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/service/KafkaReadingService.java
@@ -52,4 +52,3 @@ public void listen(ConsumerRecord<UUID, String> message, Acknowledgment ack) {
         }
     }
 }
-
diff --git a/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/service/KafkaSendingService.java b/spring-boot-3-demo-app-reactive/src/main/java/io/github/mfvanek/spring/boot3/reactive/service/KafkaSendingService.java
@@ -9,13 +9,17 @@
 
 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;
 
@@ -27,13 +31,37 @@ 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 Mono<SendResult<UUID, String>> sendNotification(@Nonnull final String message) {
-        return Mono.fromCallable(() -> {
+    public void sendNotification(@Nonnull final String message, @Nonnull final String traceId) {
+        Mono.fromFuture(() -> {
             try (MDC.MDCCloseable ignored = MDC.putCloseable("tenant.name", tenantName)) {
                 log.info("Sending message \"{}\" to Kafka", message);
-                return kafkaTemplate.sendDefault(UUID.randomUUID(), message).get();
+                // это тоже удаляет
+                //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);
             }
-        });
+        }).doOnError(e -> log.info("error ", e)).subscribe();
     }
 }
diff --git a/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/ApplicationTests.java b/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/ApplicationTests.java
@@ -13,7 +13,7 @@
 import io.micrometer.tracing.Span;
 import io.micrometer.tracing.Tracer;
 import io.micrometer.tracing.otel.bridge.OtelTracer;
-import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
+import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -43,18 +43,16 @@ void contextLoads() {
             .isInstanceOf(OtelTracer.class)
             .satisfies(t -> assertThat(t.currentSpan())
                 .isNotEqualTo(Span.NOOP));
-        assertThat(applicationContext.getBean("otelJaegerGrpcSpanExporter"))
+        assertThat(applicationContext.getBean("otlpHttpSpanExporter"))
             .isNotNull()
-            .isInstanceOf(OtlpGrpcSpanExporter.class)
+            .isInstanceOf(OtlpHttpSpanExporter.class)
             .satisfies(e -> assertThat(e.toString())
                 .contains(String.format(Locale.ROOT, """
-                    OtlpGrpcSpanExporter{exporterName=otlp, type=span, endpoint=http://localhost:%d, \
-                    endpointPath=/opentelemetry.proto.collector.trace.v1.TraceService/Export, \
-                    timeoutNanos=5000000000, connectTimeoutNanos=10000000000, compressorEncoding=null, \
-                    headers=Headers{User-Agent=OBFUSCATED}, \
-                    retryPolicy=RetryPolicy{maxAttempts=5, initialBackoff=PT1S, maxBackoff=PT5S, backoffMultiplier=1.5, \
-                    retryExceptionPredicate=null},""", JaegerInitializer.getFirstMappedPort()))
-            );
+                        OtlpHttpSpanExporter{exporterName=otlp, type=span, endpoint=http://localhost:%d, \
+                        timeoutNanos=5000000000, proxyOptions=null, compressorEncoding=null, connectTimeoutNanos=10000000000, \
+                        exportAsJson=false, headers=Headers{User-Agent=OBFUSCATED}, retryPolicy=RetryPolicy{maxAttempts=2, \
+                        initialBackoff=PT1S, maxBackoff=PT5S, backoffMultiplier=1.5, retryExceptionPredicate=null},""",
+                    JaegerInitializer.getFirstMappedPort())));
     }
 
     @Test
diff --git a/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/controllers/HomeControllerTest.java b/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/controllers/HomeControllerTest.java
@@ -27,5 +27,4 @@ void controllerResponseShouldHaveTraceIdHeader() {
         assertThat(result)
             .isEqualTo("Hello!");
     }
-
 }
diff --git a/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/controllers/TimeControllerTest.java b/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/controllers/TimeControllerTest.java
@@ -30,6 +30,7 @@
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
@@ -42,6 +43,7 @@
 import javax.annotation.Nonnull;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.within;
 
 @ExtendWith(OutputCaptureExtension.class)
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
@@ -85,11 +87,10 @@ void spanShouldBeReportedInLogs(@Nonnull final CapturedOutput output) throws Int
             .returnResult();
         final String traceId = result.getResponseHeaders().getFirst(TRACE_ID_HEADER_NAME);
         assertThat(traceId).isNotBlank();
-        assertThat(result.getResponseBody())
-            .isBefore(LocalDateTime.now(clock));
+        assertThat(result.getResponseBody()).isCloseTo(LocalDateTime.now(clock).minusDays(1), within(1, ChronoUnit.MINUTES));
         assertThat(output.getAll())
-            .contains("Called method getNow. TraceId = " + traceId)
-            .contains("Awaiting acknowledgement from Kafka");
+            .contains("Called method getNow. TraceId = " + traceId);
+        //.contains("Awaiting acknowledgement from Kafka");
 
         final ConsumerRecord<UUID, String> received = consumerRecords.poll(10, TimeUnit.SECONDS);
         assertThat(received).isNotNull();
@@ -123,7 +124,7 @@ void spanAndMdcShouldBeReportedWhenRetry(@Nonnull final CapturedOutput output) {
         final String traceId = result.getResponseHeaders().getFirst(TRACE_ID_HEADER_NAME);
         assertThat(traceId)
             .isEqualTo("38c19768104ab8ae64fabbeed65bbbdf");
-
+        assertThat(result.getResponseBody()).isCloseTo(LocalDateTime.now(clock), within(1, ChronoUnit.MINUTES));
         assertThat(output.getAll())
             .containsPattern(String.format(Locale.ROOT,
                 ".*\"message\":\"Retrying request to '/%1$s', attempt 1/1 due to error:\"," +
diff --git a/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/support/TestBase.java b/spring-boot-3-demo-app-reactive/src/test/java/io/github/mfvanek/spring/boot3/reactive/support/TestBase.java
@@ -105,5 +105,4 @@ private void stubOkButNotCorrectResponse(@Nonnull final String zoneName) {
                 .withBody(objectMapper.writeValueAsString("bad response"))
             ));
     }
-
 }

Original file line number	Diff line number	Diff line change
`@@ -52,4 +52,3 @@ public void listen(ConsumerRecord<UUID, String> message, Acknowledgment ack) {`
`52`	`52`	`}`
`53`	`53`	`}`
`54`	`54`	`}`
`55`		`-`
Original file line number	Diff line number	Diff line change
`@@ -27,5 +27,4 @@ void controllerResponseShouldHaveTraceIdHeader() {`
`27`	`27`	`assertThat(result)`
`28`	`28`	`.isEqualTo("Hello!");`
`29`	`29`	`}`
`30`		`-`
`31`	`30`	`}`
Original file line number	Diff line number	Diff line change
`@@ -105,5 +105,4 @@ private void stubOkButNotCorrectResponse(@Nonnull final String zoneName) {`
`105`	`105`	`.withBody(objectMapper.writeValueAsString("bad response"))`
`106`	`106`	`));`
`107`	`107`	`}`
`108`		`-`
`109`	`108`	`}`