Polish "Enhance PulsarConsumerTestUtil consumeMessages"

- Add test for consume w/ no args w/ default url
- Add javadocs to PulsarConsumerTestUtil

spring-pulsar-test/spring-pulsar-test.gradle

@@ -11,4 +11,5 @@ dependencies {
     implementation project(':spring-pulsar')
     testImplementation 'org.assertj:assertj-core'
     testImplementation 'org.junit.jupiter:junit-jupiter'
+	testImplementation 'org.mockito:mockito-core'
 }

spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java

@@ -42,8 +42,10 @@ import org.springframework.util.Assert;
  *
  * @param <T> the type of the message payload
  * @author Jonas Geiregat
+ * @author Kartik Shrivastava
+ * @author Chris Bono
  */
-public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, ConditionsSpec<T> {
+public final class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, ConditionsSpec<T> {
 
 	private final PulsarConsumerFactory<T> consumerFactory;
 
@@ -57,6 +59,13 @@ public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, C
 
 	private boolean untilMethodAlreadyCalled = false;
 
+	/**
+	 * Begin a builder which will consume messages from the
+	 * {@link PulsarTestContainerSupport} (if available) or the default Pulsar broker url
+	 * {@code pulsar://localhost:6650}.
+	 * @param <T> the message payload type
+	 * @return the {@link TopicSpec topic step} of the builder
+	 */
 	public static <T> TopicSpec<T> consumeMessages() {
 		if (PulsarTestContainerSupport.isContainerStarted()) {
 			return PulsarConsumerTestUtil.consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl());
@@ -64,6 +73,12 @@ public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, C
 		return PulsarConsumerTestUtil.consumeMessages("pulsar://localhost:6650");
 	}
 
+	/**
+	 * Begin a builder which will consume messages from the specified Pulsar broker url.
+	 * @param <T> the message payload type
+	 * @param url the Pulsar broker url
+	 * @return the {@link TopicSpec topic step} of the builder
+	 */
 	public static <T> TopicSpec<T> consumeMessages(String url) {
 		Assert.notNull(url, "url must not be null");
 		try {
@@ -74,16 +89,28 @@ public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, C
 		}
 	}
 
+	/**
+	 * Begin a builder which will consume messages with a provided Pulsar client.
+	 * @param <T> the message payload type
+	 * @param pulsarClient the client to consume with
+	 * @return the {@link TopicSpec topic step} of the builder
+	 */
 	public static <T> TopicSpec<T> consumeMessages(PulsarClient pulsarClient) {
 		Assert.notNull(pulsarClient, "pulsarClient must not be null");
 		return PulsarConsumerTestUtil.consumeMessages(new DefaultPulsarConsumerFactory<>(pulsarClient, List.of()));
 	}
 
+	/**
+	 * Begin a builder which will consume messages with a provided consumer factory.
+	 * @param <T> the message payload type
+	 * @param pulsarConsumerFactory the consumer factory to consume with
+	 * @return the {@link TopicSpec topic step} of the builder
+	 */
 	public static <T> TopicSpec<T> consumeMessages(PulsarConsumerFactory<T> pulsarConsumerFactory) {
 		return new PulsarConsumerTestUtil<>(pulsarConsumerFactory);
 	}
 
-	public PulsarConsumerTestUtil(PulsarConsumerFactory<T> consumerFactory) {
+	private PulsarConsumerTestUtil(PulsarConsumerFactory<T> consumerFactory) {
 		Assert.notNull(consumerFactory, "PulsarConsumerFactory must not be null");
 		this.consumerFactory = consumerFactory;
 	}

spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java

@@ -16,6 +16,7 @@
 
 package org.springframework.pulsar.test.support;
 
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.testcontainers.containers.PulsarContainer;
 import org.testcontainers.junit.jupiter.Testcontainers;
@@ -25,6 +26,7 @@ import org.testcontainers.utility.DockerImageName;
  * Provides a static {@link PulsarContainer} that can be shared across test classes.
  *
  * @author Chris Bono
+ * @author Kartik Shrivastava
  */
 @Testcontainers(disabledWithoutDocker = true)
 public interface PulsarTestContainerSupport {
@@ -40,6 +42,11 @@ public interface PulsarTestContainerSupport {
 		PULSAR_CONTAINER.start();
 	}
 
+	@AfterAll
+	static void stopContainer() {
+		PULSAR_CONTAINER.stop();
+	}
+
 	static String getPulsarBrokerUrl() {
 		return PULSAR_CONTAINER.getPulsarBrokerUrl();
 	}
@@ -52,8 +59,4 @@ public interface PulsarTestContainerSupport {
 		return PULSAR_CONTAINER.isRunning();
 	}
 
-	static void stopContainer() {
-		PULSAR_CONTAINER.stop();
-	}
-
 }

spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java

@@ -29,9 +29,13 @@ import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
+import org.springframework.pulsar.PulsarException;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
@@ -41,6 +45,8 @@ import org.springframework.pulsar.core.PulsarTemplate;
  * Tests for {@link PulsarConsumerTestUtil}.
  *
  * @author Jonas Geiregat
+ * @author Kartik Shrivastava
+ * @author Chris Bono
  */
 class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport {
 
@@ -61,9 +67,16 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport {
 		this.pulsarTemplate = new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient));
 	}
 
+	@AfterEach
+	void cleanupFromTest() throws PulsarClientException {
+		if (this.pulsarClient != null) {
+			this.pulsarClient.close();
+		}
+	}
+
 	@Test
-	void whenConditionIsSpecifiedMessagesAreConsumedUntilConditionIsMet() {
-		var topic = testTopic("a");
+	void whenConditionIsSpecifiedThenMessagesConsumedUntilConditionMet() {
+		var topic = testTopic("cond");
 		IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
 		var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
 			.fromTopic(topic)
@@ -75,8 +88,8 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport {
 	}
 
 	@Test
-	void whenConditionIsNotSpecifiedMessagesAreConsumedUntilAwaitDuration() {
-		var topic = testTopic("b");
+	void whenConditionIsNotSpecifiedThenMessagesAreConsumedUntilAwaitDuration() {
+		var topic = testTopic("no-cond");
 		IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
 		var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
 			.fromTopic(topic)
@@ -89,81 +102,8 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport {
 	}
 
 	@Test
-	void exceptionIsThrownWhenConditionNotMetWithinAwaitDuration() {
-		assertThatExceptionOfType(ConditionTimeoutException.class)
-			.isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
-				.fromTopic(testTopic("c"))
-				.withSchema(Schema.STRING)
-				.awaitAtMost(Duration.ofSeconds(5))
-				.until(ConsumedMessagesConditions.desiredMessageCount(3))
-				.get())
-			.withMessage("Condition was not met within 5 seconds");
-	}
-
-	@Test
-	void messagesAreConsumedWhenContainerIsRunningAndConsumeMessagesIsCalledWithoutArguments() {
-		// depends upon pulsarClient created in prepareForTest
-		var topic = testTopic("e1");
-		IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
-		var msgs = PulsarConsumerTestUtil.<String>consumeMessages()
-			.fromTopic(topic)
-			.withSchema(Schema.STRING)
-			.awaitAtMost(Duration.ofSeconds(5))
-			.until(desiredMessageCount(2))
-			.get();
-		assertThat(msgs).hasSize(2);
-	}
-
-	@Test
-	void messagesAreConsumedWhenContainerIsStoppedAndConsumeMessagesIsCalledWithoutArguments() {
-		PulsarTestContainerSupport.stopContainer();
-		PulsarConsumerTestUtil.<String>consumeMessages();
-		// TODO: Complete this test
-	}
-
-	@Test
-	void messagesAreConsumedWhenConsumeMessagesIsCalledWithBrokerUrl() {
-		var topic = testTopic("e2");
-		IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
-		var msgs = PulsarConsumerTestUtil.<String>consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl())
-			.fromTopic(topic)
-			.withSchema(Schema.STRING)
-			.awaitAtMost(Duration.ofSeconds(5))
-			.until(desiredMessageCount(2))
-			.get();
-		assertThat(msgs).hasSize(2);
-	}
-
-	@Test
-	void exceptionIsThrownWhenConsumeMessagesIsCalledWithNullBrokerUrl() {
-		String url = null;
-		assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(url))
-			.withMessage("url must not be null");
-	}
-
-	@Test
-	void messagesAreConsumedWhenConsumeMessagesIsCalledWithPulsarClient() {
-		var topic = testTopic("e3");
-		IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
-		var msgs = PulsarConsumerTestUtil.<String>consumeMessages(this.pulsarClient)
-			.fromTopic(topic)
-			.withSchema(Schema.STRING)
-			.awaitAtMost(Duration.ofSeconds(5))
-			.until(desiredMessageCount(2))
-			.get();
-		assertThat(msgs).hasSize(2);
-	}
-
-	@Test
-	void exceptionIsThrownWhenConsumeMessagesIsCalledWithNullPulsarClient() {
-		PulsarClient localPulsarClient = null;
-		assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(localPulsarClient))
-			.withMessage("pulsarClient must not be null");
-	}
-
-	@Test
-	void whenChainedConditionAreSpecifiedMessagesAreConsumedUntilTheyAreMet() {
-		var topic = testTopic("d");
+	void whenChainedConditionsAreSpecifiedThenMessagesConsumedUntilAllConditionsMet() {
+		var topic = testTopic("chained-cond");
 		IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
 		ConsumedMessagesCondition<String> condition1 = ConsumedMessagesConditions.desiredMessageCount(5);
 		ConsumedMessagesCondition<String> condition2 = ConsumedMessagesConditions.atLeastOneMessageMatches("message-1");
@@ -177,9 +117,75 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport {
 	}
 
 	@Test
-	void exceptionIsThrownWhenUntilIsCalledMultipleTimes() {
-		var topic = testTopic("e");
-		IntStream.range(0, 1).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
+	void whenConditionNotMetWithinAwaitDurationThenExceptionIsThrown() {
+		assertThatExceptionOfType(ConditionTimeoutException.class)
+			.isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
+				.fromTopic(testTopic("cond-not-met"))
+				.withSchema(Schema.STRING)
+				.awaitAtMost(Duration.ofSeconds(5))
+				.until(ConsumedMessagesConditions.desiredMessageCount(3))
+				.get())
+			.withMessage("Condition was not met within 5 seconds");
+	}
+
+	@Test
+	void consumeMessagesWithNoArgsUsesPulsarContainerIfAvailable() {
+		var topic = testTopic("no-arg");
+		IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
+		var msgs = PulsarConsumerTestUtil.<String>consumeMessages()
+			.fromTopic(topic)
+			.withSchema(Schema.STRING)
+			.awaitAtMost(Duration.ofSeconds(5))
+			.until(desiredMessageCount(2))
+			.get();
+		assertThat(msgs).hasSize(2);
+	}
+
+	@Test
+	void consumeMessagesWithNoArgsUsesDefaultUrlWhenPulsarContainerNotAvailable() {
+		try (MockedStatic<PulsarTestContainerSupport> containerSupport = Mockito
+			.mockStatic(PulsarTestContainerSupport.class)) {
+			containerSupport.when(PulsarTestContainerSupport::isContainerStarted).thenReturn(false);
+			var topic = testTopic("no-arg-dft-url");
+			assertThatExceptionOfType(PulsarException.class)
+				.isThrownBy(() -> PulsarConsumerTestUtil.<String>consumeMessages()
+					.fromTopic(topic)
+					.withSchema(Schema.STRING)
+					.awaitAtMost(Duration.ofSeconds(2))
+					.get())
+				.withStackTraceContaining("Connection refused: localhost");
+		}
+	}
+
+	@Test
+	void consumeMessagesWithBrokerUrl() {
+		var topic = testTopic("url-arg");
+		IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
+		var msgs = PulsarConsumerTestUtil.<String>consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl())
+			.fromTopic(topic)
+			.withSchema(Schema.STRING)
+			.awaitAtMost(Duration.ofSeconds(5))
+			.until(desiredMessageCount(2))
+			.get();
+		assertThat(msgs).hasSize(2);
+	}
+
+	@Test
+	void consumeMessagesWithPulsarClient() {
+		var topic = testTopic("client-arg");
+		IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i));
+		var msgs = PulsarConsumerTestUtil.<String>consumeMessages(this.pulsarClient)
+			.fromTopic(topic)
+			.withSchema(Schema.STRING)
+			.awaitAtMost(Duration.ofSeconds(5))
+			.until(desiredMessageCount(2))
+			.get();
+		assertThat(msgs).hasSize(2);
+	}
+
+	@Test
+	void untilCannotBeCalledMultipleTimes() {
+		var topic = testTopic("until-multi");
 		assertThatExceptionOfType(IllegalStateException.class)
 			.isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
 				.fromTopic(topic)
@@ -191,6 +197,20 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport {
 			.withMessage("Multiple calls to 'until' are not allowed. Use 'and' to combine conditions.");
 	}
 
+	@Test
+	void brokerUrlCannotBeNull() {
+		String url = null;
+		assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(url))
+			.withMessage("url must not be null");
+	}
+
+	@Test
+	void pulsarClientCannotBeNull() {
+		PulsarClient localPulsarClient = null;
+		assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(localPulsarClient))
+			.withMessage("pulsarClient must not be null");
+	}
+
 	@Test
 	void consumerFactoryCannotBeNull() {
 		PulsarConsumerFactory<String> consumerFactory = null;