Add ProducerBuilderCustomizer to support batching/chunking config

Resolves #80
2022-09-02 20:45:18 -05:00
parent 597bacde41
commit a3cc1846d0
9 changed files with 361 additions and 190 deletions
--- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc
+++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc
@@ -51,6 +51,21 @@ template.newMessage(msg)
 ----
 ====

+====== Producer customization
+A `ProducerBuilderCustomizer` can be specified in order to configure the underlying Pulsar producer builder that  ultimately constructs the producer used to send the outgoing message.
+
+WARNING: Use with caution as this gives full access to the producer builder and invoking some of its method's may have unintended side effects (eg. `create`).
+
+For example, the following code shows how to disable batching and enable chunking:
+====
+[source, java]
+----
+template.newMessage(msg)
+        .withProducerCustomizer((pb) -> pb.enableChunking(true).enableBatching(false))
+        .send();
+----
+====
+
 ====== Custom routing
 You can use custom routing when publishing records to partitioned topics. Simple specify your custom `MessageRouter` implementation on the fluent builder such as:
 ====
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java
@@ -92,28 +92,30 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	}

 	@Override
-	public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
-			List<ProducerInterceptor> producerInterceptors) {
+	protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
+			List<ProducerInterceptor> producerInterceptors,
+			List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) {
 		final String topicName = ProducerUtils.resolveTopicName(topic, this);
 		ProducerCacheKey<T> producerCacheKey = new ProducerCacheKey<>(schema, topicName, messageRouter,
 				producerInterceptors);
-		return this.producerCache.get(producerCacheKey, (st) -> {
-			try {
-				return this.doCreateProducer(st.topic, st.schema, st.router, producerInterceptors);
-			}
-			catch (PulsarClientException ex) {
-				throw new RuntimeException(ex);
-			}
-		});
+		return this.producerCache.get(producerCacheKey, (st) -> createCacheableProducer(st.topic, st.schema, st.router,
+				st.interceptors, producerBuilderCustomizers));
 	}

-	@Override
-	protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
-			List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
-		Producer<T> producer = super.doCreateProducer(topic, schema, messageRouter, producerInterceptors);
-		return wrapProducerWithCloseCallback(producer,
-				(p) -> this.logger.trace(() -> String.format("Client closed producer %s but will skip actual closing",
-						ProducerUtils.formatProducer(producer))));
+	private Producer<T> createCacheableProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
+			List<ProducerInterceptor> producerInterceptors,
+			List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) {
+		try {
+			Producer<T> producer = super.doCreateProducer(topic, schema, messageRouter, producerInterceptors,
+					producerBuilderCustomizers);
+			return wrapProducerWithCloseCallback(producer,
+					(p) -> this.logger
+							.trace(() -> String.format("Client closed producer %s but will skip actual closing",
+									ProducerUtils.formatProducer(producer))));
+		}
+		catch (PulsarClientException ex) {
+			throw new RuntimeException(ex);
+		}
 	}

 	@SuppressWarnings("unchecked")
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
@@ -57,23 +57,43 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>

 	@Override
 	public Producer<T> createProducer(String topic, Schema<T> schema) throws PulsarClientException {
-		return createProducer(topic, schema, null, null);
+		return doCreateProducer(topic, schema, null, null, null);
 	}

 	@Override
 	public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter)
 			throws PulsarClientException {
-		return createProducer(topic, schema, messageRouter, null);
+		return doCreateProducer(topic, schema, messageRouter, null, null);
 	}

 	@Override
 	public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
 			List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
-		return doCreateProducer(topic, schema, messageRouter, producerInterceptors);
+		return doCreateProducer(topic, schema, messageRouter, producerInterceptors, null);
 	}

+	@Override
+	public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
+			List<ProducerInterceptor> producerInterceptors,
+			List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException {
+		return doCreateProducer(topic, schema, messageRouter, producerInterceptors, producerBuilderCustomizers);
+	}
+
+	/**
+	 * Create the actual producer.
+	 * @param topic the topic the producer will send messages to or {@code null} to use
+	 * the default topic
+	 * @param schema the schema of the messages to be sent
+	 * @param messageRouter the optional message router to use
+	 * @param producerInterceptors the optional producer interceptors to use
+	 * @param producerBuilderCustomizers the optional list of customizers to apply to the
+	 * producer builder
+	 * @return the created producer
+	 * @throws PulsarClientException if any error occurs
+	 */
 	protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
