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

Enable observations tests

Browse Source
This commit is contained in:
Chris Bono
2022-10-11 00:18:55 -05:00
committed by Soby Chacko
parent 37309d1786
commit 1a94d40337
4 changed files with 46 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java
View File

@@ -130,11 +130,6 @@ public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarM
}
}
@Override
public boolean isRunning() {
return false;
}
public List<DefaultPulsarMessageListenerContainer<T>> getContainers() {
return this.containers;
}

5
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -142,11 +142,6 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
}
@Override
public void destroy() {
}
private void publishConsumerStartingEvent() {
this.startLatch.countDown();
ApplicationEventPublisher publisher = getApplicationEventPublisher();

53
spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java
View File

@@ -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<String> 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<FinishedSpan> finishedSpans = bb.getFinishedSpans();
@@ -165,20 +186,30 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
private PulsarTemplate<String> template;
CountDownLatch latch = new CountDownLatch(1);
Map<String, CountDownLatch> latchesByMessageListen1 = new HashMap<>();
Map<String, CountDownLatch> latchesByMessageListen2 = new HashMap<>();
ObservationIntegrationTestAppListeners(PulsarTemplate<String> 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();
}
}

10
spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java
View File

@@ -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<String> 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<String> message) {
this.message = message;
this.latch.countDown();