[feat][client] Support configure MessageCrypto in ProducerBuilder (#19939)

Signed-off-by: tison <wander4096@gmail.com>

Author: tisonkun

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

@@ -389,6 +389,17 @@ public interface ProducerBuilder<T> extends Cloneable {
      */
     ProducerBuilder<T> defaultCryptoKeyReader(Map<String, String> publicKeys);
 
+    /**
+     * Sets a {@link MessageCrypto}.
+     *
+     * <p>Contains methods to encrypt/decrypt messages for end-to-end encryption.
+     *
+     * @param messageCrypto
+     *            MessageCrypto object
+     * @return the producer builder instance
+     */
+    ProducerBuilder<T> messageCrypto(MessageCrypto messageCrypto);
+
     /**
      * Add public encryption key, used by producer to encrypt the data key.
      *

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

@@ -34,6 +34,7 @@
 import org.apache.pulsar.client.api.CompressionType;
 import org.apache.pulsar.client.api.CryptoKeyReader;
 import org.apache.pulsar.client.api.HashingScheme;
+import org.apache.pulsar.client.api.MessageCrypto;
 import org.apache.pulsar.client.api.MessageRouter;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
@@ -221,6 +222,12 @@ public ProducerBuilder<T> defaultCryptoKeyReader(@NonNull Map<String, String> pu
         return cryptoKeyReader(DefaultCryptoKeyReader.builder().publicKeys(publicKeys).build());
     }
 
+    @Override
+    public ProducerBuilder<T> messageCrypto(MessageCrypto messageCrypto) {
+        conf.setMessageCrypto(messageCrypto);
+        return this;
+    }
+
     @Override
     public ProducerBuilder<T> addEncryptionKey(String key) {
         checkArgument(StringUtils.isNotBlank(key), "Encryption key cannot be blank");

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerBuilderImplTest.java

@@ -18,6 +18,15 @@
  */
 package org.apache.pulsar.client.impl;
 
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNotNull;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageRouter;
 import org.apache.pulsar.client.api.MessageRoutingMode;
@@ -26,34 +35,24 @@
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.TopicMetadata;
 import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertNotNull;
-
 /**
  * Unit tests of {@link ProducerBuilderImpl}.
  */
 public class ProducerBuilderImplTest {
 
     private static final String TOPIC_NAME = "testTopicName";
     private PulsarClientImpl client;
-    private ProducerBuilderImpl producerBuilderImpl;
+    private ProducerBuilderImpl<byte[]> producerBuilderImpl;
 
     @BeforeClass(alwaysRun = true)
     public void setup() {
-        Producer producer = mock(Producer.class);
+        Producer<?> producer = mock(Producer.class);
         client = mock(PulsarClientImpl.class);
-        producerBuilderImpl = new ProducerBuilderImpl(client, Schema.BYTES);
+        producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES);
         when(client.newProducer()).thenReturn(producerBuilderImpl);
 
         when(client.createProducerAsync(
@@ -66,8 +65,8 @@ public void testProducerBuilderImpl() throws PulsarClientException {
         Map<String, String> properties = new HashMap<>();
         properties.put("Test-Key2", "Test-Value2");
 
-        producerBuilderImpl = new ProducerBuilderImpl(client, Schema.BYTES);
-        Producer producer = producerBuilderImpl.topic(TOPIC_NAME)
+        producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES);
+        Producer<?> producer = producerBuilderImpl.topic(TOPIC_NAME)
                 .producerName("Test-Producer")
                 .maxPendingMessages(2)
                 .addEncryptionKey("Test-EncryptionKey")
@@ -78,6 +77,14 @@ public void testProducerBuilderImpl() throws PulsarClientException {
         assertNotNull(producer);
     }
 
+    @Test
+    public void testProducerBuilderImplWhenMessageCryptoSet() throws PulsarClientException {
+        producerBuilderImpl = new ProducerBuilderImpl<>(client, Schema.BYTES);
+        producerBuilderImpl.topic(TOPIC_NAME).messageCrypto(new MessageCryptoBc("ctx1", true));
+        assertNotNull(producerBuilderImpl.create());
+        assertNotNull(producerBuilderImpl.getConf().getMessageCrypto());
+    }
+
     @Test
     public void testProducerBuilderImplWhenMessageRoutingModeAndMessageRouterAreNotSet() throws PulsarClientException {
         producerBuilderImpl = new ProducerBuilderImpl(client, Schema.BYTES);