-			List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
+			List<ProducerInterceptor> producerInterceptors,
+			List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException {
 		final String resolvedTopic = ProducerUtils.resolveTopicName(topic, this);
 		this.logger.trace(() -> String.format("Creating producer for '%s' topic", resolvedTopic));
 		final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
@@ -87,6 +107,9 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 		if (!CollectionUtils.isEmpty(producerInterceptors)) {
 			producerBuilder.intercept(producerInterceptors.toArray(new ProducerInterceptor[0]));
 		}
+		if (!CollectionUtils.isEmpty(producerBuilderCustomizers)) {
+			producerBuilderCustomizers.forEach((c) -> c.customize(producerBuilder));
+		}
 		return producerBuilder.create();
 	}

--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerBuilderCustomizer.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerBuilderCustomizer.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.core;
+
+import org.apache.pulsar.client.api.ProducerBuilder;
+
+/**
+ * The interface to customize a {@link ProducerBuilder}.
+ *
+ * @param <T> The message payload type
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface ProducerBuilderCustomizer<T> {
+
+	/**
+	 * Customizes a {@link ProducerBuilder}.
+	 * @param producerBuilder the builder to customize
+	 */
+	void customize(ProducerBuilder<T> producerBuilder);
+
+}
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java
@@ -103,6 +103,13 @@ public interface PulsarOperations<T> {
 		 */
 		SendMessageBuilder<T> withCustomRouter(MessageRouter messageRouter);

+		/**
+		 * Specifies the customizer to use to further configure the producer builder.
+		 * @param producerCustomizer the producer builder customizer
+		 * @return the current builder with the producer builder customizer specified
+		 */
+		SendMessageBuilder<T> withProducerCustomizer(ProducerBuilderCustomizer<T> producerCustomizer);
+
 		/**
 		 * Send the message in a blocking manner using the configured specification.
 		 * @return the id assigned by the broker to the published message
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
@@ -70,6 +70,22 @@ public interface PulsarProducerFactory<T> {
 	Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
 			List<ProducerInterceptor> producerInterceptors) throws PulsarClientException;

+	/**
+	 * Create a producer.
+	 * @param topic the topic the producer will send messages to or {@code null} to use
+	 * the default topic
+	 * @param schema the schema of the messages to be sent
+	 * @param messageRouter the optional message router to use
+	 * @param producerInterceptors the optional producer interceptors to use
+	 * @param producerBuilderCustomizers the optional list of customizers to apply to the
+	 * producer builder
+	 * @return the producer
+	 * @throws PulsarClientException if any error occurs
+	 */
+	Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
+			List<ProducerInterceptor> producerInterceptors,
+			List<ProducerBuilderCustomizer<T>> producerBuilderCustomizers) throws PulsarClientException;
+
 	/**
 	 * Return a map of configuration options to use when creating producers.
 	 * @return the map of configuration options
--- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
+++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
@@ -16,6 +16,7 @@

 package org.springframework.pulsar.core;

+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;

@@ -68,22 +69,22 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {

 	@Override
 	public MessageId send(T message) throws PulsarClientException {
-		return doSend(null, message, null, null);
+		return doSend(null, message, null, null, null);
 	}

 	@Override
 	public MessageId send(String topic, T message) throws PulsarClientException {
-		return doSend(topic, message, null, null);
+		return doSend(topic, message, null, null, null);
 	}

 	@Override
 	public CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
-		return doSendAsync(null, message, null, null);
+		return doSendAsync(null, message, null, null, null);
 	}

 	@Override
 	public CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException {
-		return doSendAsync(topic, message, null, null);
+		return doSendAsync(topic, message, null, null, null);
 	}

 	@Override
@@ -100,9 +101,9 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 	}

 	private MessageId doSend(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
-			MessageRouter messageRouter) throws PulsarClientException {
+			MessageRouter messageRouter, ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
 		try {
-			return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter).get();
+			return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter, producerCustomizer).get();
 		}
 		catch (Exception ex) {
 			throw PulsarClientException.unwrap(ex);
@@ -110,11 +111,11 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 	}

 	private CompletableFuture<MessageId> doSendAsync(String topic, T message,
-			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter)
-			throws PulsarClientException {
+			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter,
+			ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
 		final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory);
 		this.logger.trace(() -> String.format("Sending msg to '%s' topic", topicName));
-		final Producer<T> producer = prepareProducerForSend(topic, message, messageRouter);
+		final Producer<T> producer = prepareProducerForSend(topic, message, messageRouter, producerCustomizer);
 		TypedMessageBuilder<T> messageBuilder = producer.newMessage().value(message);
 		if (typedMessageBuilderCustomizer != null) {
 			typedMessageBuilderCustomizer.customize(messageBuilder);
@@ -132,10 +133,11 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 		});
 	}

-	private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter)
-			throws PulsarClientException {
+	private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter,
+			ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
 		Schema<T> schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message);
-		return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors);
+		return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors,
+				producerCustomizer == null ? Collections.emptyList() : Collections.singletonList(producerCustomizer));
 	}

 	public static class SendMessageBuilderImpl<T> implements SendMessageBuilder<T> {
@@ -150,6 +152,8 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {

 		private MessageRouter messageRouter;

+		private ProducerBuilderCustomizer<T> producerCustomizer;
+
 		SendMessageBuilderImpl(PulsarTemplate<T> template, T message) {
 			this.template = template;
 			this.message = message;
@@ -173,14 +177,22 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 			return this;
 		}

+		@Override
+		public SendMessageBuilder<T> withProducerCustomizer(ProducerBuilderCustomizer<T> producerCustomizer) {
+			this.producerCustomizer = producerCustomizer;
+			return this;
+		}
+
 		@Override
 		public MessageId send() throws PulsarClientException {
-			return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter);
+			return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter,
+					this.producerCustomizer);
 		}

 		@Override
 		public CompletableFuture<MessageId> sendAsync() throws PulsarClientException {
-			return this.template.doSendAsync(this.topic, this.message, this.messageCustomizer, this.messageRouter);
+			return this.template.doSendAsync(this.topic, this.message, this.messageCustomizer, this.messageRouter,
+					this.producerCustomizer);
 		}

 	}
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java
@@ -18,14 +18,18 @@ package org.springframework.pulsar.core;

 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;

+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;

 import org.apache.pulsar.client.api.MessageRouter;
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
@@ -108,6 +112,30 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests {
 		}
 	}

+	@Test
+	@SuppressWarnings("unchecked")
+	void createProducerWithSingleProducerCustomizer() throws PulsarClientException {
+		PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
+		ProducerBuilderCustomizer<String> producerCustomizer = mock(ProducerBuilderCustomizer.class);
+		try (Producer<String> producer = producerFactory.createProducer("topic0", schema, null, null,
+				Collections.singletonList(producerCustomizer))) {
+			verify(producerCustomizer).customize(any(ProducerBuilder.class));
+		}
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	void createProducerWithMultipleProducerCustomizers() throws PulsarClientException {
+		PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
+		ProducerBuilderCustomizer<String> producerCustomizer1 = mock(ProducerBuilderCustomizer.class);
+		ProducerBuilderCustomizer<String> producerCustomizer2 = mock(ProducerBuilderCustomizer.class);
+		try (Producer<String> producer = producerFactory.createProducer("topic0", schema, null, null,
+				Arrays.asList(producerCustomizer1, producerCustomizer2))) {
+			verify(producerCustomizer1).customize(any(ProducerBuilder.class));
+			verify(producerCustomizer2).customize(any(ProducerBuilder.class));
+		}
+	}
+
 	@Test
 	void createProducerWithNoTopic() {
 		PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
--- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
+++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -37,10 +38,8 @@ import java.util.stream.Stream;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageRouter;
 import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.TopicMetadata;
 import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
@@ -51,6 +50,8 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;

+import org.springframework.pulsar.core.PulsarOperations.SendMessageBuilder;
+
 /**
 * Tests for {@code PulsarTemplate}.
 *
@@ -60,52 +61,6 @@ import org.junit.jupiter.params.provider.MethodSource;
 */
 class PulsarTemplateTests extends AbstractContainerBaseTests {

-	private static final String SAMPLE_MESSAGE_KEY = "sample-key";
-
-	private static final TypedMessageBuilderCustomizer<String> sampleMessageKeyCustomizer = messageBuilder -> messageBuilder
-			.key(SAMPLE_MESSAGE_KEY);
-
-	@ParameterizedTest(name = "{0}")
-	@MethodSource("sendMessageTestProvider")
-	void sendMessageTest(String topic, Map<String, Object> producerConfig, SendHandler<Object> handler,
-			TypedMessageBuilderCustomizer<String> typedMessageBuilderCustomizer, MessageRouter router)
-			throws Exception {
-		String subscription = topic + "-sub";
-		String msgPayload = topic + "-msg";
-		if (router != null) {
-			try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build()) {
-				admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1);
-			}
-		}
-		try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) {
-			try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic)
-					.subscriptionName(subscription).subscribe()) {
-				PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
-						producerConfig);
-				PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory);
-
-				Object sendResponse = handler.doSend(pulsarTemplate, topic, msgPayload, typedMessageBuilderCustomizer,
-						router);
-				if (sendResponse instanceof CompletableFuture) {
-					sendResponse = ((CompletableFuture<?>) sendResponse).get(3, TimeUnit.SECONDS);
-				}
-				assertThat(sendResponse).isNotNull();
-
-				CompletableFuture<Message<String>> receiveMsgFuture = consumer.receiveAsync();
-				Message<String> msg = receiveMsgFuture.get(3, TimeUnit.SECONDS);
-				if (typedMessageBuilderCustomizer != null) {
-					assertThat(msg.getKey()).isEqualTo(SAMPLE_MESSAGE_KEY);
-				}
-				assertThat(msg.getData()).asString().isEqualTo(msgPayload);
-
-				// Make sure the producer was closed by the template (albeit indirectly as
-				// client removes closed producers)
-				await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(client).extracting("producers")
-						.asInstanceOf(InstanceOfAssertFactories.COLLECTION).isEmpty());
-			}
-		}
-	}
-
 	@ParameterizedTest(name = "{0}")
 	@MethodSource("interceptorInvocationTestProvider")
 	void interceptorInvocationTest(String topic, List<ProducerInterceptor> interceptors) throws Exception {
@@ -120,6 +75,14 @@ class PulsarTemplateTests extends AbstractContainerBaseTests {
 		}
 	}

