From 8fca3da7e6d7f9e78740c27dc67cf158590acbf3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Rosi=C3=A8re?= Date: Wed, 26 Mar 2025 16:55:23 +0100 Subject: [PATCH] GH-3808: ReplyingKafkaTemplate observation on reply Fixes: #3808 Issue link: https://github.com/spring-projects/spring-kafka/issues/3808 Since `ReplyingKafkaTemplate` is a `BatchMessageListener` for the provided listener container, we cannot rely on the observation from that container. * Implement consumer observation from the `ReplyingKafkaTemplate.BatchMessageListener`. Signed-off-by: Francois Rosiere [artem.bilan@broadcom.com Improve commit message] **Auto-cherry-pick to `3.3.x` & `3.2.x`** Signed-off-by: Artem Bilan --- .../kafka/core/KafkaTemplate.java | 17 ++++- .../requestreply/ReplyingKafkaTemplate.java | 71 +++++++++++-------- .../support/micrometer/ObservationTests.java | 70 +++++++++++++----- 3 files changed, 113 insertions(+), 45 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index cd38a39a..c1e3ddc5 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -104,6 +104,7 @@ import org.springframework.util.StringUtils; * @author Gurps Bassi * @author Valentina Armenise * @author Christian Fredriksson + * @author Francois Rosiere */ public class KafkaTemplate implements KafkaOperations, ApplicationContextAware, BeanNameAware, ApplicationListener, DisposableBean, SmartInitializingSingleton { @@ -465,6 +466,15 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo this.observationRegistry = observationRegistry; } + /** + * Return the {@link ObservationRegistry} used by the template. + * @return the observation registry + * @since 3.2.9 + */ + protected ObservationRegistry getObservationRegistry() { + return this.observationRegistry; + } + /** * Return the {@link KafkaAdmin}, used to find the cluster id for observation, if * present. @@ -533,8 +543,13 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo return removeLeadingAndTrailingBrackets(adminServers == null ? "" : adminServers); } + /** + * Return the cluster id, if available. + * @return the cluster id. + * @since 3.2.9 + */ @Nullable - private String clusterId() { + protected String clusterId() { if (this.kafkaAdmin != null && this.clusterId == null) { this.clusterIdLock.lock(); try { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java index dd272eba..127f0936 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java @@ -29,6 +29,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Function; +import io.micrometer.observation.Observation; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; @@ -52,6 +53,8 @@ import org.springframework.kafka.listener.GenericMessageListenerContainer; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaUtils; import org.springframework.kafka.support.TopicPartitionOffset; +import org.springframework.kafka.support.micrometer.KafkaListenerObservation; +import org.springframework.kafka.support.micrometer.KafkaRecordReceiverContext; import org.springframework.kafka.support.serializer.DeserializationException; import org.springframework.kafka.support.serializer.SerializationUtils; import org.springframework.messaging.Message; @@ -69,6 +72,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan * @author Borahm Lee + * @author Francois Rosiere * * @since 2.1.3 * @@ -502,39 +506,50 @@ public class ReplyingKafkaTemplate extends KafkaTemplate implemen @Override public void onMessage(List> data) { data.forEach(record -> { - Header correlationHeader = record.headers().lastHeader(this.correlationHeaderName); - Object correlationId = null; - if (correlationHeader != null) { - correlationId = this.binaryCorrelation - ? new CorrelationKey(correlationHeader.value()) - : new String(correlationHeader.value(), StandardCharsets.UTF_8); - } - if (correlationId == null) { - this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.format(record) - + " - to use request/reply semantics, the responding server must return the correlation id " - + " in the '" + this.correlationHeaderName + "' header"); + ContainerProperties containerProperties = this.replyContainer.getContainerProperties(); + Observation observation = KafkaListenerObservation.LISTENER_OBSERVATION.observation( + containerProperties.getObservationConvention(), + KafkaListenerObservation.DefaultKafkaListenerObservationConvention.INSTANCE, + () -> new KafkaRecordReceiverContext(record, this.replyContainer.getListenerId(), containerProperties.getClientId(), this.replyContainer.getGroupId(), + this::clusterId), + getObservationRegistry()); + observation.observe(() -> handleReply(record)); + }); + } + + private void handleReply(ConsumerRecord record) { + Header correlationHeader = record.headers().lastHeader(this.correlationHeaderName); + Object correlationId = null; + if (correlationHeader != null) { + correlationId = this.binaryCorrelation + ? new CorrelationKey(correlationHeader.value()) + : new String(correlationHeader.value(), StandardCharsets.UTF_8); + } + if (correlationId == null) { + this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.format(record) + + " - to use request/reply semantics, the responding server must return the correlation id " + + " in the '" + this.correlationHeaderName + "' header"); + } + else { + RequestReplyFuture future = this.futures.remove(correlationId); + Object correlationKey = correlationId; + if (future == null) { + logLateArrival(record, correlationId); } else { - RequestReplyFuture future = this.futures.remove(correlationId); - Object correlationKey = correlationId; - if (future == null) { - logLateArrival(record, correlationId); + boolean ok = true; + Exception exception = checkForErrors(record); + if (exception != null) { + ok = false; + future.completeExceptionally(exception); } - else { - boolean ok = true; - Exception exception = checkForErrors(record); - if (exception != null) { - ok = false; - future.completeExceptionally(exception); - } - if (ok) { - this.logger.debug(() -> "Received: " + KafkaUtils.format(record) - + WITH_CORRELATION_ID + correlationKey); - future.complete(record); - } + if (ok) { + this.logger.debug(() -> "Received: " + KafkaUtils.format(record) + + WITH_CORRELATION_ID + correlationKey); + future.complete(record); } } - }); + } } /** diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java index 10f32d2a..78124e5b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java @@ -17,6 +17,7 @@ package org.springframework.kafka.support.micrometer; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Arrays; import java.util.Deque; import java.util.List; @@ -81,12 +82,14 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.listener.RecordInterceptor; +import org.springframework.kafka.requestreply.ReplyingKafkaTemplate; import org.springframework.kafka.support.ProducerListener; import org.springframework.kafka.support.micrometer.KafkaListenerObservation.DefaultKafkaListenerObservationConvention; import org.springframework.kafka.support.micrometer.KafkaTemplateObservation.DefaultKafkaTemplateObservationConvention; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.util.StringUtils; @@ -102,13 +105,15 @@ import static org.mockito.Mockito.mock; * @author Wang Zhiyang * @author Christian Mergenthaler * @author Soby Chacko + * @author Francois Rosiere * * @since 3.0 */ @SpringJUnitConfig -@EmbeddedKafka(topics = { ObservationTests.OBSERVATION_TEST_1, ObservationTests.OBSERVATION_TEST_2, - ObservationTests.OBSERVATION_TEST_3, ObservationTests.OBSERVATION_RUNTIME_EXCEPTION, - ObservationTests.OBSERVATION_ERROR, ObservationTests.OBSERVATION_TRACEPARENT_DUPLICATE }, partitions = 1) +@EmbeddedKafka(topics = {ObservationTests.OBSERVATION_TEST_1, ObservationTests.OBSERVATION_TEST_2, + ObservationTests.OBSERVATION_TEST_3, ObservationTests.OBSERVATION_TEST_4, ObservationTests.OBSERVATION_REPLY, + ObservationTests.OBSERVATION_RUNTIME_EXCEPTION, ObservationTests.OBSERVATION_ERROR, + ObservationTests.OBSERVATION_TRACEPARENT_DUPLICATE}, partitions = 1) @DirtiesContext public class ObservationTests { @@ -118,6 +123,10 @@ public class ObservationTests { public final static String OBSERVATION_TEST_3 = "observation.testT3"; + public final static String OBSERVATION_TEST_4 = "observation.testT4"; + + public final static String OBSERVATION_REPLY = "observation.reply"; + public final static String OBSERVATION_RUNTIME_EXCEPTION = "observation.runtime-exception"; public final static String OBSERVATION_ERROR = "observation.error.sync"; @@ -135,11 +144,12 @@ public class ObservationTests { @Autowired KafkaListenerEndpointRegistry endpointRegistry, @Autowired KafkaAdmin admin, @Autowired @Qualifier("customTemplate") KafkaTemplate customTemplate, @Autowired Config config) - throws InterruptedException, ExecutionException, TimeoutException { + throws InterruptedException, ExecutionException, TimeoutException { AtomicReference spanFromCallback = new AtomicReference<>(); template.setProducerInterceptor(new ProducerInterceptor<>() { + @Override public ProducerRecord onSend(ProducerRecord record) { tracer.currentSpanCustomizer().tag("key", "value"); @@ -327,10 +337,10 @@ public class ObservationTests { meterRegistryAssert.hasTimerWithNameAndTags("spring.kafka.template", KeyValues.of("spring.kafka.template.name", "template", - "messaging.operation", "publish", - "messaging.system", "kafka", - "messaging.destination.kind", "topic", - "messaging.destination.name", destName) + "messaging.operation", "publish", + "messaging.system", "kafka", + "messaging.destination.kind", "topic", + "messaging.destination.name", destName) .and(keyValues)); } @@ -339,12 +349,12 @@ public class ObservationTests { meterRegistryAssert.hasTimerWithNameAndTags("spring.kafka.listener", KeyValues.of( - "messaging.kafka.consumer.group", consumerGroup, - "messaging.operation", "receive", - "messaging.source.kind", "topic", - "messaging.source.name", destName, - "messaging.system", "kafka", - "spring.kafka.listener.id", listenerId) + "messaging.kafka.consumer.group", consumerGroup, + "messaging.operation", "receive", + "messaging.source.kind", "topic", + "messaging.source.name", destName, + "messaging.system", "kafka", + "spring.kafka.listener.id", listenerId) .and(keyValues)); } @@ -394,7 +404,7 @@ public class ObservationTests { void observationErrorException(@Autowired ExceptionListener listener, @Autowired SimpleTracer tracer, @Autowired @Qualifier("throwableTemplate") KafkaTemplate errorTemplate, @Autowired KafkaListenerEndpointRegistry endpointRegistry) - throws ExecutionException, InterruptedException, TimeoutException { + throws ExecutionException, InterruptedException, TimeoutException { errorTemplate.send(OBSERVATION_ERROR, "testError").get(10, TimeUnit.SECONDS); assertThat(listener.latch5.await(10, TimeUnit.SECONDS)).isTrue(); @@ -485,6 +495,7 @@ public class ObservationTests { @Autowired SimpleTracer tracer) throws Exception { CompletableFuture> producerRecordFuture = new CompletableFuture<>(); template.setProducerListener(new ProducerListener<>() { + @Override public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) { producerRecordFuture.complete(producerRecord); @@ -511,6 +522,18 @@ public class ObservationTests { tracer.getSpans().clear(); } + @Test + void testReplyingKafkaTemplateObservation( + @Autowired ReplyingKafkaTemplate template, + @Autowired ObservationRegistry observationRegistry) { + assertThat(template.sendAndReceive(new ProducerRecord<>(OBSERVATION_TEST_4, "test")) + // the current observation must be retrieved from the consumer thread of the reply + .thenApply(replyRecord -> observationRegistry.getCurrentObservation().getContext())) + .isCompletedWithValueMatchingWithin(observationContext -> + observationContext instanceof KafkaRecordReceiverContext + && "spring.kafka.listener".equals(observationContext.getName()), Duration.ofSeconds(30)); + } + @Configuration @EnableKafka public static class Config { @@ -584,13 +607,22 @@ public class ObservationTests { return template; } + @Bean + ReplyingKafkaTemplate replyingKafkaTemplate(ProducerFactory pf, ConcurrentKafkaListenerContainerFactory containerFactory) { + ReplyingKafkaTemplate kafkaTemplate = new ReplyingKafkaTemplate<>(pf, containerFactory.createContainer(OBSERVATION_REPLY)); + kafkaTemplate.setObservationEnabled(true); + return kafkaTemplate; + } + @Bean ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory( - ConsumerFactory cf, ObservationRegistry observationRegistry) { + ConsumerFactory cf, ObservationRegistry observationRegistry, + KafkaTemplate kafkaTemplate) { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(cf); + factory.setReplyTemplate(kafkaTemplate); factory.getContainerProperties().setObservationEnabled(true); factory.setContainerCustomizer(container -> { if (container.getListenerId().equals("obs3")) { @@ -721,6 +753,12 @@ public class ObservationTests { void listen3(ConsumerRecord in) { } + @KafkaListener(id = "obsReply", topics = OBSERVATION_TEST_4) + @SendTo // default REPLY_TOPIC header + public String replyListener(ConsumerRecord in) { + return in.value().toUpperCase(); + } + } public static class ExceptionListener {