From 2bfcb32628ab51279d90d52bec530b5cea2c7b62 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 16 Aug 2022 10:31:14 -0400 Subject: [PATCH] Initial support for Micrometer Observation (#3845) * Initial support for Micrometer Observation * Add respective Observation dependencies * Refactor an `AbstractMessageHandler` logic for potential Observation hooks * Introduce an `ObservationPropagationChannelInterceptor` to propagate an `Observation` from one thread to another through message channels * Adds an example of propagation * Fixed the user code and receiving spans * * Clean up for Tracing unit test * Make `micrometer-observation` as an `api` dep - non-optional for direct API usage Co-authored-by: Marcin Grzejszczak --- build.gradle | 3 + ...ervationPropagationChannelInterceptor.java | 77 ++++ ...eadStatePropagationChannelInterceptor.java | 18 +- .../handler/AbstractMessageHandler.java | 39 +- .../management/observation/package-info.java | 7 + ...ionPropagationChannelInterceptorTests.java | 335 ++++++++++++++++++ .../transformer/TransformerContextTests.java | 24 +- 7 files changed, 465 insertions(+), 38 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/package-info.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java diff --git a/build.gradle b/build.gradle index 0176146e54..811755f142 100644 --- a/build.gradle +++ b/build.gradle @@ -491,6 +491,7 @@ project('spring-integration-core') { exclude group: 'org.springframework' } api 'io.projectreactor:reactor-core' + api 'io.micrometer:micrometer-observation' optionalApi 'com.fasterxml.jackson.core:jackson-databind' optionalApi 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8' @@ -512,6 +513,8 @@ project('spring-integration-core') { testImplementation "org.aspectj:aspectjweaver:$aspectjVersion" testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion" + testImplementation 'io.micrometer:micrometer-observation-test' + testImplementation 'io.micrometer:micrometer-tracing-test' } dokkaHtmlPartial { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java new file mode 100644 index 0000000000..f12ca5cdd3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptor.java @@ -0,0 +1,77 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel.interceptor; + +import org.springframework.aop.support.AopUtils; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; + +import io.micrometer.common.lang.Nullable; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; + +/** + * The {@link org.springframework.messaging.support.ExecutorChannelInterceptor} + * implementation responsible for an {@link Observation} propagation from one message + * flow's thread to another through the {@link MessageChannel}s involved in the flow. + * Opens a new {@link Observation.Scope} on another thread and cleans up it in the end. + * + * @author Artem Bilan + * + * @since 6.0 + */ +public class ObservationPropagationChannelInterceptor extends ThreadStatePropagationChannelInterceptor { + + private final ThreadLocal scopes = new ThreadLocal<>(); + + private final ObservationRegistry observationRegistry; + + public ObservationPropagationChannelInterceptor(ObservationRegistry observationRegistry) { + Assert.notNull(observationRegistry, "'observationRegistry' must noty be null"); + this.observationRegistry = observationRegistry; + } + + @Override + @Nullable + protected Observation obtainPropagatingContext(Message message, MessageChannel channel) { + if (!DirectChannel.class.isAssignableFrom(AopUtils.getTargetClass(channel))) { + return this.observationRegistry.getCurrentObservation(); + } + return null; + } + + @Override + protected void populatePropagatedContext(@Nullable Observation state, Message message, MessageChannel channel) { + if (state != null) { + Observation.Scope scope = state.openScope(); + this.scopes.set(scope); + } + } + + @Override + public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, Exception ex) { + Observation.Scope scope = this.scopes.get(); + if (scope != null && scope == this.observationRegistry.getCurrentObservationScope()) { + scope.close(); + this.scopes.remove(); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java index a4f1ff8795..80993872a9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java @@ -23,6 +23,8 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.ExecutorChannelInterceptor; +import io.micrometer.common.lang.Nullable; + /** * The {@link ExecutorChannelInterceptor} implementation responsible for * the {@link Thread} (any?) state propagation from one message flow's thread to another @@ -47,16 +49,16 @@ import org.springframework.messaging.support.ExecutorChannelInterceptor; * * @author Artem Bilan * @author Gary Russell + * * @since 4.2 */ -public abstract class ThreadStatePropagationChannelInterceptor - implements ExecutorChannelInterceptor { +public abstract class ThreadStatePropagationChannelInterceptor implements ExecutorChannelInterceptor { @Override public final Message preSend(Message message, MessageChannel channel) { S threadContext = obtainPropagatingContext(message, channel); if (threadContext != null) { - return new MessageWithThreadState(message, threadContext); + return new MessageWithThreadState<>(message, threadContext); } else { return message; @@ -80,15 +82,10 @@ public abstract class ThreadStatePropagationChannelInterceptor return postReceive(message, channel); } - @Override - public void afterMessageHandled(Message message, MessageChannel channel, MessageHandler handler, - Exception ex) { - // No-op - } - + @Nullable protected abstract S obtainPropagatingContext(Message message, MessageChannel channel); - protected abstract void populatePropagatedContext(S state, Message message, MessageChannel channel); + protected abstract void populatePropagatedContext(@Nullable S state, Message message, MessageChannel channel); private static final class MessageWithThreadState implements Message, MessageDecorator { @@ -129,4 +126,3 @@ public abstract class ThreadStatePropagationChannelInterceptor } } - diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 074385b876..b23a46a90d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 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. @@ -39,31 +39,42 @@ public abstract class AbstractMessageHandler extends MessageHandlerSupport @Override // NOSONAR public void handleMessage(Message message) { - Message messageToUse = message; - Assert.notNull(messageToUse, "Message must not be null"); + Assert.notNull(message, "Message must not be null"); if (isLoggingEnabled() && this.logger.isDebugEnabled()) { - this.logger.debug(this + " received message: " + messageToUse); + this.logger.debug(this + " received message: " + message); } - SampleFacade sample = null; MetricsCaptor metricsCaptor = getMetricsCaptor(); if (metricsCaptor != null) { - sample = metricsCaptor.start(); + handleWithMetrics(message, metricsCaptor); } + else { + doHandleMessage(message); + } + } + + private void handleWithMetrics(Message message, MetricsCaptor metricsCaptor) { + SampleFacade sample = metricsCaptor.start(); + try { + doHandleMessage(message); + sample.stop(sendTimer()); + } + catch (Exception ex) { + sample.stop(buildSendTimer(false, ex.getClass().getSimpleName())); + throw ex; + } + } + + private void doHandleMessage(Message message) { + Message messageToUse = message; try { if (shouldTrack()) { messageToUse = MessageHistory.write(messageToUse, this, getMessageBuilderFactory()); } handleMessageInternal(messageToUse); - if (sample != null) { - sample.stop(sendTimer()); - } } - catch (Exception e) { - if (sample != null) { - sample.stop(buildSendTimer(false, e.getClass().getSimpleName())); - } + catch (Exception ex) { throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(messageToUse, - () -> "error occurred in message handler [" + this + "]", e); + () -> "error occurred in message handler [" + this + "]", ex); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/package-info.java new file mode 100644 index 0000000000..e4db774be9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/observation/package-info.java @@ -0,0 +1,7 @@ +/** + * Provides classes to support of Micrometer Observation API. + */ + +@org.springframework.lang.NonNullApi +@org.springframework.lang.NonNullFields +package org.springframework.integration.support.management.observation; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java new file mode 100644 index 0000000000..44de3fcaf7 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ObservationPropagationChannelInterceptorTests.java @@ -0,0 +1,335 @@ +/* + * Copyright 2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel.interceptor; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.BridgeTo; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.ExecutorChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationHandler; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.observation.tck.TestObservationRegistry; +import io.micrometer.observation.tck.TestObservationRegistryAssert; +import io.micrometer.observation.transport.ReceiverContext; +import io.micrometer.observation.transport.SenderContext; +import io.micrometer.tracing.Span; +import io.micrometer.tracing.TraceContext; +import io.micrometer.tracing.Tracer; +import io.micrometer.tracing.exporter.FinishedSpan; +import io.micrometer.tracing.handler.DefaultTracingObservationHandler; +import io.micrometer.tracing.handler.PropagatingReceiverTracingObservationHandler; +import io.micrometer.tracing.handler.PropagatingSenderTracingObservationHandler; +import io.micrometer.tracing.propagation.Propagator; +import io.micrometer.tracing.test.simple.SimpleTracer; +import io.micrometer.tracing.test.simple.SpansAssert; +import io.micrometer.tracing.test.simple.TracerAssert; + +/** + * @author Artem Bilan + * + * @since 6.0 + */ +@SpringJUnitConfig +public class ObservationPropagationChannelInterceptorTests { + + @Autowired + ObservationRegistry observationRegistry; + + @Autowired + SimpleTracer simpleTracer; + + @Autowired + SubscribableChannel directChannel; + + @Autowired + SubscribableChannel executorChannel; + + @Autowired + PollableChannel queueChannel; + + @Autowired + DirectChannel testConsumer; + + @Autowired + ExecutorChannel testTracingChannel; + + @BeforeEach + void setup() { + this.simpleTracer.getSpans().clear(); + } + + @Test + void observationPropagatedOverDirectChannel() throws InterruptedException { + AtomicReference scopeReference = new AtomicReference<>(); + CountDownLatch handleLatch = new CountDownLatch(1); + this.directChannel.subscribe(m -> { + scopeReference.set(this.observationRegistry.getCurrentObservationScope()); + handleLatch.countDown(); + }); + + AtomicReference originalScope = new AtomicReference<>(); + + Observation.createNotStarted("test1", this.observationRegistry) + .observe(() -> { + originalScope.set(this.observationRegistry.getCurrentObservationScope()); + this.directChannel.send(new GenericMessage<>("test")); + }); + + assertThat(handleLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(scopeReference.get()) + .isNotNull() + .isSameAs(originalScope.get()); + + TestObservationRegistryAssert.assertThat(this.observationRegistry) + .doesNotHaveAnyRemainingCurrentObservation(); + + TracerAssert.assertThat(this.simpleTracer) + .onlySpan() + .hasNameEqualTo("test1"); + } + + @Test + void observationPropagatedOverExecutorChannel() throws InterruptedException { + AtomicReference scopeReference = new AtomicReference<>(); + CountDownLatch handleLatch = new CountDownLatch(1); + this.executorChannel.subscribe(m -> { + scopeReference.set(this.observationRegistry.getCurrentObservationScope()); + handleLatch.countDown(); + }); + + AtomicReference originalScope = new AtomicReference<>(); + + Observation.createNotStarted("test2", this.observationRegistry) + .observe(() -> { + originalScope.set(this.observationRegistry.getCurrentObservationScope()); + this.executorChannel.send(new GenericMessage<>("test")); + }); + + assertThat(handleLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(scopeReference.get()) + .isNotNull() + .isNotSameAs(originalScope.get()); + + assertThat(scopeReference.get().getCurrentObservation()) + .isSameAs(originalScope.get().getCurrentObservation()); + + TestObservationRegistryAssert.assertThat(this.observationRegistry) + .doesNotHaveAnyRemainingCurrentObservation(); + + TracerAssert.assertThat(this.simpleTracer) + .onlySpan() + .hasNameEqualTo("test2"); + } + + @Test + void observationPropagatedOverQueueChannel() throws InterruptedException { + AtomicReference scopeReference = new AtomicReference<>(); + CountDownLatch handleLatch = new CountDownLatch(1); + this.testConsumer.subscribe(m -> { + scopeReference.set(this.observationRegistry.getCurrentObservationScope()); + handleLatch.countDown(); + }); + + AtomicReference originalScope = new AtomicReference<>(); + + Observation.createNotStarted("test3", this.observationRegistry) + .observe(() -> { + originalScope.set(this.observationRegistry.getCurrentObservationScope()); + this.queueChannel.send(new GenericMessage<>("test")); + }); + + assertThat(handleLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(scopeReference.get()) + .isNotNull() + .isNotSameAs(originalScope.get()); + + assertThat(scopeReference.get().getCurrentObservation()) + .isSameAs(originalScope.get().getCurrentObservation()); + + TestObservationRegistryAssert.assertThat(this.observationRegistry) + .doesNotHaveAnyRemainingCurrentObservation(); + + TracerAssert.assertThat(this.simpleTracer) + .onlySpan() + .hasNameEqualTo("test3"); + } + + @Test + @SuppressWarnings({ "unchecked", "rawtypes" }) + void observationContextPropagatedOverDirectChannel() throws InterruptedException { + CountDownLatch handleLatch = new CountDownLatch(1); + this.testTracingChannel.subscribe(m -> { + // This would be the instrumentation code on the receiver side + // We would need to check if Zipkin wouldn't require us to create the receiving span and then an additional one for the user code... + ReceiverContext> receiverContext = + new ReceiverContext<>((carrier, key) -> carrier.getHeaders().get(key, String.class)); + receiverContext.setCarrier(m); + + // ...and this would be the user's code + Observation.createNotStarted("user.code", receiverContext, this.observationRegistry) + .observe(() -> { + // Let's assume that this is the user code + handleLatch.countDown(); + }); + }); + + // This would be the instrumentation code on the sender side (user's code would call e.g. MessageTemplate and this code + // would lay in MessageTemplate) + // We need to mutate the carrier, so we need to use the builder not the message since message headers are immutable + SenderContext> senderContext = + new SenderContext<>((carrier, key, value) -> Objects.requireNonNull(carrier).setHeader(key, value)); + MessageBuilder builder = MessageBuilder.withPayload("test"); + senderContext.setCarrier(builder); + + Observation.createNotStarted("sending", senderContext, this.observationRegistry) + .observe(() -> this.testTracingChannel.send(builder.build())); + + assertThat(handleLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + TestObservationRegistryAssert.assertThat(this.observationRegistry) + .doesNotHaveAnyRemainingCurrentObservation(); + + TracerAssert.assertThat(this.simpleTracer) + .reportedSpans() + .hasSize(2) + .satisfies(simpleSpans -> SpansAssert.assertThat((Collection) (Collection) simpleSpans) + .hasASpanWithName("sending") + .assertThatASpanWithNameEqualTo("user.code") + .hasTag("foo", "some foo value") + .hasTag("bar", "some bar value") + .hasKindEqualTo(Span.Kind.CONSUMER)); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + SimpleTracer simpleTracer() { + return new SimpleTracer(); + } + + @Bean + ObservationRegistry observationRegistry(Tracer tracer, Propagator propagator) { + TestObservationRegistry observationRegistry = TestObservationRegistry.create(); + observationRegistry.observationConfig().observationHandler( + // Composite will pick the first matching handler + new ObservationHandler.FirstMatchingCompositeObservationHandler( + // This is responsible for creating a child span on the sender side + new PropagatingSenderTracingObservationHandler<>(tracer, propagator), + // This is responsible for creating a span on the receiver side + new PropagatingReceiverTracingObservationHandler<>(tracer, propagator), + // This is responsible for creating a default span + new DefaultTracingObservationHandler(tracer))); + return observationRegistry; + } + + @Bean + @GlobalChannelInterceptor(patterns = "*Channel") + public ChannelInterceptor observationPropagationInterceptor(ObservationRegistry observationRegistry) { + return new ObservationPropagationChannelInterceptor(observationRegistry); + } + + @Bean + @BridgeTo(value = "testConsumer", poller = @Poller(fixedDelay = "100")) + public PollableChannel queueChannel() { + return new QueueChannel(); + } + + @Bean + public SubscribableChannel executorChannel() { + return new ExecutorChannel(Executors.newSingleThreadExecutor()); + } + + @Bean + public SubscribableChannel directChannel() { + return new DirectChannel(); + } + + @Bean + public DirectChannel testConsumer() { + return new DirectChannel(); + } + + @Bean + public ExecutorChannel testTracingChannel() { + return new ExecutorChannel(Executors.newSingleThreadExecutor()); + } + + @Bean + public Propagator propagator(Tracer tracer) { + return new Propagator() { + + // List of headers required for tracing propagation + @Override + public List fields() { + return Arrays.asList("foo", "bar"); + } + + // 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) { + setter.set(carrier, "foo", "some foo value"); + setter.set(carrier, "bar", "some bar value"); + } + + + // This is called on the consumer side when the message is consumed + // Normally we would use tools like Extractor from tracing but for tests we are just manually creating a span + @Override + public Span.Builder extract(C carrier, Getter getter) { + String foo = getter.get(carrier, "foo"); + String bar = getter.get(carrier, "bar"); + return tracer.spanBuilder().kind(Span.Kind.CONSUMER).tag("foo", foo).tag("bar", bar); + } + }; + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java index d085bf4d81..2ce303a5cb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/transformer/TransformerContextTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2022 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. @@ -18,8 +18,7 @@ package org.springframework.integration.transformer; import static org.assertj.core.api.Assertions.assertThat; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -30,8 +29,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * Also in JMX - changes here should be reflected there. @@ -40,8 +38,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Gary Russell * @author Artem Bilan */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig public class TransformerContextTests { private static volatile int adviceCalled; @@ -66,22 +63,22 @@ public class TransformerContextTests { @Test public void methodInvokingTransformer() { - this.input.send(new GenericMessage("foo")); + this.input.send(new GenericMessage<>("foo")); Message reply = this.output.receive(0); assertThat(reply.getPayload()).isEqualTo("FOO"); assertThat(adviceCalled).isEqualTo(1); - this.direct.send(new GenericMessage("foo")); + this.direct.send(new GenericMessage<>("foo")); reply = this.output.receive(0); assertThat(reply.getPayload()).isEqualTo("FOO"); StackTraceElement[] st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertThat(st[6].getMethodName()).isEqualTo("doSend"); // no MethodInvokerHelper + assertThat(st[7].getMethodName()).isEqualTo("doSend"); // no MethodInvokerHelper - this.directRef.send(new GenericMessage("foo")); + this.directRef.send(new GenericMessage<>("foo")); reply = this.output.receive(0); assertThat(reply.getPayload()).isEqualTo("FOO"); st = (StackTraceElement[]) reply.getHeaders().get("callStack"); - assertThat(st[6].getMethodName()).isEqualTo("doSend"); // no MethodInvokerHelper + assertThat(st[7].getMethodName()).isEqualTo("doSend"); // no MethodInvokerHelper assertThat(this.testBean.isRunning()).isTrue(); this.pojoTransformer.stop(); @@ -89,7 +86,7 @@ public class TransformerContextTests { this.pojoTransformer.start(); assertThat(this.testBean.isRunning()).isTrue(); - this.directRef.send(new GenericMessage("bar")); + this.directRef.send(new GenericMessage<>("bar")); assertThat(this.output.receive(0)).isNull(); } @@ -117,4 +114,5 @@ public class TransformerContextTests { } } + }