Skip to content

GH-8664: Do not use broken observeWithContext() #8665

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jul 12, 2023
Merged

GH-8664: Do not use broken observeWithContext() #8665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions ...n-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -595,14 +595,16 @@ private Message<?> convertToRequestMessage(Object object, boolean shouldConvert)
private Message<?> sendAndReceiveWithObservation(MessageChannel requestChannel, Object object,
Message<?> requestMessage) {

MessageRequestReplyReceiverContext context =
new MessageRequestReplyReceiverContext(requestMessage, getComponentName());

return IntegrationObservation.GATEWAY.observation(this.observationConvention,
DefaultMessageRequestReplyReceiverObservationConvention.INSTANCE,
() -> new MessageRequestReplyReceiverContext(requestMessage, getComponentName()),
this.observationRegistry)
.<MessageRequestReplyReceiverContext, Message<?>>observeWithContext((ctx) -> {
() -> context, this.observationRegistry)
.observe(() -> {
Message<?> replyMessage = doSendAndReceive(requestChannel, object, requestMessage);
if (replyMessage != null) {
ctx.setResponse(replyMessage);
context.setResponse(replyMessage);
}
return replyMessage;
});
Expand Down
34 changes: 30 additions & 4 deletions ...ework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand All @@ -26,12 +26,14 @@
import io.micrometer.tracing.test.SampleTestRunner;
import io.micrometer.tracing.test.simple.SpansAssert;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.EndpointId;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.interceptor.ObservationPropagationChannelInterceptor;
import org.springframework.integration.config.EnableIntegration;
Expand Down Expand Up @@ -64,20 +66,35 @@ public TracingSetup[] getTracingSetup() {
public SampleTestRunnerConsumer yourCode() {
return (bb, meterRegistry) -> {
ObservationRegistry observationRegistry = getObservationRegistry();

observationRegistry.observationConfig()
.observationPredicate((name, context) ->
!(context instanceof MessageRequestReplyReceiverContext messageRequestReplyReceiverContext)
|| !messageRequestReplyReceiverContext.getGatewayName()
.equals("skippedObservationInboundGateway"));

try (AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext()) {
applicationContext.registerBean(ObservationRegistry.class, () -> observationRegistry);
applicationContext.register(ObservationIntegrationTestConfiguration.class);
applicationContext.refresh();

TestMessagingGatewaySupport messagingGateway =
applicationContext.getBean(TestMessagingGatewaySupport.class);
applicationContext.getBean("testInboundGateway", TestMessagingGatewaySupport.class);

Message<?> receive = messagingGateway.process(new GenericMessage<>("test data"));

assertThat(receive).isNotNull()
.extracting("payload").isEqualTo("test data");
var configuration = applicationContext.getBean(ObservationIntegrationTestConfiguration.class);

messagingGateway =
applicationContext.getBean("skippedObservationInboundGateway",
TestMessagingGatewaySupport.class);

receive = messagingGateway.process(new GenericMessage<>("void data"));

assertThat(receive).isNull();

assertThat(configuration.observedHandlerLatch.await(10, TimeUnit.SECONDS)).isTrue();
}

Expand Down Expand Up @@ -112,7 +129,7 @@ public SampleTestRunnerConsumer yourCode() {
@EnableIntegration
@EnableIntegrationManagement(
observationPatterns = {
"${spring.integration.management.observation-patterns:testInboundGateway,queueChannel,observedEndpoint}",
"${spring.integration.management.observation-patterns:testInboundGateway,skippedObservationInboundGateway,queueChannel,observedEndpoint}",
"${spring.integration.management.observation-patterns:}"
})
public static class ObservationIntegrationTestConfiguration {
Expand All @@ -126,7 +143,7 @@ public ChannelInterceptor observationPropagationInterceptor(ObservationRegistry
}

@Bean
TestMessagingGatewaySupport testInboundGateway(PollableChannel queueChannel) {
TestMessagingGatewaySupport testInboundGateway(@Qualifier("queueChannel") PollableChannel queueChannel) {
TestMessagingGatewaySupport messagingGatewaySupport = new TestMessagingGatewaySupport();
messagingGatewaySupport.setObservationConvention(
new DefaultMessageRequestReplyReceiverObservationConvention() {
Expand All @@ -146,6 +163,15 @@ public PollableChannel queueChannel() {
return new QueueChannel();
}


@Bean
TestMessagingGatewaySupport skippedObservationInboundGateway() {
TestMessagingGatewaySupport messagingGatewaySupport = new TestMessagingGatewaySupport();
messagingGatewaySupport.setRequestChannel(new NullChannel());
messagingGatewaySupport.setReplyTimeout(0);
return messagingGatewaySupport;
}

@Bean
@EndpointId("observedEndpoint")
@ServiceActivator(inputChannel = "queueChannel",
Expand Down