Allow dead letter topics to be non-fully-qualified.

The consumer factories now respect the configured PulsarTopicBuilder and
will fully-qualify the `deadLetterTopic` and `retryLetterTopic` on the
`DeadLetterPolicy`. This allows users to not have to fully-qualify
these topic names.

Resolves #1241

Signed-off-by: Andrey Litvitski <andrey1010102008@gmail.com>

Author: therepanic

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

@@ -101,6 +101,18 @@ protected void ensureTopicNamesFullyQualified(ReactiveMessageConsumerBuilder<T>
 			var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
 			mutableSpec.setTopicNames(fullyQualifiedTopics);
 		}
+		if (mutableSpec.getDeadLetterPolicy() != null) {
+			var dlt = mutableSpec.getDeadLetterPolicy().getDeadLetterTopic();
+			if (dlt != null) {
+				mutableSpec.getDeadLetterPolicy()
+					.setDeadLetterTopic(this.topicBuilder.getFullyQualifiedNameForTopic(dlt));
+			}
+			var rlt = mutableSpec.getDeadLetterPolicy().getRetryLetterTopic();
+			if (rlt != null) {
+				mutableSpec.getDeadLetterPolicy()
+					.setRetryLetterTopic(this.topicBuilder.getFullyQualifiedNameForTopic(rlt));
+			}
+		}
 	}
 
 	protected void ensureTopicsPatternFullyQualified(ReactiveMessageConsumerBuilder<T> consumerBuilder) {

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

@@ -149,6 +149,20 @@ protected void ensureTopicNamesFullyQualified(ConsumerBuilder<T> builder) {
 			var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
 			builderImpl.getConf().setTopicNames(new HashSet<>(fullyQualifiedTopics));
 		}
+		if (builderImpl.getConf().getDeadLetterPolicy() != null) {
+			var dlt = builderImpl.getConf().getDeadLetterPolicy().getDeadLetterTopic();
+			if (dlt != null) {
+				builderImpl.getConf()
+					.getDeadLetterPolicy()
+					.setDeadLetterTopic(this.topicBuilder.getFullyQualifiedNameForTopic(dlt));
+			}
+			var rlt = builderImpl.getConf().getDeadLetterPolicy().getRetryLetterTopic();
+			if (rlt != null) {
+				builderImpl.getConf()
+					.getDeadLetterPolicy()
+					.setRetryLetterTopic(this.topicBuilder.getFullyQualifiedNameForTopic(rlt));
+			}
+		}
 	}
 
 	protected void ensureTopicsPatternFullyQualified(ConsumerBuilder<T> builder) {