Refactored ClientConfuguration to use ClientConfigurationData shared …

…with ClientBuilderImpl (#1276)

* Refactored ClientConfuguration to use ClientConfigurationData shared with ClientBuilderImpl

* Fixed unit tests

* Fixed cloning issue after refactoring

* Fixed another test

* Fixed cloning issues

* Fixes for mock tests

* Fixed refactoring problem in TopicsConsumerImpl

pom.xml

@@ -150,7 +150,7 @@ flexible messaging model and an intuitive client API.</description>
           </exclusion>
         </exclusions>
       </dependency>
-      
+
       <dependency>
         <groupId>org.testng</groupId>
         <artifactId>testng</artifactId>
@@ -405,7 +405,7 @@ flexible messaging model and an intuitive client API.</description>
         <artifactId>jersey-container-servlet</artifactId>
         <version>2.23.2</version>
       </dependency>
-      
+
       <dependency>
         <groupId>javax.ws.rs</groupId>
         <artifactId>javax.ws.rs-api</artifactId>
@@ -629,6 +629,13 @@ flexible messaging model and an intuitive client API.</description>
       <artifactId>powermock-module-testng</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <version>1.16.20</version>
+      <scope>provided</scope>
+    </dependency>
   </dependencies>
 
   <build>

pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawMessageImpl.java

@@ -22,7 +22,6 @@
 
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.RawMessage;
-import org.apache.pulsar.common.api.ByteBufPair;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
 import org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream;
 import org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream;

pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java

@@ -22,49 +22,39 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
-import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
-import org.apache.commons.codec.digest.DigestUtils;
+
 import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerConfiguration;
 import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.RawReader;
 import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.api.RawReader;
 import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.client.impl.PulsarClientImpl;
-import org.apache.pulsar.client.impl.ConsumerImpl.SubscriptionMode;
-import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
+import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
 import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
-import io.netty.buffer.ByteBuf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.netty.buffer.ByteBuf;
+
 public class RawReaderImpl implements RawReader {
 
     final static int DEFAULT_RECEIVER_QUEUE_SIZE = 1000;
-    private final PulsarClientImpl client;
-    private final String topic;
-    private final String subscription;
-    private final ConsumerConfiguration consumerConfiguration;
+    private final ConsumerConfigurationData consumerConfiguration;
     private RawConsumerImpl consumer;
 
     public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
                          CompletableFuture<Consumer> consumerFuture) {
-        this.client = client;
-        this.subscription = subscription;
-        this.topic = topic;
-
-        consumerConfiguration = new ConsumerConfiguration();
+        consumerConfiguration = new ConsumerConfigurationData();
+        consumerConfiguration.getTopicNames().add(topic);
+        consumerConfiguration.setSubscriptionName(subscription);
         consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
         consumerConfiguration.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
 
-        consumer = new RawConsumerImpl(client, topic, subscription, consumerConfiguration,
+        consumer = new RawConsumerImpl(client, consumerConfiguration,
                                        consumerFuture);
     }
 
@@ -92,11 +82,10 @@ static class RawConsumerImpl extends ConsumerImpl {
         final BlockingQueue<RawMessageAndCnx> incomingRawMessages;
         final Queue<CompletableFuture<RawMessage>> pendingRawReceives;
 
-        RawConsumerImpl(PulsarClientImpl client, String topic, String subscription, ConsumerConfiguration conf,
-                        CompletableFuture<Consumer> consumerFuture) {
-            super(client, topic, subscription, conf,
-                  client.externalExecutorProvider().getExecutor(), -1, consumerFuture,
-                  SubscriptionMode.Durable, MessageId.earliest);
+        RawConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData conf,
+                CompletableFuture<Consumer> consumerFuture) {
+            super(client, conf.getSingleTopic(), conf, client.externalExecutorProvider().getExecutor(), -1,
+                    consumerFuture, SubscriptionMode.Durable, MessageId.earliest);
             incomingRawMessages = new GrowableArrayBlockingQueue<>();
             pendingRawReceives = new ConcurrentLinkedQueue<>();
         }

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java

@@ -88,6 +88,7 @@
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
@@ -1151,17 +1152,18 @@ public void testClosingReplicationProducerTwice() throws Exception {
         brokerService.getReplicationClients().put(remoteCluster, client);
         PersistentReplicator replicator = new PersistentReplicator(topic, cursor, localCluster, remoteCluster, brokerService);
 
-        doReturn(new CompletableFuture<Producer>()).when(clientImpl).createProducerAsync(matches(globalTopicName), any());
+        doReturn(new CompletableFuture<Producer>()).when(clientImpl)
+                .createProducerAsync(any(ProducerConfigurationData.class));
 
         replicator.startProducer();
-        verify(clientImpl).createProducerAsync(matches(globalTopicName), any());
+        verify(clientImpl).createProducerAsync(any(ProducerConfigurationData.class));
 
         replicator.disconnect(false);
         replicator.disconnect(false);
 
         replicator.startProducer();
 
-        verify(clientImpl, Mockito.times(2)).createProducerAsync(matches(globalTopicName), any());
+        verify(clientImpl, Mockito.times(2)).createProducerAsync(any(ProducerConfigurationData.class));
     }
 
     @Test

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java

@@ -61,6 +61,7 @@
 import org.apache.pulsar.client.api.RawReader;
 import org.apache.pulsar.client.impl.ProducerImpl;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
 import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.common.naming.DestinationName;
 import org.apache.pulsar.common.naming.NamespaceBundle;
@@ -242,7 +243,7 @@ public void testConcurrentReplicator() throws Exception {
         }
         Thread.sleep(3000);
 
-        Mockito.verify(pulsarClient, Mockito.times(1)).createProducerAsync(Mockito.anyString(), Mockito.anyObject());
+        Mockito.verify(pulsarClient, Mockito.times(1)).createProducerAsync(Mockito.any(ProducerConfigurationData.class));
 
     }
 

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConnectionPoolTest.java

@@ -23,7 +23,7 @@
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
-import org.apache.pulsar.client.api.ClientConfiguration;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.common.util.netty.EventLoopUtil;
 import org.mockito.Mockito;
 import org.testng.annotations.AfterClass;
@@ -53,10 +53,11 @@ protected void cleanup() throws Exception {
 
     @Test
     public void testSingleIpAddress() throws Exception {
-        ClientConfiguration conf = new ClientConfiguration();
+        ClientConfigurationData conf = new ClientConfigurationData();
         EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, new DefaultThreadFactory("test"));
         ConnectionPool pool = Mockito.spy(new ConnectionPool(conf, eventLoop));
-        PulsarClientImpl client = new PulsarClientImpl(serviceUrl, conf, eventLoop, pool);
+        conf.setServiceUrl(serviceUrl);
+        PulsarClientImpl client = new PulsarClientImpl(conf, eventLoop, pool);
 
         List<InetAddress> result = Lists.newArrayList();
         result.add(InetAddress.getByName("127.0.0.1"));
@@ -71,10 +72,11 @@ public void testSingleIpAddress() throws Exception {
     public void testDoubleIpAddress() throws Exception {
         String serviceUrl = "pulsar://non-existing-dns-name:" + BROKER_PORT;
 
-        ClientConfiguration conf = new ClientConfiguration();
+        ClientConfigurationData conf = new ClientConfigurationData();
         EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, new DefaultThreadFactory("test"));
         ConnectionPool pool = Mockito.spy(new ConnectionPool(conf, eventLoop));
-        PulsarClientImpl client = new PulsarClientImpl(serviceUrl, conf, eventLoop, pool);
+        conf.setServiceUrl(serviceUrl);
+        PulsarClientImpl client = new PulsarClientImpl(conf, eventLoop, pool);
 
         List<InetAddress> result = Lists.newArrayList();
 

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageIdTest.java

@@ -51,9 +51,10 @@
 import org.apache.pulsar.client.api.ProducerConfiguration;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.ProducerImpl.OpSendMsg;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.common.api.ByteBufPair;
 import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.common.api.Commands.ChecksumType;
-import org.apache.pulsar.common.api.ByteBufPair;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata.Builder;
 import org.slf4j.Logger;
@@ -295,7 +296,7 @@ public void testChecksumVersionComptability() throws Exception {
         ((PulsarClientImpl) pulsarClient).timer().stop();
 
         ClientCnx mockClientCnx = spy(
-                new ClientCnx(new ClientConfiguration(), ((PulsarClientImpl) pulsarClient).eventLoopGroup()));
+                new ClientCnx(new ClientConfigurationData(), ((PulsarClientImpl) pulsarClient).eventLoopGroup()));
         doReturn(producer.brokerChecksumSupportedVersion() - 1).when(mockClientCnx).getRemoteEndpointProtocolVersion();
         prod.setClientCnx(mockClientCnx);
 
@@ -360,7 +361,7 @@ public void testChecksumReconnection() throws Exception {
 
         // set clientCnx mock to get non-checksum supported version
         ClientCnx mockClientCnx = spy(
-                new ClientCnx(new ClientConfiguration(), ((PulsarClientImpl) pulsarClient).eventLoopGroup()));
+                new ClientCnx(new ClientConfigurationData(), ((PulsarClientImpl) pulsarClient).eventLoopGroup()));
         doReturn(producer.brokerChecksumSupportedVersion() - 1).when(mockClientCnx).getRemoteEndpointProtocolVersion();
         prod.setClientCnx(mockClientCnx);
 
@@ -489,7 +490,7 @@ public void testCorruptMessageRemove() throws Exception {
         MessageImpl msg1 = (MessageImpl) MessageBuilder.create().setContent("message-1".getBytes()).build();
         future = producer.sendAsync(msg1);
         ClientCnx cnx = spy(
-                new ClientCnx(new ClientConfiguration(), ((PulsarClientImpl) pulsarClient).eventLoopGroup()));
+                new ClientCnx(new ClientConfigurationData(), ((PulsarClientImpl) pulsarClient).eventLoopGroup()));
         String exc = "broker is already stopped";
         // when client-try to recover checksum by resending to broker: throw exception as broker is stopped
         doThrow(new IllegalStateException(exc)).when(cnx).ctx();