Create new client properties map on client parameters duplication

To avoid duplicates use the same map instance and so interfere.

Fixes #234

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -2028,11 +2028,11 @@ public int hashCode() {
 
   public static class ClientParameters {
 
-    private final Map<String, String> clientProperties = new HashMap<>();
+    private final Map<String, String> clientProperties = new ConcurrentHashMap<>();
     EventLoopGroup eventLoopGroup;
-    Codec codec;
-    String host = "localhost";
-    int port = DEFAULT_PORT;
+    private Codec codec;
+    private String host = "localhost";
+    private int port = DEFAULT_PORT;
     CompressionCodecFactory compressionCodecFactory;
     private String virtualHost = "/";
     private Duration requestedHeartbeat = Duration.ofSeconds(60);
@@ -2227,12 +2227,32 @@ public ClientParameters rpcTimeout(Duration rpcTimeout) {
       return this;
     }
 
+    String host() {
+      return this.host;
+    }
+
+    int port() {
+      return this.port;
+    }
+
+    Map<String, String> clientProperties() {
+      return Collections.unmodifiableMap(this.clientProperties);
+    }
+
+    Codec codec() {
+      return this.codec;
+    }
+
     ClientParameters duplicate() {
       ClientParameters duplicate = new ClientParameters();
       for (Field field : ClientParameters.class.getDeclaredFields()) {
         field.setAccessible(true);
         try {
-          field.set(duplicate, field.get(this));
+          Object value = field.get(this);
+          if (value instanceof Map) {
+            value = new ConcurrentHashMap<>((Map<?, ?>) value);
+          }
+          field.set(duplicate, value);
         } catch (IllegalAccessException e) {
           throw new StreamException("Error while duplicating client parameters", e);
         }

src/main/java/com/rabbitmq/stream/impl/ConsumersCoordinator.java

@@ -594,12 +594,11 @@ private ClientSubscriptionsManager(
             }
             return result;
           };
+      String connectionName = connectionNamingStrategy.apply(ClientConnectionType.CONSUMER);
       ClientFactoryContext clientFactoryContext =
           ClientFactoryContext.fromParameters(
                   clientParameters
-                      .clientProperty(
-                          "connection_name",
-                          connectionNamingStrategy.apply(ClientConnectionType.CONSUMER))
+                      .clientProperty("connection_name", connectionName)
                       .chunkListener(chunkListener)
                       .creditNotification(creditNotification)
                       .messageListener(messageListener)
@@ -608,6 +607,7 @@ private ClientSubscriptionsManager(
                       .consumerUpdateListener(consumerUpdateListener))
               .key(owner.name);
       this.client = clientFactory.client(clientFactoryContext);
+      LOGGER.debug("Created consumer connection '{}'", connectionName);
       maybeExchangeCommandVersions(client);
       clientInitializedInManager.set(true);
     }

src/main/java/com/rabbitmq/stream/impl/ProducersCoordinator.java

@@ -490,18 +490,18 @@ private ClientProducersManager(
               }
             }
           };
+      String connectionName = connectionNamingStrategy.apply(ClientConnectionType.PRODUCER);
       ClientFactoryContext connectionFactoryContext =
           ClientFactoryContext.fromParameters(
                   clientParameters
                       .publishConfirmListener(publishConfirmListener)
                       .publishErrorListener(publishErrorListener)
                       .shutdownListener(shutdownListener)
                       .metadataListener(metadataListener)
-                      .clientProperty(
-                          "connection_name",
-                          connectionNamingStrategy.apply(ClientConnectionType.PRODUCER)))
+                      .clientProperty("connection_name", connectionName))
               .key(owner.name);
       this.client = cf.client(connectionFactoryContext);