+	private static Stream<Arguments> interceptorInvocationTestProvider() {
+		return Stream.of(
+				arguments(Named.of("testSingleInterceptor", "iit-topic-1"),
+						Collections.singletonList(mock(ProducerInterceptor.class))),
+				arguments(Named.of("testMultipleInterceptors", "iit-topic-2"),
+						List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class))));
+	}
+
 	@Test
 	void sendMessageWithSpecificSchemaTest() throws Exception {
 		String topic = "smt-specific-schema-topic";
@@ -138,123 +101,192 @@ class PulsarTemplateTests extends AbstractContainerBaseTests {
 		}
 	}

+	@ParameterizedTest(name = "{0}")
+	@MethodSource("sendMessageTestProvider")
+	void sendMessageTest(String testName, SendTestArgs testArgs) throws Exception {
+		// Use the test args to construct the params to pass to send handler
+		String topic = testName;
+		String subscription = topic + "-sub";
+		String msgPayload = topic + "-msg";
+		MessageRouter router = null;
+		if (testArgs.useCustomRouter) {
+			router = mock(MessageRouter.class);
+			when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0);
+		}
+		TypedMessageBuilderCustomizer<String> messageCustomizer = null;
+		if (testArgs.useMessageCustomizer) {
+			messageCustomizer = (mb) -> mb.key("foo-key");
+		}
+		ProducerBuilderCustomizer<String> producerCustomizer = null;
+		if (testArgs.useProducerCustomizer) {
+			producerCustomizer = (pb) -> pb.producerName("foo-producer");
+		}
+
+		if (router != null) {
+			try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build()) {
+				admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1);
+			}
+		}
+		try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) {
+			try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic)
+					.subscriptionName(subscription).subscribe()) {
+				Map<String, Object> producerConfig = testArgs.useSpecificTopic ? Collections.emptyMap()
+						: Collections.singletonMap("topicName", topic);
+				PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
+						producerConfig);
+				PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory);
+				Object sendResponse;
+				if (testArgs.useSimpleApi) {
+					if (testArgs.useAsyncSend) {
+						sendResponse = testArgs.useSpecificTopic ? pulsarTemplate.sendAsync(topic, msgPayload)
+								: pulsarTemplate.sendAsync(msgPayload);
+					}
+					else {
+						sendResponse = testArgs.useSpecificTopic ? pulsarTemplate.send(topic, msgPayload)
+								: pulsarTemplate.send(msgPayload);
+					}
+				}
+				else {
+					SendMessageBuilder<String> messageBuilder = pulsarTemplate.newMessage(msgPayload);
+					if (testArgs.useSpecificTopic) {
+						messageBuilder = messageBuilder.withTopic(topic);
+					}
+					if (messageCustomizer != null) {
+						messageBuilder = messageBuilder.withMessageCustomizer(messageCustomizer);
+					}
+					if (router != null) {
+						messageBuilder = messageBuilder.withCustomRouter(router);
+					}
+					if (producerCustomizer != null) {
+						messageBuilder = messageBuilder.withProducerCustomizer(producerCustomizer);
+					}
+					sendResponse = testArgs.useAsyncSend ? messageBuilder.sendAsync() : messageBuilder.send();
+				}
+
+				if (sendResponse instanceof CompletableFuture) {
+					sendResponse = ((CompletableFuture<?>) sendResponse).get(3, TimeUnit.SECONDS);
+				}
+				assertThat(sendResponse).isNotNull();
+
+				CompletableFuture<Message<String>> receiveMsgFuture = consumer.receiveAsync();
+				Message<String> msg = receiveMsgFuture.get(3, TimeUnit.SECONDS);
+
+				assertThat(msg.getData()).asString().isEqualTo(msgPayload);
+				if (messageCustomizer != null) {
+					assertThat(msg.getKey()).isEqualTo("foo-key");
+				}
+				if (router != null) {
+					verify(router).choosePartition(argThat((Message<String> m) -> m.getTopicName().equals(topic)),
+							any(TopicMetadata.class));
+				}
+				if (producerCustomizer != null) {
+					assertThat(msg.getProducerName()).isEqualTo("foo-producer");
+				}
+				// Make sure the producer was closed by the template (albeit indirectly as
+				// client removes closed producers)
+				await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> assertThat(client).extracting("producers")
+						.asInstanceOf(InstanceOfAssertFactories.COLLECTION).isEmpty());
+			}
+		}
+	}
+
 	private static Stream<Arguments> sendMessageTestProvider() {
-		return Stream.of(
-				arguments("sendMessageToDefaultTopic",
-						Collections.singletonMap("topicName", "sendMessageToDefaultTopic"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.send(),
-						null, null),
+		return Stream.of(arguments("sendMessageToDefaultTopic", SendTestArgs.useSpecificTopic(false)),
 				arguments("sendMessageToDefaultTopicWithSimpleApi",
-						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithSimpleApi"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(msg), null,
-						null),
+						SendTestArgs.useSpecificTopic(false).useSimpleApi(true)),
 				arguments("sendMessageToDefaultTopicWithRouter",
-						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithRouter"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withCustomRouter(router).send(),
-						null, mockRouter()),
-				arguments("sendMessageToDefaultTopicWithCustomizer",
-						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizer"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withMessageCustomizer(customizer).send(),
-						sampleMessageKeyCustomizer, null),
-				arguments("sendMessageToDefaultTopicWithCustomizerAndRouter",
-						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizerAndRouter"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withMessageCustomizer(customizer).withCustomRouter(router).send(),
-						sampleMessageKeyCustomizer, mockRouter()),
-				arguments("sendMessageToSpecificTopic", Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withTopic(topic).send(),
-						null, null),
-				arguments("sendMessageToSpecificTopicWithSimpleApi", Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(topic,
-								msg),
-						null, null),
-				arguments("sendMessageToSpecificTopicWithRouter", Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withTopic(topic).withCustomRouter(router).send(),
-						null, mockRouter()),
-				arguments("sendMessageToSpecificTopicWithCustomizer", Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withMessageCustomizer(customizer).withTopic(topic).send(),
-						sampleMessageKeyCustomizer, null),
-				arguments("sendMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
-								.withMessageCustomizer(customizer).withTopic(topic).withCustomRouter(router).send(),
-						sampleMessageKeyCustomizer, mockRouter()),
-				arguments("sendAsyncMessageToDefaultTopic",
-						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopic"),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).sendAsync(),
-						null, null),
+						SendTestArgs.useSpecificTopic(false).useCustomRouter(true)),
+				arguments("sendMessageToDefaultTopicWithMessageCustomizer",
+						SendTestArgs.useSpecificTopic(false).useMessageCustomizer(true)),
+				arguments("sendMessageToDefaultTopicWithProducerCustomizer",
+						SendTestArgs.useSpecificTopic(false).useProducerCustomizer(true)),
+				arguments("sendMessageToDefaultTopicWithAllOptions",
+						SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useMessageCustomizer(true)
+								.useProducerCustomizer(true)),
+				arguments("sendMessageToSpecificTopic", SendTestArgs.useSpecificTopic(true)),
+				arguments("sendMessageToSpecificTopicWithSimpleApi",
+						SendTestArgs.useSpecificTopic(true).useSimpleApi(true)),
+				arguments("sendMessageToSpecificTopicWithRouter",
+						SendTestArgs.useSpecificTopic(true).useCustomRouter(true)),
+				arguments("sendMessageToSpecificTopicWithMessageCustomizer",
+						SendTestArgs.useSpecificTopic(true).useMessageCustomizer(true)),
+				arguments("sendMessageToSpecificTopicWithProducerCustomizer",
+						SendTestArgs.useSpecificTopic(true).useProducerCustomizer(true)),
+				arguments("sendMessageToSpecificTopicWithAllOptions",
+						SendTestArgs.useSpecificTopic(true).useCustomRouter(true).useMessageCustomizer(true)
+								.useProducerCustomizer(true)),
+				arguments("sendAsyncMessageToDefaultTopic", SendTestArgs.useSpecificTopic(false).useAsyncSend(true)),
 				arguments("sendAsyncMessageToDefaultTopicWithSimpleApi",
-						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithSimpleApi"),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(msg),
-						null, null),
+						SendTestArgs.useSpecificTopic(false).useAsyncSend(true).useSimpleApi(true)),
 				arguments("sendAsyncMessageToDefaultTopicWithRouter",
-						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithRouter"),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withCustomRouter(router).sendAsync(),
-						null, mockRouter()),
-				arguments("sendAsyncMessageToDefaultTopicWithCustomizer",
-						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizer"),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withMessageCustomizer(customizer).sendAsync(),
-						sampleMessageKeyCustomizer, null),
-				arguments("sendAsyncMessageToDefaultTopicWithCustomizerAndRouter",
-						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizerAndRouter"),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withMessageCustomizer(customizer)
-										.withCustomRouter(router).sendAsync(),
-						sampleMessageKeyCustomizer, mockRouter()),
-				arguments("sendAsyncMessageToSpecificTopic", Collections.emptyMap(),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withTopic(topic).sendAsync(),
-						null, null),
-				arguments("sendAsyncMessageToSpecificTopicWithSimpleApi", Collections.emptyMap(),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(topic, msg),
-						null, null),
-				arguments("sendAsyncMessageToSpecificTopicWithRouter", Collections.emptyMap(),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withTopic(topic).withCustomRouter(router)
-										.sendAsync(),
-						null, mockRouter()),
-				arguments("sendAsyncMessageToSpecificTopicWithCustomizer", Collections.emptyMap(),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic)
-										.sendAsync(),
-						sampleMessageKeyCustomizer, null),
-				arguments("sendAsyncMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(),
-						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic)
-										.withCustomRouter(router).sendAsync(),
-						sampleMessageKeyCustomizer, mockRouter()));
+						SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToDefaultTopicWithMessageCustomizer",
+						SendTestArgs.useSpecificTopic(false).useMessageCustomizer(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToDefaultTopicWithProducerCustomizer",
+						SendTestArgs.useSpecificTopic(false).useProducerCustomizer(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToDefaultTopicWithAllOptions",
+						SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useMessageCustomizer(true)
+								.useProducerCustomizer(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToSpecificTopic", SendTestArgs.useSpecificTopic(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToSpecificTopicWithSimpleApi",
+						SendTestArgs.useSpecificTopic(true).useAsyncSend(true).useSimpleApi(true)),
+				arguments("sendAsyncMessageToSpecificTopicWithRouter",
+						SendTestArgs.useSpecificTopic(true).useCustomRouter(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToSpecificTopicWithMessageCustomizer",
+						SendTestArgs.useSpecificTopic(true).useMessageCustomizer(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToSpecificTopicWithProducerCustomizer",
+						SendTestArgs.useSpecificTopic(true).useProducerCustomizer(true).useAsyncSend(true)),
+				arguments("sendAsyncMessageToSpecificTopicWithAllOptions",
+						SendTestArgs.useSpecificTopic(true).useCustomRouter(true).useMessageCustomizer(true)
+								.useProducerCustomizer(true).useAsyncSend(true)));
 	}

-	private static Stream<Arguments> interceptorInvocationTestProvider() {
-		return Stream.of(
-				arguments(Named.of("testSingleInterceptor", "iit-topic-1"),
-						Collections.singletonList(mock(ProducerInterceptor.class))),
-				arguments(Named.of("testMultipleInterceptors", "iit-topic-2"),
-						List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class))));
-	}
+	static final class SendTestArgs {

-	private static MessageRouter mockRouter() {
-		MessageRouter router = mock(MessageRouter.class);
-		when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0);
-		return router;
-	}
+		private boolean useSpecificTopic;

