Add fluent API to PulsarTemplate

Fixes #68

spring-pulsar-docs/src/main/asciidoc/pulsar.adoc

@@ -23,7 +23,7 @@ include::application-properties/pulsar-client.adoc[lines=3..-1]
 [[pulsar-producer]]
 ==== Pulsar Producer
 
-On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarOperations.html[several variants of 'send' methods] to publish records through its contract.
+On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides methods to publish records through its contract.
 
 There are two categories of these send API methods - `send` and `sendAsync`.
 The `send` methods are blocking calls using the synchronous sending capabilities on the Pulsar producer.
@@ -31,20 +31,40 @@ They return the `MessageId` of the message that was published once the message i
 The `sendAsync` method calls are asynchronous calls that are non-blocking.
 They return a `CompletableFuture` using which you can asynchronously receive the message id once the messages are published.
 
-Both `send` and `sendAsync` methods have a variant that allows to publish simply with the message.
-When you do that, the application must provide the topic name using the property `spring.pulsar.producer.topicName`.
-There is another variant for both flavors that takes the topic name as an argument.
+===== Simple API
+The template provides a handful of methods ({javadocs}/org/springframework/pulsar/core/PulsarOperations.html[prefixed with _'send'_]) for simple send requests that contain only a message and/or destination topic. For more complicated send requests there is a fluent API that allows the user to configure more options (see below).
 
-The methods using a `TypedMessageBuilderCustomizer` allow modifying the basic `TypedMessageBuilder`.
-This can be used to send a keyed message for example:
+TIP: Both `send` and `sendAsync` methods have a variant that allows to publish simply with the message.
+When you do that, the application must provide the topic name using the property `spring.pulsar.producer.topic-name`.
 
+===== Fluent API
+The template provides a {javadocs}/org/springframework/pulsar/core/PulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests.
+
+====== Message customization
+A `TypedMessageBuilderCustomizer` can be specified in order to configure the outgoing message. For example, the following code shows how to send a keyed message:
 ====
 [source, java]
 ----
-template.send(msg, (messageBuilder -> messageBuilder.key(myMessageKey)));
+template.newMessage(msg)
+    .withMessageCustomizer((mb) -> mb.key("foo-msg-key"))
+    .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:
+====
+[source, java]
+----
+template.newMessage(msg)
+    .withCustomRouter(myCustomRouter)
+    .send();
+----
+====
+
+TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`.
+
+===== Schema
 If you are using simple Java primitive types, then the framework auto-detects the schema for you, and you do not need to specify any schema types for publishing the data.
 However, if you are using any complex types such as `JSON`, `AVRO`, `PROTOBUF`, etc. then you need to set the proper schema type on the `PulsarTemplate` before invoking any send operations as shown below.
 
@@ -55,11 +75,6 @@ pulsarTemplate.setSchema(Schema.JSON(Foo.class));
 ----
 ====
 
-
-When using partitioned topics, then you can use custom partitioning routing when publishing records.
-For this purpose, you can provide an implementation of a `MessageRouter` and pass it along with both `send` and `sendAsync` methods.
-Note that, when using a `MessageRouter`, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`.
-
 [[producer-application-properties]]
 .[.underline]#Click ##here## to view the available **Pulsar Producer Properties**#
 [%collapsible]
