sijie closed pull request #2004: Issue 1943: remove serializable from builders
and add loadData to load configuration from a config map
URL: https://github.com/apache/incubator-pulsar/pull/2004
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
index 071e292932..e1fe3ecff6 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.client.api;
-import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -29,13 +28,33 @@
*
* @since 2.0.0
*/
-public interface ClientBuilder extends Serializable, Cloneable {
+public interface ClientBuilder extends Cloneable {
/**
* @return the new {@link PulsarClient} instance
*/
PulsarClient build() throws PulsarClientException;
+ /**
+ * Load the configuration from provided <tt>config</tt> map.
+ *
+ * <p>Example:
+ * <pre>
+ * Map<String, Object> config = new HashMap<>();
+ * config.put("serviceUrl", "pulsar://localhost:5550");
+ * config.put("numIoThreads", 20);
+ *
+ * ClientBuilder builder = ...;
+ * builder = builder.loadConf(config);
+ *
+ * PulsarClient client = builder.build();
+ * </pre>
+ *
+ * @param config configuration to load
+ * @return client builder instance
+ */
+ ClientBuilder loadConf(Map<String, Object> config);
+
/**
* Create a copy of the current client builder.
* <p>
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
index f0014a58f2..8657859103 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.client.api;
-import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -32,7 +31,7 @@
*
* @since 2.0.0
*/
-public interface ConsumerBuilder<T> extends Serializable, Cloneable {
+public interface ConsumerBuilder<T> extends Cloneable {
/**
* Create a copy of the current consumer builder.
@@ -52,6 +51,26 @@
*/
ConsumerBuilder<T> clone();
+ /**
+ * Load the configuration from provided <tt>config</tt> map.
+ *
+ * <p>Example:
+ * <pre>
+ * Map<String, Object> config = new HashMap<>();
+ * config.put("ackTimeoutMillis", 1000);
+ * config.put("receiverQueueSize", 2000);
+ *
+ * ConsumerBuilder<byte[]> builder = ...;
+ * builder = builder.loadConf(config);
+ *
+ * Consumer<byte[]> consumer = builder.subscribe();
+ * </pre>
+ *
+ * @param config configuration to load
+ * @return consumer builder instance
+ */
+ ConsumerBuilder<T> loadConf(Map<String, Object> config);
+
/**
* Finalize the {@link Consumer} creation by subscribing to the topic.
*
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerEventListener.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerEventListener.java
index e2e627414f..af0bd507da 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerEventListener.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerEventListener.java
@@ -18,10 +18,12 @@
*/
package org.apache.pulsar.client.api;
+import java.io.Serializable;
+
/**
* Listener on the consumer state changes.
*/
-public interface ConsumerEventListener {
+public interface ConsumerEventListener extends Serializable {
/**
* Notified when the consumer group is changed, and the consumer becomes
the active consumer.
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/CryptoKeyReader.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/CryptoKeyReader.java
index 4acb6e9cf8..274191f723 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/CryptoKeyReader.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/CryptoKeyReader.java
@@ -18,9 +18,10 @@
*/
package org.apache.pulsar.client.api;
+import java.io.Serializable;
import java.util.Map;
-public interface CryptoKeyReader {
+public interface CryptoKeyReader extends Serializable {
/**
* Return the encryption key corresponding to the key name in the argument
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageId.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageId.java
index f74c169dc8..1bb3e086ed 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageId.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageId.java
@@ -20,6 +20,7 @@
import java.io.IOException;
+import java.io.Serializable;
import org.apache.pulsar.client.impl.MessageIdImpl;
/**
@@ -30,7 +31,7 @@
*
*
*/
-public interface MessageId extends Comparable<MessageId>{
+public interface MessageId extends Comparable<MessageId>, Serializable {
/**
* Serialize the message ID into a byte array
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
index a9723a00b4..8256b4a282 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.client.api;
-import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -30,7 +29,7 @@
*
* @see PulsarClient#newProducer()
*/
-public interface ProducerBuilder<T> extends Serializable, Cloneable {
+public interface ProducerBuilder<T> extends Cloneable {
/**
* Finalize the creation of the {@link Producer} instance.
@@ -58,6 +57,26 @@
*/
CompletableFuture<Producer<T>> createAsync();
+ /**
+ * Load the configuration from provided <tt>config</tt> map.
+ *
+ * <p>Example:
+ * <pre>
+ * Map<String, Object> config = new HashMap<>();
+ * config.put("producerName", "test-producer");
+ * config.put("sendTimeoutMs", 2000);
+ *
+ * ProducerBuilder<byte[]> builder = ...;
+ * builder = builder.loadConf(config);
+ *
+ * Producer<byte[]> producer = builder.create();
+ * </pre>
+ *
+ * @param config configuration to load
+ * @return producer builder instance
+ */
+ ProducerBuilder<T> loadConf(Map<String, Object> config);
+
/**
* Create a copy of the current {@link ProducerBuilder}.
* <p>
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
index 98af0b3fd2..5827573a3c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
@@ -19,10 +19,13 @@
package org.apache.pulsar.client.api;
import java.io.Closeable;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.impl.ClientBuilderImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
/**
* Class that provides a client interface to Pulsar.
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
index 314af7b14a..bc162bd06b 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
@@ -18,7 +18,7 @@
*/
package org.apache.pulsar.client.api;
-import java.io.Serializable;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
/**
@@ -28,7 +28,7 @@
*
* @since 2.0.0
*/
-public interface ReaderBuilder<T> extends Serializable, Cloneable {
+public interface ReaderBuilder<T> extends Cloneable {
/**
* Finalize the creation of the {@link Reader} instance.
@@ -54,6 +54,26 @@
*/
CompletableFuture<Reader<T>> createAsync();
+ /**
+ * Load the configuration from provided <tt>config</tt> map.
+ *
+ * <p>Example:
+ * <pre>
+ * Map<String, Object> config = new HashMap<>();
+ * config.put("topicName", "test-topic");
+ * config.put("receiverQueueSize", 2000);
+ *
+ * ReaderBuilder<byte[]> builder = ...;
+ * builder = builder.loadConf(config);
+ *
+ * Reader<byte[]> reader = builder.create();
+ * </pre>
+ *
+ * @param config configuration to load
+ * @return reader builder instance
+ */
+ ReaderBuilder<T> loadConf(Map<String, Object> config);
+
/**
* Create a copy of the current {@link ReaderBuilder}.
* <p>
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
index ad4b5107b4..657efd3008 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java
@@ -28,18 +28,17 @@
import org.apache.pulsar.client.api.PulsarClientException;
import
org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
public class ClientBuilderImpl implements ClientBuilder {
- private static final long serialVersionUID = 1L;
-
- final ClientConfigurationData conf;
+ ClientConfigurationData conf;
public ClientBuilderImpl() {
this(new ClientConfigurationData());
}
- private ClientBuilderImpl(ClientConfigurationData conf) {
+ public ClientBuilderImpl(ClientConfigurationData conf) {
this.conf = conf;
}
@@ -57,6 +56,13 @@ public ClientBuilder clone() {
return new ClientBuilderImpl(conf.clone());
}
+ @Override
+ public ClientBuilder loadConf(Map<String, Object> config) {
+ conf = ConfigurationDataUtils.loadData(
+ config, conf, ClientConfigurationData.class);
+ return this;
+ }
+
@Override
public ClientBuilder serviceUrl(String serviceUrl) {
conf.setServiceUrl(serviceUrl);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
index af11ef5945..f0067f7f10 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -39,6 +39,7 @@
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.util.FutureUtil;
@@ -48,10 +49,8 @@
public class ConsumerBuilderImpl<T> implements ConsumerBuilder<T> {
- private static final long serialVersionUID = 1L;
-
private final PulsarClientImpl client;
- private final ConsumerConfigurationData<T> conf;
+ private ConsumerConfigurationData<T> conf;
private final Schema<T> schema;
private static long MIN_ACK_TIMEOUT_MILLIS = 1000;
@@ -60,12 +59,18 @@
this(client, new ConsumerConfigurationData<T>(), schema);
}
- private ConsumerBuilderImpl(PulsarClientImpl client,
ConsumerConfigurationData<T> conf, Schema<T> schema) {
+ ConsumerBuilderImpl(PulsarClientImpl client, ConsumerConfigurationData<T>
conf, Schema<T> schema) {
this.client = client;
this.conf = conf;
this.schema = schema;
}
+ @Override
+ public ConsumerBuilder<T> loadConf(Map<String, Object> config) {
+ this.conf = ConfigurationDataUtils.loadData(config, conf,
ConsumerConfigurationData.class);
+ return this;
+ }
+
@Override
public ConsumerBuilder<T> clone() {
return new ConsumerBuilderImpl<>(client, conf.clone(), schema);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
index 0f5eab013f..199665be2a 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
@@ -33,6 +33,7 @@
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.util.FutureUtil;
@@ -40,10 +41,8 @@
public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
- private static final long serialVersionUID = 1L;
-
private final PulsarClientImpl client;
- private final ProducerConfigurationData conf;
+ private ProducerConfigurationData conf;
private final Schema<T> schema;
ProducerBuilderImpl(PulsarClientImpl client, Schema<T> schema) {
@@ -88,6 +87,13 @@ private ProducerBuilderImpl(PulsarClientImpl client,
ProducerConfigurationData c
return client.createProducerAsync(conf, schema);
}
+ @Override
+ public ProducerBuilder<T> loadConf(Map<String, Object> config) {
+ conf = ConfigurationDataUtils.loadData(
+ config, conf, ProducerConfigurationData.class);
+ return this;
+ }
+
@Override
public ProducerBuilder<T> topic(String topicName) {
conf.setTopicName(topicName);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
index 9e8bce82aa..d74dc83b8c 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -29,16 +30,15 @@
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.ReaderListener;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
import org.apache.pulsar.common.util.FutureUtil;
public class ReaderBuilderImpl<T> implements ReaderBuilder<T> {
- private static final long serialVersionUID = 1L;
-
private final PulsarClientImpl client;
- private final ReaderConfigurationData<T> conf;
+ private ReaderConfigurationData<T> conf;
private final Schema<T> schema;
@@ -94,6 +94,12 @@ private ReaderBuilderImpl(PulsarClientImpl client,
ReaderConfigurationData<T> co
return client.createReaderAsync(conf, schema);
}
+ @Override
+ public ReaderBuilder<T> loadConf(Map<String, Object> config) {
+ conf = ConfigurationDataUtils.loadData(config, conf,
ReaderConfigurationData.class);
+ return this;
+ }
+
@Override
public ReaderBuilder<T> topic(String topicName) {
conf.setTopicName(topicName);
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
index cedc892cab..b51d61f515 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
@@ -63,4 +63,6 @@ public ClientConfigurationData clone() {
throw new RuntimeException("Failed to clone
ClientConfigurationData");
}
}
+
+
}
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java
new file mode 100644
index 0000000000..4523939e07
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java
@@ -0,0 +1,75 @@
+/**
+ * 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.impl.conf;
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Maps;
+import io.netty.util.concurrent.FastThreadLocal;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Utils for loading configuration data.
+ */
+public final class ConfigurationDataUtils {
+
+ public static ObjectMapper create() {
+ ObjectMapper mapper = new ObjectMapper();
+ // forward compatibility for the properties may go away in the future
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
true);
+
mapper.configure(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL,
false);
+ mapper.setSerializationInclusion(Include.NON_NULL);
+ return mapper;
+ }
+
+ private static final FastThreadLocal<ObjectMapper> mapper = new
FastThreadLocal<ObjectMapper>() {
+ @Override
+ protected ObjectMapper initialValue() throws Exception {
+ return create();
+ }
+ };
+
+ public static ObjectMapper getThreadLocal() {
+ return mapper.get();
+ }
+
+ private ConfigurationDataUtils() {}
+
+ public static <T> T loadData(Map<String, Object> config,
+ T existingData,
+ Class<T> dataCls) {
+ ObjectMapper mapper = getThreadLocal();
+ try {
+ String existingConfigJson =
mapper.writeValueAsString(existingData);
+ Map<String, Object> existingConfig =
mapper.readValue(existingConfigJson, Map.class);
+ Map<String, Object> newConfig = Maps.newHashMap();
+ newConfig.putAll(existingConfig);
+ newConfig.putAll(config);
+ String configJson = mapper.writeValueAsString(newConfig);
+ return mapper.readValue(configJson, dataCls);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to load config into existing
configuration data", e);
+ }
+
+ }
+
+}
+
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
index 9f068091b3..57ca135b68 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
@@ -80,6 +80,7 @@
* Returns true if encryption keys are added
*
*/
+ @JsonIgnore
public boolean isEncryptionEnabled() {
return (this.encryptionKeys != null) && !this.encryptionKeys.isEmpty()
&& (this.cryptoKeyReader != null);
}
diff --git
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java
new file mode 100644
index 0000000000..83ed4e883c
--- /dev/null
+++
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtilsTest.java
@@ -0,0 +1,111 @@
+/**
+ * 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.impl.conf;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+/**
+ * Unit test {@link ConfigurationDataUtils}.
+ */
+public class ConfigurationDataUtilsTest {
+
+ @Test
+ public void testLoadClientConfigurationData() {
+ ClientConfigurationData confData = new ClientConfigurationData();
+ confData.setServiceUrl("pulsar://unknown:6650");
+ confData.setMaxLookupRequest(600);
+ confData.setNumIoThreads(33);
+ Map<String, Object> config = new HashMap<>();
+ config.put("serviceUrl", "pulsar://localhost:6650");
+ config.put("maxLookupRequest", 70000);
+ confData = ConfigurationDataUtils.loadData(config, confData,
ClientConfigurationData.class);
+ assertEquals("pulsar://localhost:6650", confData.getServiceUrl());
+ assertEquals(70000, confData.getMaxLookupRequest());
+ assertEquals(33, confData.getNumIoThreads());
+ }
+
+ @Test
+ public void testLoadProducerConfigurationData() {
+ ProducerConfigurationData confData = new ProducerConfigurationData();
+ confData.setProducerName("unset");
+ confData.setBatchingEnabled(true);
+ confData.setBatchingMaxMessages(1234);
+ Map<String, Object> config = new HashMap<>();
+ config.put("producerName", "test-producer");
+ config.put("batchingEnabled", false);
+ confData = ConfigurationDataUtils.loadData(config, confData,
ProducerConfigurationData.class);
+ assertEquals("test-producer", confData.getProducerName());
+ assertEquals(false, confData.isBatchingEnabled());
+ assertEquals(1234, confData.getBatchingMaxMessages());
+ }
+
+ @Test
+ public void testLoadConsumerConfigurationData() {
+ ConsumerConfigurationData confData = new ConsumerConfigurationData();
+ confData.setSubscriptionName("unknown-subscription");
+ confData.setPriorityLevel(10000);
+ confData.setConsumerName("unknown-consumer");
+ Map<String, Object> config = new HashMap<>();
+ config.put("subscriptionName", "test-subscription");
+ config.put("priorityLevel", 100);
+ confData = ConfigurationDataUtils.loadData(config, confData,
ConsumerConfigurationData.class);
+ assertEquals("test-subscription", confData.getSubscriptionName());
+ assertEquals(100, confData.getPriorityLevel());
+ assertEquals("unknown-consumer", confData.getConsumerName());
+ }
+
+ @Test
+ public void testLoadReaderConfigurationData() {
+ ReaderConfigurationData confData = new ReaderConfigurationData();
+ confData.setTopicName("unknown");
+ confData.setReceiverQueueSize(1000000);
+ confData.setReaderName("unknown-reader");
+ Map<String, Object> config = new HashMap<>();
+ config.put("topicName", "test-topic");
+ config.put("receiverQueueSize", 100);
+ confData = ConfigurationDataUtils.loadData(config, confData,
ReaderConfigurationData.class);
+ assertEquals("test-topic", confData.getTopicName());
+ assertEquals(100, confData.getReceiverQueueSize());
+ assertEquals("unknown-reader", confData.getReaderName());
+ }
+
+ @Test
+ public void testLoadConfigurationDataWithUnknownFields() {
+ ReaderConfigurationData confData = new ReaderConfigurationData();
+ confData.setTopicName("unknown");
+ confData.setReceiverQueueSize(1000000);
+ confData.setReaderName("unknown-reader");
+ Map<String, Object> config = new HashMap<>();
+ config.put("unknown", "test-topic");
+ config.put("receiverQueueSize", 100);
+ try {
+ ConfigurationDataUtils.loadData(config, confData,
ReaderConfigurationData.class);
+ fail("Should fail loading configuration data with unknown fields");
+ } catch (RuntimeException re) {
+ assertTrue(re.getCause() instanceof IOException);
+ }
+ }
+}
diff --git
a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
index 81cbbda1d2..def0926f9e 100644
---
a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
+++
b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/producers/MultiConsumersOneOutputTopicProducersTest.java
@@ -88,6 +88,11 @@
}
}
+ @Override
+ public ProducerBuilder<byte[]> loadConf(Map<String, Object> config) {
+ return this;
+ }
+
@Override
public ProducerBuilder<byte[]> clone() {
return this;
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
