Allow to configure TypedMessageBuilder through a Map conf object

pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java

@@ -18,15 +18,23 @@
  */
 package org.apache.pulsar.client.api;
 
+import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.fail;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+
 import java.time.Clock;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+
+import lombok.Cleanup;
+
 import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy;
 import org.apache.pulsar.broker.service.schema.SchemaRegistry;
 import org.apache.pulsar.client.api.schema.GenericRecord;
@@ -623,4 +631,76 @@ public void testAutoBytesProducer() throws Exception {
         log.info("-- Exiting {} test --", methodName);
 
     }
+
+    @Test
+    public void testMessageBuilderLoadConf() throws Exception {
+        String topic = "my-topic-" + System.nanoTime();
+
+        @Cleanup
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topic)
+                .subscriptionName("my-subscriber-name")
+                .subscribe();
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic)
+                .create();
+
+        Map<String, String> properties = new HashMap<>();
+        properties.put("a", "1");
+        properties.put("b", "2");
+
+        Map<String, Object> msgConf = new HashMap<>();
+        msgConf.put("key", "key-1");
+        msgConf.put("properties", properties);
+        msgConf.put("eventTime", 1234L);
+        msgConf.put("sequenceId", 5L);
+        msgConf.put("replicationClusters", Lists.newArrayList("a", "b", "c"));
+        msgConf.put("disableReplication", false);
+
+        producer.newMessage()
+            .value("my-message")
+            .loadConf(msgConf)
+            .send();
+
+
+        Message<String> msg = consumer.receive();
+        assertEquals(msg.getKey(), "key-1");
+        assertEquals(msg.getProperties().get("a"), "1");
+        assertEquals(msg.getProperties().get("b"), "2");
+        assertEquals(msg.getEventTime(), 1234);
+        assertEquals(msg.getSequenceId(), 5);
+
+        consumer.acknowledge(msg);
+
+        // Try with invalid confs
+        msgConf.clear();
+        msgConf.put("nonExistingKey", "key-1");
+
+        try {
+            producer.newMessage()
+                    .value("my-message")
+                    .loadConf(msgConf)
+                    .send();
+            fail("Should have failed");
+        } catch (RuntimeException e) {
+            // expected
+        }
+
+        // Try with invalid type
+        msgConf.clear();
+        msgConf.put("eventTime", "hello");
+
+        try {
+            producer.newMessage()
+                    .value("my-message")
+                    .loadConf(msgConf)
+                    .send();
+            fail("Should have failed");
+        } catch (RuntimeException e) {
+            // expected
+        }
+    }
+
 }

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

@@ -174,4 +174,46 @@ public interface TypedMessageBuilder<T> extends Serializable {
      * @return the message builder instance
      */
     TypedMessageBuilder<T> disableReplication();
+
+    /**
+     * Configure the {@link TypedMessageBuilder} from a config map, as an alternative compared
+     * to call the individual builder methods.
+     * <p>
+     * The "value" of the message itself cannot be set on the config map.
+     * <p>
+     * Example:
+     *
+     * <pre>{@code
+     * Map<String, Object> conf = new HashMap<>();
+     * conf.put("key", "my-key");
+     * conf.put("eventTime", System.currentTimeMillis());
+     *
+     * producer.newMessage()
+     *             .value("my-message")
+     *             .loadConf(conf)
+     *             .send();
+     * }</pre>
+     *
+     * The available options are:
+     * <table border="1">
+     *  <tr><th>Constant</th><th>Name</th><th>Type</th><th>Doc</th></tr>
+     *  <tr><td>{@link #CONF_KEY}</td><td>{@code key}</td><td>{@code String}</td><td>{@link #key(String)}</td></tr>
+     *  <tr><td>{@link #CONF_PROPERTIES}</td><td>{@code properties}</td><td>{@code Map<String,String>}</td><td>{@link #properties(Map)}</td></tr>
+     *  <tr><td>{@link #CONF_EVENT_TIME}</td><td>{@code eventTime}</td><td>{@code long}</td><td>{@link #eventTime(long)}</td></tr>
+     *  <tr><td>{@link #CONF_SEQUENCE_ID}</td><td>{@code sequenceId}</td><td>{@code long}</td><td>{@link #sequenceId(long)}</td></tr>
+     *  <tr><td>{@link #CONF_REPLICATION_CLUSTERS}</td><td>{@code replicationClusters}</td><td>{@code List<String>}</td><td>{@link #replicationClusters(List)}</td></tr>
+     *  <tr><td>{@link #CONF_DISABLE_REPLICATION}</td><td>{@code disableReplication}</td><td>{@code boolean}</td><td>{@link #disableReplication()}</td></tr>
+     * </table>
+     *
+     * @param config a map with the configuration options for the message
+     * @return the message builder instance
+     */
+    TypedMessageBuilder<T> loadConf(Map<String, Object> config);
+
+    static final String CONF_KEY = "key";
+    static final String CONF_PROPERTIES = "properties";
+    static final String CONF_EVENT_TIME = "eventTime";
+    static final String CONF_SEQUENCE_ID = "sequenceId";
+    static final String CONF_REPLICATION_CLUSTERS = "replicationClusters";
+    static final String CONF_DISABLE_REPLICATION = "disableReplication";
 }

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

@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.impl;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.pulsar.client.util.TypeCheckUtil.checkType;
 
 import com.google.common.base.Preconditions;
 
@@ -130,6 +131,32 @@ public TypedMessageBuilder<T> disableReplication() {
         return this;
     }
 
+    @SuppressWarnings("unchecked")
+    @Override
+    public TypedMessageBuilder<T> loadConf(Map<String, Object> config) {
+        config.forEach((key, value) -> {
+            if (key.equals(CONF_KEY)) {
+                this.key(checkType(value, String.class));
+            } else if (key.equals(CONF_PROPERTIES)) {
+                this.properties(checkType(value, Map.class));
+            } else if (key.equals(CONF_EVENT_TIME)) {
+                this.eventTime(checkType(value, Long.class));
+            } else if (key.equals(CONF_SEQUENCE_ID)) {
+                this.sequenceId(checkType(value, Long.class));
+            } else if (key.equals(CONF_REPLICATION_CLUSTERS)) {
+                this.replicationClusters(checkType(value, List.class));
+            } else if (key.equals(CONF_DISABLE_REPLICATION)) {
+                boolean disableReplication = checkType(value, Boolean.class);
+                if (disableReplication) {
+                    this.disableReplication();
+                }
+            } else {
+                throw new RuntimeException("Invalid message config key '" + key + "'");
+            }
+        });
+        return this;
+    }
+
     public MessageMetadata.Builder getMetadataBuilder() {
         return msgMetadataBuilder;
     }

pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java

@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   http://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.apache.pulsar.client.util;
+
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class TypeCheckUtil {
+    @SuppressWarnings("unchecked")
+    public static <T> T checkType(Object o, Class<T> clazz) {
+        if (!clazz.isInstance(o)) {
+            throw new RuntimeException(
+                    String.format("Invalid object type '%s' when exepcting '%s'", o.getClass().getName(), clazz.getName()));
+        }
+        return (T) o;
+    }
+}