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 890f79a5..5e99e1a9 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 @@ -103,6 +103,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Gurps Bassi * @author Valentina Armenise + * @author Francois Rosiere */ public class KafkaTemplate implements KafkaOperations, ApplicationContextAware, BeanNameAware, ApplicationListener, DisposableBean, SmartInitializingSingleton { @@ -456,6 +457,15 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo this.observationConvention = observationConvention; } + /** + * 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. @@ -520,8 +530,13 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo return removeLeadingAndTrailingBrackets(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 3ae1ef42..131721f5 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; @@ -51,6 +52,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.lang.Nullable; @@ -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 * @@ -501,39 +505,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 24c2a62c..46c458d4 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; @@ -76,13 +77,14 @@ import org.springframework.kafka.core.KafkaAdmin; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.MessageListenerContainer; +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.lang.Nullable; +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; @@ -98,13 +100,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) + 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 { @@ -114,6 +118,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"; @@ -356,7 +364,7 @@ public class ObservationTests { void observationRuntimeException(@Autowired ExceptionListener listener, @Autowired SimpleTracer tracer, @Autowired @Qualifier("throwableTemplate") KafkaTemplate runtimeExceptionTemplate, @Autowired KafkaListenerEndpointRegistry endpointRegistry) - throws ExecutionException, InterruptedException, TimeoutException { + throws ExecutionException, InterruptedException, TimeoutException { runtimeExceptionTemplate.send(OBSERVATION_RUNTIME_EXCEPTION, "testRuntimeException").get(10, TimeUnit.SECONDS); assertThat(listener.latch4.await(10, TimeUnit.SECONDS)).isTrue(); @@ -459,6 +467,19 @@ 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())) + .succeedsWithin(Duration.ofSeconds(30)) + .isInstanceOf(KafkaRecordReceiverContext.class) + .extracting("name") + .isEqualTo("spring.kafka.listener"); + } + @Configuration @EnableKafka public static class Config { @@ -530,13 +551,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) { + 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")) { @@ -585,7 +615,7 @@ public class ObservationTests { // This is called on the producer side when the message is being sent // Normally we would pass information from tracing context - for tests we don't need to @Override - public void inject(TraceContext context, @Nullable C carrier, Setter setter) { + public void inject(TraceContext context, C carrier, Setter setter) { setter.set(carrier, "foo", "some foo value"); setter.set(carrier, "bar", "some bar value"); @@ -649,6 +679,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 {