[improve][client] Add newMessage with schema and transactions

Pulsar Client allows callers to create messages with a schema or a transaction, but
not both. This commit adds a new method in the producer that allows callers to create
a message with both a schema and transaction.

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java

@@ -132,6 +132,21 @@ public interface Producer<T> extends Closeable {
      * @since 2.7.0
      */
     TypedMessageBuilder<T> newMessage(Transaction txn);
+
+    /**
+     * Create a new message builder with transaction and schema, not required same parameterized type with the
+     * producer.
+     *
+     * <p>After the transaction commit, it will be made visible to consumer.
+     *
+     * <p>After the transaction abort, it will never be visible to consumer.
+     *
+     * @return a typed message builder that can be used to construct the message to be sent through this producer
+     * @see #newMessage()
+     */
+    <V> TypedMessageBuilder<V> newMessage(Schema<V> schema,
+                                          Transaction txn);
+
     /**
      * Get the last sequence id that was published by this producer.
      *

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java

@@ -79,6 +79,7 @@ public TypedMessageBuilder<T> newMessage() {
         return new TypedMessageBuilderImpl<>(this, schema);
     }
 
+    @Override
     public <V> TypedMessageBuilder<V> newMessage(Schema<V> schema) {
         checkArgument(schema != null);
         return new TypedMessageBuilderImpl<>(this, schema);
@@ -92,6 +93,14 @@ public TypedMessageBuilder<T> newMessage(Transaction txn) {
         return new TypedMessageBuilderImpl<>(this, schema, (TransactionImpl) txn);
     }
 
+    @Override
+    public <V> TypedMessageBuilder<V> newMessage(Schema<V> schema,
+                                                 Transaction txn) {
+        checkArgument(txn instanceof TransactionImpl);
+        checkArgument(schema != null);
+        return new TypedMessageBuilderImpl<>(this, schema, (TransactionImpl) txn);
+    }
+
     abstract CompletableFuture<MessageId> internalSendAsync(Message<?> message);
 
     abstract CompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message, Transaction txn);