@@ -366,7 +381,7 @@ It is a topic that is partitioned and for this sample we assume that the topic i
 public class PulsarBootPartitioned {
 
 	public static void main(String[] args) {
-		SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition");
+		SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.message-routing-mode=CustomPartition");
 	}
 
 	@Bean
@@ -421,7 +436,7 @@ In order to do that, we are providing a message router object with the send meth
 Look at the three message routers implemented.
 `FooRouter` always sends data to partition `0`, `BarRouter` to partition `1` and `BuzzRouter` to partition `2`.
 Also note that, we are now using the `sendAsync` method of `PulsarTemplate` that returns a `CompletableFuture`.
-When running the application, we also need to set the `messageRoutingMode` on the producer to `CustomPartition` (`spring.pulsar.producer.messageRoutingMode`).
+When running the application, we also need to set the `messageRoutingMode` on the producer to `CustomPartition` (`spring.pulsar.producer.message-routing-mode`).
 
 On the consumer side, we are using a `PulsarListener` with the exclusive subscription type.
 This means that data from all the partitions will end up in the same consumer and there is no ordering guarantee.

spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java

@@ -45,9 +45,12 @@ public class FailoverConsumerApp {
 		String topic = "failover-demo-topic";
 		return args -> {
 			for (int i = 0; i < 10; i++) {
-				pulsarTemplate.sendAsync(topic, "hello john doe 0 ", new FooRouter());
-				pulsarTemplate.sendAsync(topic, "hello alice doe 1", new BarRouter());
-				pulsarTemplate.sendAsync(topic, "hello buzz doe 2", new BuzzRouter());
+				pulsarTemplate.newMessage("hello john doe 0 ").withTopic(topic).withCustomRouter(new FooRouter())
+						.sendAsync();
+				pulsarTemplate.newMessage("hello alice doe 1").withTopic(topic).withCustomRouter(new BarRouter())
+						.sendAsync();
+				pulsarTemplate.newMessage("hello buzz doe 2").withTopic(topic).withCustomRouter(new BuzzRouter())
+						.sendAsync();
 				Thread.sleep(1_000);
 			}
 		};

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

@@ -34,201 +34,90 @@ public interface PulsarOperations<T> {
 	/**
 	 * Sends a message to the default topic in a blocking manner.
 	 * @param message the message to send
-	 * @return the id of the sent message
+	 * @return the id assigned by the broker to the published message
 	 * @throws PulsarClientException if an error occurs
 	 */
-	default MessageId send(T message) throws PulsarClientException {
-		return send(null, message);
-	}
+	MessageId send(T message) throws PulsarClientException;
 
 	/**
 	 * Sends a message to the specified topic in a blocking manner.
 	 * @param topic the topic to send the message to or {@code null} to send to the
 	 * default topic
 	 * @param message the message to send
-	 * @return the id of the sent message
+	 * @return the id assigned by the broker to the published message
 	 * @throws PulsarClientException if an error occurs
 	 */
-	default MessageId send(String topic, T message) throws PulsarClientException {
-		return send(topic, message, (MessageRouter) null);
-	}
-
-	/**
-	 * Sends a message to the default topic in a blocking manner.
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer
-	 * @return the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default MessageId send(T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer)
-			throws PulsarClientException {
-		return send(message, typedMessageBuilderCustomizer, null);
-	}
-
-	/**
-	 * Sends a message to the default topic in a blocking manner.
-	 * @param message the message to send
-	 * @param messageRouter the optional message router to use
-	 * @return the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default MessageId send(T message, MessageRouter messageRouter) throws PulsarClientException {
-		return send(null, message, messageRouter);
-	}
-
-	/**
-	 * Sends a message to the specified topic in a blocking manner.
-	 * @param topic the topic to send the message to or {@code null} to send to the
-	 * default topic
-	 * @param message the message to send
-	 * @param messageRouter the optional message router to use
-	 * @return the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException {
-		return send(topic, message, null, messageRouter);
-	}
-
-	/**
-	 * Sends a message to the default topic in a blocking manner.
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the TypeMessageBuilder customizer
-	 * @param messageRouter the optional message router to use
-	 * @return the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default MessageId send(T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
-			MessageRouter messageRouter) throws PulsarClientException {
-		return send(null, message, typedMessageBuilderCustomizer, messageRouter);
-	}
-
-	/**
-	 * Sends a message to the specified topic in a blocking manner.
-	 * @param topic the topic to send the message to or {@code null} to send to the
-	 * default topic
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the TypeMessageBuilder customizer
-	 * @return the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default MessageId send(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer)
-			throws PulsarClientException {
-		return send(topic, message, typedMessageBuilderCustomizer, null);
-	}
-
-	/**
-	 * Sends a message to the specified topic in a blocking manner.
-	 * @param topic the topic to send the message to or {@code null} to send to the
-	 * default topic
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the optional TypedMessageBuilder customizer
-	 * @param messageRouter the optional message router to use
-	 * @return the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	MessageId send(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
-			MessageRouter messageRouter) throws PulsarClientException;
+	MessageId send(String topic, T message) throws PulsarClientException;
 
 	/**
 	 * Sends a message to the default topic in a non-blocking manner.
 	 * @param message the message to send
-	 * @return a future that holds the id of the sent message
+	 * @return a future that holds the id assigned by the broker to the published message
 	 * @throws PulsarClientException if an error occurs
 	 */
-	default CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
-		return sendAsync(null, message);
-	}
+	CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException;
 
 	/**
 	 * Sends a message to the specified topic in a non-blocking manner.
 	 * @param topic the topic to send the message to or {@code null} to send to the
 	 * default topic
 	 * @param message the message to send
-	 * @return a future that holds the id of the sent message
+	 * @return a future that holds the id assigned by the broker to the published message
 	 * @throws PulsarClientException if an error occurs
 	 */
-	default CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException {
-		return sendAsync(topic, message, (MessageRouter) null);
-	}
+	CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException;
 
 	/**
-	 * Sends a message to the default topic in a non-blocking manner.
-	 * @param message the message to send
-	 * @param messageRouter the optional message router to use
-	 * @return a future that holds the id of the sent message
-	 * @throws PulsarClientException if an error occurs
+	 * Create a {@link SendMessageBuilder builder} for configuring and sending a message.
+	 * @param message the payload of the message
+	 * @return the builder to configure and send the message
 	 */
-	default CompletableFuture<MessageId> sendAsync(T message, MessageRouter messageRouter)
-			throws PulsarClientException {
-		return sendAsync(null, message, messageRouter);
-	}
+	SendMessageBuilder<T> newMessage(T message);
 
 	/**
-	 * Sends a message to the default topic in a non-blocking manner.
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer
-	 * @return a future that holds the id of the sent message
-	 * @throws PulsarClientException if an error occurs
+	 * Builder that can be used to configure and send a message. Provides more options
+	 * than the basic send/sendAsync methods provided by {@link PulsarOperations}.
+	 *
+	 * @param <T> the message payload type
 	 */
-	default CompletableFuture<MessageId> sendAsync(T message,
-			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer) throws PulsarClientException {
-		return sendAsync(null, message, typedMessageBuilderCustomizer);
-	}
+	interface SendMessageBuilder<T> {
 
-	/**
-	 * Sends a message to the specified topic in a non-blocking manner.
-	 * @param topic the topic to send the message to or {@code null} to send to the
-	 * default topic
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer
-	 * @return a future that holds the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default CompletableFuture<MessageId> sendAsync(String topic, T message,
-			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer) throws PulsarClientException {
-		return sendAsync(topic, message, typedMessageBuilderCustomizer, null);
-	}
+		/**
+		 * Specify the topic to send the message to.
+		 * @param topic the destination topic
+		 * @return the current builder with the destination topic specified
+		 */
+		SendMessageBuilder<T> withTopic(String topic);
 
-	/**
-	 * Sends a message to the default in a non-blocking manner.
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer
-	 * @param messageRouter the optional message router to use
-	 * @return a future that holds the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default CompletableFuture<MessageId> sendAsync(T message,
-			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter)
-			throws PulsarClientException {
-		return sendAsync(null, message, typedMessageBuilderCustomizer, messageRouter);
-	}
+		/**
+		 * Specifies the message customizer to use to further configure the message.
+		 * @param messageCustomizer the message customizer
+		 * @return the current builder with the message customizer specified
+		 */
+		SendMessageBuilder<T> withMessageCustomizer(TypedMessageBuilderCustomizer<T> messageCustomizer);
 
-	/**
-	 * Sends a message to the specified topic in a non-blocking manner.
-	 * @param topic the topic to send the message to or {@code null} to send to the
-	 * default topic
-	 * @param message the message to send
-	 * @param messageRouter the optional message router to use
-	 * @return a future that holds the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	default CompletableFuture<MessageId> sendAsync(String topic, T message, MessageRouter messageRouter)
-			throws PulsarClientException {
-		return sendAsync(topic, message, null, messageRouter);
-	}
+		/**
+		 * Specifies the custom message router to use when sending the message.
+		 * @param messageRouter the custom message router
+		 * @return the current builder with the custom message router specified
+		 */
+		SendMessageBuilder<T> withCustomRouter(MessageRouter messageRouter);
 
-	/**
-	 * Sends a message to the specified topic in a non-blocking manner.
-	 * @param topic the topic to send the message to or {@code null} to send to the
-	 * default topic
-	 * @param message the message to send
-	 * @param typedMessageBuilderCustomizer the optional TypedMessageBuilder customizer
-	 * @param messageRouter the optional message router to use
-	 * @return a future that holds the id of the sent message
-	 * @throws PulsarClientException if an error occurs
-	 */
-	CompletableFuture<MessageId> sendAsync(String topic, T message,
-			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter)
-			throws PulsarClientException;
+		/**
+		 * Send the message in a blocking manner using the configured specification.
+		 * @return the id assigned by the broker to the published message
+		 * @throws PulsarClientException if an error occurs
+		 */
+		MessageId send() throws PulsarClientException;
+
+		/**
+		 * Uses the configured specification to send the message in a non-blocking manner.
+		 * @return a future that holds the id assigned by the broker to the published
+		 * message
+		 * @throws PulsarClientException if an error occurs
+		 */
+		CompletableFuture<MessageId> sendAsync() throws PulsarClientException;
+
+	}
 
 }

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

@@ -49,19 +49,17 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 	private Schema<T> schema;
 
 	/**
-	 * Constructs a template instance.
-	 * @param producerFactory the producer factory used to create the backing Pulsar
-	 * producers.
+	 * Construct a template instance.
+	 * @param producerFactory the factory used to create the backing Pulsar producers.
 	 */
 	public PulsarTemplate(PulsarProducerFactory<T> producerFactory) {
 		this(producerFactory, null);
 	}
 
 	/**
-	 * Constructs a template instance.
-	 * @param producerFactory the producer factory used to create the backing Pulsar
-	 * producers.
-	 * @param interceptors the {@link ProducerInterceptor}s to add to the producer.
+	 * Construct a template instance.
+	 * @param producerFactory the factory used to create the backing Pulsar producers.
+	 * @param interceptors the interceptors to add to the producer.
 	 */
 	public PulsarTemplate(PulsarProducerFactory<T> producerFactory, List<ProducerInterceptor> interceptors) {
 		this.producerFactory = producerFactory;
@@ -69,18 +67,49 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 	}
 
 	@Override
-	public MessageId send(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
+	public MessageId send(T message) throws PulsarClientException {
+		return doSend(null, message, null, null);
+	}
+
+	@Override
+	public MessageId send(String topic, T message) throws PulsarClientException {
+		return doSend(topic, message, null, null);
+	}
+
+	@Override
+	public CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
+		return doSendAsync(null, message, null, null);
+	}
+
+	@Override
+	public CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException {
+		return doSendAsync(topic, message, null, null);
+	}
+
+	@Override
+	public SendMessageBuilder<T> newMessage(T message) {
+		return new SendMessageBuilderImpl<>(this, message);
+	}
+
+	/**
+	 * Setter for schema.
+	 * @param schema provides the {@link Schema} used on this template
+	 */
+	public void setSchema(Schema<T> schema) {
+		this.schema = schema;
+	}
+
+	private MessageId doSend(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
 			MessageRouter messageRouter) throws PulsarClientException {
 		try {
-			return this.sendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter).get();
+			return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter).get();
 		}
 		catch (Exception ex) {
 			throw PulsarClientException.unwrap(ex);
 		}
 	}
 
