From 1a94d403379dbd2efcb723ac4b0b0f872e925d02 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Tue, 11 Oct 2022 00:18:55 -0500 Subject: [PATCH] Enable observations tests --- ...currentPulsarMessageListenerContainer.java | 5 -- ...DefaultPulsarMessageListenerContainer.java | 5 -- .../ObservationIntegrationTests.java | 53 +++++++++++++++---- .../pulsar/observation/ObservationTests.java | 10 ++-- 4 files changed, 46 insertions(+), 27 deletions(-) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java index c4829677..3bb48568 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java @@ -130,11 +130,6 @@ public class ConcurrentPulsarMessageListenerContainer extends AbstractPulsarM } } - @Override - public boolean isRunning() { - return false; - } - public List> getContainers() { return this.containers; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index db30a948..8f14c077 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -142,11 +142,6 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } } - @Override - public void destroy() { - - } - private void publishConsumerStartingEvent() { this.startLatch.countDown(); ApplicationEventPublisher publisher = getApplicationEventPublisher(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java index bf211e44..593bab7e 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java @@ -19,6 +19,7 @@ package org.springframework.pulsar.observation; import static org.assertj.core.api.Assertions.assertThat; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -27,7 +28,6 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; -import org.junit.jupiter.api.Disabled; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; @@ -64,7 +64,6 @@ import io.micrometer.tracing.test.simple.SpansAssert; * @author Chris Bono * @see SampleTestRunner */ -@Disabled public class ObservationIntegrationTests extends SampleTestRunner implements PulsarTestContainerSupport { @SuppressWarnings("unchecked") @@ -73,13 +72,35 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul // template -> listener -> template -> listener return (bb, meterRegistry) -> { ObservationRegistry observationRegistry = getObservationRegistry(); - try (AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext()) { - applicationContext.registerBean(ObservationRegistry.class, () -> observationRegistry); - applicationContext.register(ObservationIntegrationTestAppConfig.class); - applicationContext.refresh(); - applicationContext.getBean(PulsarTemplate.class).send("obs1-topic", "hello"); - CountDownLatch latch = applicationContext.getBean(ObservationIntegrationTestAppListeners.class).latch; - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + try (AnnotationConfigApplicationContext appContext = new AnnotationConfigApplicationContext()) { + appContext.registerBean(ObservationRegistry.class, () -> observationRegistry); + appContext.register(ObservationIntegrationTestAppConfig.class); + appContext.refresh(); + + ObservationIntegrationTestAppListeners listeners = appContext + .getBean(ObservationIntegrationTestAppListeners.class); + PulsarTemplate template = appContext.getBean(PulsarTemplate.class); + + String msg = "hello-" + System.currentTimeMillis(); + listeners.expectMessage(msg); + template.send("obs1-topic", msg); + + boolean listen1Completed = listeners.latchesByMessageListen1.get(msg).await(10, TimeUnit.SECONDS); + boolean listen2Completed = listeners.latchesByMessageListen2.get(msg).await(10, TimeUnit.SECONDS); + assertThat(listen1Completed).withFailMessage( + "Message %s not received in listen1 (latchesByMessageListen1 = %s and latchesByMessageListen2 = %s)", + msg, listeners.latchesByMessageListen1, listeners.latchesByMessageListen2).isTrue(); + assertThat(listen2Completed).withFailMessage( + "Message %s not received in listen2 (latchesByMessageListen1 = %s and latchesByMessageListen2 = %s)", + msg, listeners.latchesByMessageListen1, listeners.latchesByMessageListen2).isTrue(); + + // Without this sleep, the 2nd tracingSetup run sometimes fails due to + // messages from 1st run being + // delivered during the 2nd run. The test runs share the same listener + // config, including the + // same subscription names. Seems like the listener in run2 is getting + // message from run1. + Thread.sleep(5000); } List finishedSpans = bb.getFinishedSpans(); @@ -165,20 +186,30 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul private PulsarTemplate template; - CountDownLatch latch = new CountDownLatch(1); + Map latchesByMessageListen1 = new HashMap<>(); + + Map latchesByMessageListen2 = new HashMap<>(); ObservationIntegrationTestAppListeners(PulsarTemplate template) { this.template = template; } + void expectMessage(String message) { + latchesByMessageListen1.put(message, new CountDownLatch(1)); + latchesByMessageListen2.put(message, new CountDownLatch(1)); + } + @PulsarListener(id = "obs1-id", properties = { "subscriptionName=obs1-sub", "topicNames=obs1-topic" }) void listen1(String message) throws PulsarClientException { + assertThat(latchesByMessageListen1).containsKey(message); + latchesByMessageListen1.get(message).countDown(); this.template.send("obs2-topic", message); } @PulsarListener(id = "obs2-id", properties = { "subscriptionName=obs2-sub", "topicNames=obs2-topic" }) void listen2(String message) { - latch.countDown(); + assertThat(latchesByMessageListen2).containsKey(message); + latchesByMessageListen2.get(message).countDown(); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java index 6570016f..43bc46d5 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java @@ -31,7 +31,6 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; @@ -83,7 +82,6 @@ import io.micrometer.tracing.test.simple.SimpleTracer; * * @author Chris Bono */ -@Disabled @SpringJUnitConfig @DirtiesContext public class ObservationTests implements PulsarTestContainerSupport { @@ -217,8 +215,8 @@ public class ObservationTests implements PulsarTestContainerSupport { return super.getLowCardinalityKeyValues(context).and(RECEIVER_EXTRA_TAG, context.getListenerId()); } }); - return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, - new PulsarContainerProperties(), observationRegistry); + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties, + observationRegistry); } @Bean @@ -301,12 +299,12 @@ public class ObservationTests implements PulsarTestContainerSupport { this.template = template; } - @PulsarListener(id = "obs1-id", properties = { "subscriptionName=obs1-sub", "topicNames=obs1-topic" }) + @PulsarListener(id = "obs1-id", properties = { "subscriptionName=obsTest-sub1", "topicNames=obs1-topic" }) void listen1(Message message) throws PulsarClientException { this.template.send("obs2-topic", message.getValue()); } - @PulsarListener(id = "obs2-id", properties = { "subscriptionName=obs2-sub", "topicNames=obs2-topic" }) + @PulsarListener(id = "obs2-id", properties = { "subscriptionName=obsTest-sub2", "topicNames=obs2-topic" }) void listen2(Message message) { this.message = message; this.latch.countDown();