+      LOGGER.debug("Created producer connection '{}'", connectionName);
       clientInitializedInManager.set(true);
       ref.set(this.client);
     }

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -175,7 +175,7 @@ class StreamEnvironment implements Environment {
     if (uris.isEmpty()) {
       this.addresses =
           Collections.singletonList(
-              new Address(clientParametersPrototype.host, clientParametersPrototype.port));
+              new Address(clientParametersPrototype.host(), clientParametersPrototype.port()));
     } else {
       int defaultPort = tls ? Client.DEFAULT_TLS_PORT : Client.DEFAULT_PORT;
       this.addresses =
@@ -241,15 +241,15 @@ class StreamEnvironment implements Environment {
                               : addresses.get(random.nextInt(addresses.size()));
                       address = addressResolver.resolve(address);
                       LOGGER.debug("Trying to reconnect locator on {}", address);
+                      String connectionName =
+                          connectionNamingStrategy.apply(ClientConnectionType.LOCATOR);
                       Client newLocator =
                           clientFactory.apply(
                               newLocatorParameters
                                   .host(address.host())
                                   .port(address.port())
-                                  .clientProperty(
-                                      "connection_name",
-                                      connectionNamingStrategy.apply(
-                                          ClientConnectionType.LOCATOR)));
+                                  .clientProperty("connection_name", connectionName));
+                      LOGGER.debug("Created locator connection '{}'", connectionName);
                       LOGGER.debug("Locator connected on {}", address);
                       return newLocator;
                     })
@@ -267,17 +267,17 @@ class StreamEnvironment implements Environment {
           RuntimeException lastException = null;
           for (Address address : addresses) {
             address = addressResolver.resolve(address);
+            String connectionName = connectionNamingStrategy.apply(ClientConnectionType.LOCATOR);
             Client.ClientParameters locatorParameters =
                 clientParametersForInit
                     .duplicate()
                     .host(address.host())
                     .port(address.port())
-                    .clientProperty(
-                        "connection_name",
-                        connectionNamingStrategy.apply(ClientConnectionType.LOCATOR))
+                    .clientProperty("connection_name", connectionName)
                     .shutdownListener(shutdownListenerReference.get());
             try {
               this.locator = clientFactory.apply(locatorParameters);
+              LOGGER.debug("Created locator connection '{}'", connectionName);
               LOGGER.debug("Locator connected to {}", address);
               break;
             } catch (RuntimeException e) {
@@ -297,7 +297,9 @@ class StreamEnvironment implements Environment {
       this.locatorInitializationSequence = () -> {};
     }
     this.codec =
-        clientParametersPrototype.codec == null ? Codecs.DEFAULT : clientParametersPrototype.codec;
+        clientParametersPrototype.codec() == null
+            ? Codecs.DEFAULT
+            : clientParametersPrototype.codec();
     this.clockRefreshFuture =
         this.scheduledExecutorService.scheduleAtFixedRate(
             () -> this.clock.refresh(), 1, 1, SECONDS);

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -114,7 +114,7 @@ static short encodeResponseCode(Short code) {
   static ClientFactory coordinatorClientFactory(StreamEnvironment environment) {
     return context -> {
       ClientParameters parametersCopy = context.parameters().duplicate();
-      Address address = new Address(parametersCopy.host, parametersCopy.port);
+      Address address = new Address(parametersCopy.host(), parametersCopy.port());
       address = environment.addressResolver().resolve(address);
       parametersCopy.host(address.host()).port(address.port());
 

src/test/java/com/rabbitmq/stream/impl/ClientParametersTest.java

@@ -15,21 +15,24 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.lang.reflect.Field;
+import com.rabbitmq.stream.impl.Client.ClientParameters;
 import org.junit.jupiter.api.Test;
 
 public class ClientParametersTest {
 
   @Test
-  void duplicate() throws Exception {
-    Client.ClientParameters clientParameters =
+  void duplicate() {
+    Client.ClientParameters clientParameters1 =
         new Client.ClientParameters().host("rabbitmq").port(5556);
-    clientParameters = clientParameters.duplicate();
-    Field hostField = Client.ClientParameters.class.getDeclaredField("host");
-    hostField.setAccessible(true);
-    Field portField = Client.ClientParameters.class.getDeclaredField("port");
-    portField.setAccessible(true);
-    assertThat(hostField.get(clientParameters)).isEqualTo("rabbitmq");
-    assertThat(portField.get(clientParameters)).isEqualTo(5556);
+    clientParameters1.clientProperty("connection_name", "producer");
+    ClientParameters clientParameters2 = clientParameters1.duplicate();
+    assertThat(clientParameters2.host()).isEqualTo("rabbitmq");
+    assertThat(clientParameters2.port()).isEqualTo(5556);
+
+    // same as original
+    assertThat(clientParameters2.clientProperties()).containsEntry("connection_name", "producer");
+    // changing the copy should not change the original
+    clientParameters2.clientProperty("connection_name", "consumer");
+    assertThat(clientParameters1.clientProperties()).containsEntry("connection_name", "producer");
   }
 }