-	@Override
-	public CompletableFuture<MessageId> sendAsync(String topic, T message,
+	private CompletableFuture<MessageId> doSendAsync(String topic, T message,
 			TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter)
 			throws PulsarClientException {
 		final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory);
@@ -109,12 +138,51 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
 		return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors);
 	}
 
-	/**
-	 * Setter for schema.
-	 * @param schema provides the {@link Schema} used on this template
-	 */
-	public void setSchema(Schema<T> schema) {
-		this.schema = schema;
+	public static class SendMessageBuilderImpl<T> implements SendMessageBuilder<T> {
+
+		private final PulsarTemplate<T> template;
+
+		private final T message;
+
+		private String topic;
+
+		private TypedMessageBuilderCustomizer<T> messageCustomizer;
+
+		private MessageRouter messageRouter;
+
+		SendMessageBuilderImpl(PulsarTemplate<T> template, T message) {
+			this.template = template;
+			this.message = message;
+		}
+
+		@Override
+		public SendMessageBuilder<T> withTopic(String topic) {
+			this.topic = topic;
+			return this;
+		}
+
+		@Override
+		public SendMessageBuilder<T> withMessageCustomizer(TypedMessageBuilderCustomizer<T> messageCustomizer) {
+			this.messageCustomizer = messageCustomizer;
+			return this;
+		}
+
+		@Override
+		public SendMessageBuilder<T> withCustomRouter(MessageRouter messageRouter) {
+			this.messageRouter = messageRouter;
+			return this;
+		}
+
+		@Override
+		public MessageId send() throws PulsarClientException {
+			return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter);
+		}
+
+		@Override
+		public CompletableFuture<MessageId> sendAsync() throws PulsarClientException {
+			return this.template.doSendAsync(this.topic, this.message, this.messageCustomizer, this.messageRouter);
+		}
+
 	}
 
 }

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

