Polish "Support relative topicPattern in PulsarListener"

Adds to the previous commit by
 - replacing commons StringUtils with Spring StringUtils
 - only resetting the topics pattern if it changed

Signed-off-by: onobc <[email protected]>

Author: onobc

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java

@@ -20,7 +20,6 @@
 import java.util.List;
 import java.util.regex.Pattern;
 
-import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumer;
 import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder;
@@ -29,6 +28,7 @@
 
 import org.springframework.pulsar.core.PulsarTopicBuilder;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Default implementation for {@link ReactivePulsarConsumerFactory}.
@@ -109,10 +109,12 @@ protected void ensureTopicsPatternFullyQualified(ReactiveMessageConsumerBuilder<
 		}
 		var mutableSpec = consumerBuilder.getMutableSpec();
 		var topicsPattern = mutableSpec.getTopicsPattern();
-		if (topicsPattern != null && StringUtils.isNoneBlank(topicsPattern.pattern())) {
+		if (topicsPattern != null && StringUtils.hasText(topicsPattern.pattern())) {
 			var topicsPatternStr = topicsPattern.pattern();
-			var fullyQualifiedTopicsPatternStr = this.topicBuilder.getFullyQualifiedNameForTopic(topicsPatternStr);
-			mutableSpec.setTopicsPattern(Pattern.compile(fullyQualifiedTopicsPatternStr));
+			var fqTopicsPatternStr = this.topicBuilder.getFullyQualifiedNameForTopic(topicsPatternStr);
+			if (!topicsPatternStr.equals(fqTopicsPatternStr)) {
+				mutableSpec.setTopicsPattern(Pattern.compile(fqTopicsPatternStr));
+			}
 		}
 	}
 

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java

@@ -138,7 +138,7 @@ void createConsumerEnsureTopicsPatternFullyQualified() {
 			var fullyQualifiedTopicsPattern = "persistent://public/default/my-topic-.*";
 			var consumer = consumerFactory.createConsumer(SCHEMA,
 					Collections.singletonList(builder -> builder.topicsPattern(Pattern.compile(inputTopicsPattern))));
-			ReactiveMessageConsumerSpec reactiveMessageConsumerSpec = assertThat(consumer)
+			var reactiveMessageConsumerSpec = assertThat(consumer)
 				.extracting("consumerSpec", InstanceOfAssertFactories.type(ReactiveMessageConsumerSpec.class))
 				.actual();
 			assertThat(reactiveMessageConsumerSpec.getTopicsPattern().pattern()).isEqualTo(fullyQualifiedTopicsPattern);

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java

@@ -25,7 +25,6 @@
 import java.util.TreeMap;
 import java.util.regex.Pattern;
 
-import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -36,6 +35,7 @@
 
 import org.springframework.pulsar.PulsarException;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Default implementation for {@link PulsarConsumerFactory}.
@@ -157,10 +157,12 @@ protected void ensureTopicsPatternFullyQualified(ConsumerBuilder<T> builder) {
 		}
 		var builderImpl = (ConsumerBuilderImpl<T>) builder;
 		var topicsPattern = builderImpl.getConf().getTopicsPattern();
-		if (topicsPattern != null && StringUtils.isNoneBlank(topicsPattern.pattern())) {
+		if (topicsPattern != null && StringUtils.hasText(topicsPattern.pattern())) {
 			var topicsPatternStr = topicsPattern.pattern();
-			var fullyQualifiedTopicsPatternStr = this.topicBuilder.getFullyQualifiedNameForTopic(topicsPatternStr);
-			builderImpl.getConf().setTopicsPattern(Pattern.compile(fullyQualifiedTopicsPatternStr));
+			var fqTopicsPatternStr = this.topicBuilder.getFullyQualifiedNameForTopic(topicsPatternStr);
+			if (!topicsPatternStr.equals(fqTopicsPatternStr)) {
+				builderImpl.getConf().setTopicsPattern(Pattern.compile(fqTopicsPatternStr));
+			}
 		}
 	}
 

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
+import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -280,7 +281,7 @@ void withPulsarTopicBuilderEnsureTopicNamesFullyQualified() throws PulsarClientE
 			DefaultPulsarConsumerFactory<String> consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
 					null);
 			consumerFactory.setTopicBuilder(pulsarTopicBuilder);
-			try (var consumer = consumerFactory.createConsumer(SCHEMA, Collections.singletonList("topic1"),
+			try (Consumer<String> consumer = consumerFactory.createConsumer(SCHEMA, Collections.singletonList("topic1"),
 					"with-pulsar-topic-builder-ensure-topic-names-fully-qualified-sub", null, null)) {
 				assertThat(consumer.getTopic()).isEqualTo("persistent://public/default/topic1");
 				verify(pulsarTopicBuilder).getFullyQualifiedNameForTopic("topic1");
@@ -302,16 +303,22 @@ void withPulsarTopicBuilderEnsureTopicsPatternFullyQualified() throws PulsarClie
 				var topicsPattern = patternMultiTopicsConsumer.getPattern();
 				assertThat(topicsPattern.inputPattern()).isEqualTo("persistent://public/default/topic-.*");
 				verify(pulsarTopicBuilder).getFullyQualifiedNameForTopic("topic-.*");
-				CompletableFuture<?> watcherFuture = assertThat(patternMultiTopicsConsumer)
-					.extracting("watcherFuture", InstanceOfAssertFactories.type(CompletableFuture.class))
-					.actual();
-				// Seems some bugs in PatternMultiTopicsConsumerImpl.closeAsync() method.
-				// If watcherFuture is not completed, invoke pulsarClient.close() first
-				// will cause exception.
-				watcherFuture.join();
+				temporarilyDealWithPulsar24698(patternMultiTopicsConsumer);
 			}
 		}
 
+		// TODO remove when Pulsar client updates to 4.2.0
+		private void temporarilyDealWithPulsar24698(PatternMultiTopicsConsumerImpl<String> consumer) {
+			// See https://github.com/apache/pulsar/pull/24698
+			// If this is not here there will be numerous exceptions when
+			// PulsarClient.close
+			CompletableFuture<?> watcherFuture = assertThat(consumer)
+				.extracting("watcherFuture", InstanceOfAssertFactories.type(CompletableFuture.class))
+				.actual();
+			watcherFuture.join();
+
+		}
+
 	}
 
 }