Merge pull request #994 from rabbitmq/refactor-duplicate

Do not use reflection to duplicate ClientParameters and consumer builder

Author: acogoluegnes

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

@@ -115,7 +115,6 @@
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import java.io.OutputStream;
-import java.lang.reflect.Field;
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
@@ -2578,6 +2577,7 @@ public static ClientParameters maybeSetUpClientParametersFromUris(
   }
 
   public static class ClientParameters {
+    // IMPORTANT: When adding new fields, make sure to update the copy constructor method below!
 
     private final Map<String, String> clientProperties = new ConcurrentHashMap<>();
     EventLoopGroup eventLoopGroup;
@@ -2622,6 +2622,44 @@ public static class ClientParameters {
     // for other server frames
     private ExecutorServiceFactory executorServiceFactory;
 
+    public ClientParameters() {}
+
+    public ClientParameters(ClientParameters other) {
+      // Copy clientProperties map
+      this.clientProperties.putAll(other.clientProperties);
+
+      // Copy all other fields
+      this.eventLoopGroup = other.eventLoopGroup;
+      this.codec = other.codec;
+      this.host = other.host;
+      this.port = other.port;
+      this.compressionCodecFactory = other.compressionCodecFactory;
+      this.virtualHost = other.virtualHost;
+      this.requestedHeartbeat = other.requestedHeartbeat;
+      this.requestedMaxFrameSize = other.requestedMaxFrameSize;
+      this.publishConfirmListener = other.publishConfirmListener;
+      this.publishErrorListener = other.publishErrorListener;
+      this.chunkListener = other.chunkListener;
+      this.messageListener = other.messageListener;
+      this.messageIgnoredListener = other.messageIgnoredListener;
+      this.metadataListener = other.metadataListener;
+      this.creditNotification = other.creditNotification;
+      this.consumerUpdateListener = other.consumerUpdateListener;
+      this.shutdownListener = other.shutdownListener;
+      this.saslConfiguration = other.saslConfiguration;
+      this.credentialsProvider = other.credentialsProvider;
+      this.credentialsManager = other.credentialsManager;
+      this.chunkChecksum = other.chunkChecksum;
+      this.metricsCollector = other.metricsCollector;
+      this.sslContext = other.sslContext;
+      this.byteBufAllocator = other.byteBufAllocator;
+      this.rpcTimeout = other.rpcTimeout;
+      this.channelCustomizer = other.channelCustomizer;
+      this.bootstrapCustomizer = other.bootstrapCustomizer;
+      this.dispatchingExecutorServiceFactory = other.dispatchingExecutorServiceFactory;
+      this.executorServiceFactory = other.executorServiceFactory;
+    }
+
     public ClientParameters host(String host) {
       this.host = host;
       return this;
@@ -2831,20 +2869,7 @@ Duration rpcTimeout() {
     }
 
     public ClientParameters duplicate() {
-      ClientParameters duplicate = new ClientParameters();
-      for (Field field : ClientParameters.class.getDeclaredFields()) {
-        field.setAccessible(true);
-        try {
-          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);
-        }
-      }
-      return duplicate;
+      return new ClientParameters(this);
     }
   }
 

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -25,11 +25,8 @@
 import com.rabbitmq.stream.MessageHandler;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.Resource;
-import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.SubscriptionListener;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
-import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -64,6 +61,31 @@ public StreamConsumerBuilder(StreamEnvironment environment) {
     this.environment = environment;
   }
 
+  public StreamConsumerBuilder(StreamConsumerBuilder other) {
+    // Copy environment reference
+    this.environment = other.environment;
+
+    // Copy subscriptionProperties map
+    this.subscriptionProperties.putAll(other.subscriptionProperties);
+
+    // Copy all other fields
+    this.stream = other.stream;
+    this.superStream = other.superStream;
+    this.offsetSpecification = other.offsetSpecification;
+    this.messageHandler = other.messageHandler;
+    this.name = other.name;
+    this.autoTrackingStrategy = other.autoTrackingStrategy;
+    this.manualTrackingStrategy = other.manualTrackingStrategy;
+    this.noTrackingStrategy = other.noTrackingStrategy;
+    this.lazyInit = other.lazyInit;
+    this.subscriptionListener = other.subscriptionListener;
+    // Note: flowConfiguration is initialized in field declaration, copy its strategy
+    this.flowConfiguration.strategy(other.flowConfiguration.getStrategy());
+    this.consumerUpdateListener = other.consumerUpdateListener;
+    this.filterConfiguration = other.filterConfiguration;
+    this.listeners.addAll(other.listeners);
+  }
+
   @Override
   public ConsumerBuilder stream(String stream) {
     this.stream = stream;
@@ -276,19 +298,7 @@ public Consumer build() {
   }
 
   StreamConsumerBuilder duplicate() {
-    StreamConsumerBuilder duplicate = new StreamConsumerBuilder(this.environment);
-    for (Field field : StreamConsumerBuilder.class.getDeclaredFields()) {
-      if (Modifier.isStatic(field.getModifiers())) {
-        continue;
-      }
-      field.setAccessible(true);
-      try {
-        field.set(duplicate, field.get(this));
-      } catch (IllegalAccessException e) {
-        throw new StreamException("Error while duplicating stream producer builder", e);
-      }
-    }
-    return duplicate;
+    return new StreamConsumerBuilder(this);
   }
 
   static class TrackingConfiguration {
@@ -476,6 +486,10 @@ public FlowConfiguration strategy(ConsumerFlowStrategy strategy) {
     public ConsumerBuilder builder() {
       return this.consumerBuilder;
     }
+
+    private ConsumerFlowStrategy getStrategy() {
+      return this.strategy;
+    }
   }
 
   // to help testing

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
@@ -15,8 +15,11 @@
 package com.rabbitmq.stream.impl;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 import com.rabbitmq.stream.impl.Client.ClientParameters;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
 import org.junit.jupiter.api.Test;
 
 public class ClientParametersTest {
@@ -36,4 +39,20 @@ void duplicate() {
     clientParameters2.clientProperty("connection_name", "consumer");
     assertThat(clientParameters1.clientProperties()).containsEntry("connection_name", "producer");
   }
+
+  @Test
+  void duplicateMethodHandlesAllFields() {
+    Field[] allFields = ClientParameters.class.getDeclaredFields();
+    int nonStaticFields = 0;
+    for (Field field : allFields) {
+      if (!Modifier.isStatic(field.getModifiers())) {
+        nonStaticFields++;
+      }
+    }
+
+    assertEquals(
+        30,
+        nonStaticFields,
+        "If this fails, update the copy constructor method to handle the new field(s)");
+  }
 }

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

@@ -0,0 +1,40 @@
+// Copyright (c) 2026 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+package com.rabbitmq.stream.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import org.junit.jupiter.api.Test;
+
+public class StreamConsumerBuilderTest {
+
+  @Test
+  void duplicateMethodHandlesAllFields() {
+    Field[] allFields = StreamConsumerBuilder.class.getDeclaredFields();
+    int nonStaticFields = 0;
+    for (Field field : allFields) {
+      if (!Modifier.isStatic(field.getModifiers())) {
+        nonStaticFields++;
+      }
+    }
+
+    assertEquals(
+        16,
+        nonStaticFields,
+        "If this fails, update the copy constructor method to handle the new field(s)");
+  }
+}