@@ -82,9 +82,10 @@ class FailoverConsumerTests extends AbstractContainerBaseTests {
 				pulsarClient, prodConfig);
 		final PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
 
-		pulsarTemplate.sendAsync("hello john doe", new FooRouter());
-		pulsarTemplate.sendAsync("hello alice doe", new BarRouter());
-		pulsarTemplate.sendAsync("hello buzz doe", new BuzzRouter());
+		pulsarTemplate.newMessage("hello john doe").withCustomRouter(new FooRouter()).sendAsync();
+		pulsarTemplate.newMessage("hello alice doe").withCustomRouter(new BarRouter()).sendAsync();
+		pulsarTemplate.newMessage("hello buzz doe").withCustomRouter(new BuzzRouter()).sendAsync();
+
 		final boolean await = latch.await(10, TimeUnit.SECONDS);
 		assertThat(await).isTrue();
 	}

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

@@ -140,76 +140,98 @@ class PulsarTemplateTests extends AbstractContainerBaseTests {
 
 	private static Stream<Arguments> sendMessageTestProvider() {
 		return Stream.of(
-
-				arguments(Named.of("sendMessageToDefaultTopic", "smt-topic-1"),
-						Collections.singletonMap("topicName", "smt-topic-1"),
+				arguments("sendMessageToDefaultTopic",
+						Collections.singletonMap("topicName", "sendMessageToDefaultTopic"),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.send(),
+						null, null),
+				arguments("sendMessageToDefaultTopicWithSimpleApi",
+						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithSimpleApi"),
 						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(msg), null,
 						null),
-				arguments(Named.of("sendMessageToDefaultTopicWithRouter", "smt-topic-2"),
-						Collections.singletonMap("topicName", "smt-topic-2"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(msg,
-								router),
+				arguments("sendMessageToDefaultTopicWithRouter",
+						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithRouter"),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.withCustomRouter(router).send(),
 						null, mockRouter()),
-				arguments(Named.of("sendMessageToDefaultTopicWithCustomizer", "smt-topic-3"),
-						Collections.singletonMap("topicName", "smt-topic-3"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(msg,
-								customizer),
+				arguments("sendMessageToDefaultTopicWithCustomizer",
+						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizer"),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.withMessageCustomizer(customizer).send(),
 						sampleMessageKeyCustomizer, null),
-				arguments(Named.of("sendMessageToDefaultTopicWithCustomizerAndRouter", "smt-topic-4"),
-						Collections.singletonMap("topicName", "smt-topic-4"),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(msg,
-								customizer, router),
+				arguments("sendMessageToDefaultTopicWithCustomizerAndRouter",
+						Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizerAndRouter"),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.withMessageCustomizer(customizer).withCustomRouter(router).send(),
 						sampleMessageKeyCustomizer, mockRouter()),
-				arguments(Named.of("sendMessageToSpecificTopic", "smt-topic-5"), Collections.emptyMap(),
+				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(Named.of("sendMessageToSpecificTopicWithRouter", "smt-topic-6"), Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(topic, msg,
-								null, router),
+				arguments("sendMessageToSpecificTopicWithRouter", Collections.emptyMap(),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.withTopic(topic).withCustomRouter(router).send(),
 						null, mockRouter()),
-				arguments(Named.of("sendMessageToSpecificTopicWithCustomizer", "smt-topic-7"), Collections.emptyMap(),
-						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.send(topic, msg,
-								customizer),
+				arguments("sendMessageToSpecificTopicWithCustomizer", Collections.emptyMap(),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.withMessageCustomizer(customizer).withTopic(topic).send(),
 						sampleMessageKeyCustomizer, null),
-				arguments(Named.of("sendMessageToSpecificTopicWithCustomizerAndRouter", "smt-topic-8"),
-						Collections.emptyMap(), (SendHandler<MessageId>) PulsarTemplate::send,
+				arguments("sendMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(),
+						(SendHandler<MessageId>) (template, topic, msg, customizer, router) -> template.newMessage(msg)
+								.withMessageCustomizer(customizer).withTopic(topic).withCustomRouter(router).send(),
 						sampleMessageKeyCustomizer, mockRouter()),
-				arguments(Named.of("sendAsyncMessageToDefaultTopic", "smt-topic-9"),
-						Collections.singletonMap("topicName", "smt-topic-9"),
+				arguments("sendAsyncMessageToDefaultTopic",
+						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopic"),
+						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
+								router) -> template.newMessage(msg).sendAsync(),
+						null, null),
+				arguments("sendAsyncMessageToDefaultTopicWithSimpleApi",
+						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithSimpleApi"),
 						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
 								router) -> template.sendAsync(msg),
 						null, null),
-				arguments(Named.of("sendAsyncMessageToDefaultTopicWithRouter", "smt-topic-10"),
-						Collections.singletonMap("topicName", "smt-topic-10"),
+				arguments("sendAsyncMessageToDefaultTopicWithRouter",
+						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithRouter"),
 						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(msg, router),
+								router) -> template.newMessage(msg).withCustomRouter(router).sendAsync(),
 						null, mockRouter()),
-				arguments(Named.of("sendAsyncMessageToDefaultTopicWithCustomizer", "smt-topic-11"),
-						Collections.singletonMap("topicName", "smt-topic-11"),
+				arguments("sendAsyncMessageToDefaultTopicWithCustomizer",
+						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizer"),
 						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(msg, customizer),
+								router) -> template.newMessage(msg).withMessageCustomizer(customizer).sendAsync(),
 						sampleMessageKeyCustomizer, null),
-				arguments(Named.of("sendAsyncMessageToDefaultTopicWithCustomizerAndRouter", "smt-topic-12"),
-						Collections.singletonMap("topicName", "smt-topic-12"),
+				arguments("sendAsyncMessageToDefaultTopicWithCustomizerAndRouter",
+						Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizerAndRouter"),
 						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(msg, customizer, router),
+								router) -> template.newMessage(msg).withMessageCustomizer(customizer)
+										.withCustomRouter(router).sendAsync(),
 						sampleMessageKeyCustomizer, mockRouter()),
-				arguments(Named.of("sendAsyncMessageToSpecificTopic", "smt-topic-13"), Collections.emptyMap(),
+				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(Named.of("sendAsyncMessageToSpecificTopicWithRouter", "smt-topic-14"), Collections.emptyMap(),
+				arguments("sendAsyncMessageToSpecificTopicWithRouter", Collections.emptyMap(),
 						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(topic, msg, null, router),
+								router) -> template.newMessage(msg).withTopic(topic).withCustomRouter(router)
+										.sendAsync(),
 						null, mockRouter()),
-				arguments(Named.of("sendAsyncMessageToSpecificTopicWithCustomizer", "smt-topic-15"),
-						Collections.emptyMap(),
+				arguments("sendAsyncMessageToSpecificTopicWithCustomizer", Collections.emptyMap(),
 						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
-								router) -> template.sendAsync(topic, msg, customizer),
+								router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic)
+										.sendAsync(),
 						sampleMessageKeyCustomizer, null),
-				arguments(Named.of("sendAsyncMessageToSpecificTopicWithCustomizerAndRouter", "smt-topic-16"),
-						Collections.emptyMap(), (SendHandler<CompletableFuture<MessageId>>) PulsarTemplate::sendAsync,
+				arguments("sendAsyncMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(),
+						(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, customizer,
+								router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic)
+										.withCustomRouter(router).sendAsync(),
 						sampleMessageKeyCustomizer, mockRouter()));
 	}