-	@FunctionalInterface
-	interface SendHandler<V> {
+		private boolean useCustomRouter;

-		V doSend(PulsarTemplate<String> template, String topic, String msg,
-				TypedMessageBuilderCustomizer<String> typedMessageBuilderCustomizer, MessageRouter router)
-				throws PulsarClientException;
+		private boolean useMessageCustomizer;
+
+		private boolean useProducerCustomizer;
+
+		private boolean useAsyncSend;
+
+		private boolean useSimpleApi;
+
+		private SendTestArgs(boolean useSpecificTopic) {
+			this.useSpecificTopic = useSpecificTopic;
+		}
+
+		static SendTestArgs useSpecificTopic(boolean useSpecificTopic) {
+			return new SendTestArgs(useSpecificTopic);
+		}
+
+		SendTestArgs useCustomRouter(boolean useCustomRouter) {
+			this.useCustomRouter = useCustomRouter;
+			return this;
+		}
+
+		SendTestArgs useMessageCustomizer(boolean useMessageCustomizer) {
+			this.useMessageCustomizer = useMessageCustomizer;
+			return this;
+		}
+
+		SendTestArgs useProducerCustomizer(boolean useProducerCustomizer) {
+			this.useProducerCustomizer = useProducerCustomizer;
+			return this;
+		}
+
+		SendTestArgs useAsyncSend(boolean useAsyncSend) {
+			this.useAsyncSend = useAsyncSend;
+			return this;
+		}
+
+		SendTestArgs useSimpleApi(boolean useSimpleApi) {
+			this.useSimpleApi = useSimpleApi;
+			return this;
+		}

 	}