Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3808: ReplyingKafkaTemplate observation on reply

Browse Source 
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 <francois.rosiere@gmail.com>

[artem.bilan@broadcom.com Improve commit message]

**Auto-cherry-pick to `3.3.x` & `3.2.x`**

Signed-off-by: Artem Bilan <artem.bilan@broadcom.com>
This commit is contained in:
François Rosière
2025-03-26 16:55:23 +01:00
committed by GitHub
parent f92f766146
commit 8fca3da7e6
3 changed files with 113 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
View File

@@ -104,6 +104,7 @@ import org.springframework.util.StringUtils;
* @author Gurps Bassi
* @author Valentina Armenise
* @author Christian Fredriksson
* @author Francois Rosiere
*/
public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationContextAware, BeanNameAware,
ApplicationListener<ContextStoppedEvent>, DisposableBean, SmartInitializingSingleton {
@@ -465,6 +466,15 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, 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<K, V> implements KafkaOperations<K, V>, 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 {

71
spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java
View File

@@ -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<K, V, R> extends KafkaTemplate<K, V> implemen
@Override
public void onMessage(List<ConsumerRecord<K, R>> 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<K, R> 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<K, V, R> future = this.futures.remove(correlationId);
Object correlationKey = correlationId;
if (future == null) {
logLateArrival(record, correlationId);
}
else {
RequestReplyFuture<K, V, R> 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);
}
}
});
}
}
/**

70
spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java
View File

@@ -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<Integer, String> customTemplate,
@Autowired Config config)
throws InterruptedException, ExecutionException, TimeoutException {
throws InterruptedException, ExecutionException, TimeoutException {
AtomicReference<SimpleSpan> spanFromCallback = new AtomicReference<>();
template.setProducerInterceptor(new ProducerInterceptor<>() {
@Override
public ProducerRecord<Integer, String> onSend(ProducerRecord<Integer, String> 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<Integer, String> 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<ProducerRecord<Integer, String>> producerRecordFuture = new CompletableFuture<>();
template.setProducerListener(new ProducerListener<>() {
@Override
public void onSuccess(ProducerRecord<Integer, String> producerRecord, RecordMetadata recordMetadata) {
producerRecordFuture.complete(producerRecord);
@@ -511,6 +522,18 @@ public class ObservationTests {
tracer.getSpans().clear();
}
@Test
void testReplyingKafkaTemplateObservation(
@Autowired ReplyingKafkaTemplate<Integer, String, String> 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<Integer, String, String> replyingKafkaTemplate(ProducerFactory<Integer, String> pf, ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
ReplyingKafkaTemplate<Integer, String, String> kafkaTemplate = new ReplyingKafkaTemplate<>(pf, containerFactory.createContainer(OBSERVATION_REPLY));
kafkaTemplate.setObservationEnabled(true);
return kafkaTemplate;
}
@Bean
ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory(
ConsumerFactory<Integer, String> cf, ObservationRegistry observationRegistry) {
ConsumerFactory<Integer, String> cf, ObservationRegistry observationRegistry,
KafkaTemplate<Integer, String> kafkaTemplate) {
ConcurrentKafkaListenerContainerFactory<Integer, String> 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<Integer, String> in) {
}
@KafkaListener(id = "obsReply", topics = OBSERVATION_TEST_4)
@SendTo // default REPLY_TOPIC header
public String replyListener(ConsumerRecord<Integer, String> in) {
return in.value().toUpperCase();
}
}
public static class ExceptionListener {