fix: avoid discovery proxy rebind in NlbSimulator

NlbSimulator.addNode() was tearing down and recreating the discovery
RoundRobinProxy on every call, which failed with "Address already in
use" because the new proxy tried to bind while the old one still held
the port.

Add addTarget()/removeTarget() to RoundRobinProxy so the discovery
proxy is created once (on the first addNode) and updated in-place as
nodes join or leave.

Fixes #851

Author: dkropachev

integration-tests/src/test/java/com/datastax/oss/driver/core/clientroutes/NlbSimulator.java

@@ -112,20 +112,16 @@ public synchronized void addNode(int nodeId) throws IOException {
     String nodeIp = ccmBridge.getNodeIpAddress(nodeId);
     InetSocketAddress nodeAddr = new InetSocketAddress(nodeIp, 9042);
 
-    // Create per-node proxy first, before updating state
     int nodePort = getNodePort(nodeId);
     TcpProxy proxy = new TcpProxy(bindAddress, nodePort, nodeAddr);
 
-    // Update state and rebuild discovery proxy; if rebuild fails, close the new proxy
     activeNodes.add(nodeId);
     nodeProxies.put(nodeId, proxy);
-    try {
-      rebuildDiscoveryProxy();
-    } catch (IOException e) {
-      activeNodes.remove(Integer.valueOf(nodeId));
-      nodeProxies.remove(nodeId);
-      proxy.close();
-      throw e;
+
+    if (discoveryProxy == null) {
+      discoveryProxy = buildDiscoveryProxy(nodeAddr);
+    } else {
+      discoveryProxy.addTarget(nodeAddr);
     }
 
     LOG.info("NLB: added node{} ({}:{}) -> proxy port {}", nodeId, nodeIp, 9042, nodePort);
@@ -139,37 +135,20 @@ public synchronized void removeNode(int nodeId) throws IOException {
     }
     activeNodes.remove(Integer.valueOf(nodeId));
 
-    // Rebuild discovery proxy with updated node list
-    rebuildDiscoveryProxy();
+    if (discoveryProxy != null) {
+      String nodeIp = ccmBridge.getNodeIpAddress(nodeId);
+      discoveryProxy.removeTarget(new InetSocketAddress(nodeIp, 9042));
+    }
 
     LOG.info("NLB: removed node{}", nodeId);
   }
 
-  private void rebuildDiscoveryProxy() throws IOException {
-    RoundRobinProxy oldProxy = discoveryProxy;
-
-    if (activeNodes.isEmpty()) {
-      discoveryProxy = null;
-      if (oldProxy != null) {
-        oldProxy.close();
-      }
-      return;
-    }
-
-    // Build target list from active nodes
+  private RoundRobinProxy buildDiscoveryProxy(InetSocketAddress firstTarget) throws IOException {
     List<InetSocketAddress> targets = new ArrayList<>();
-    for (int nodeId : activeNodes) {
-      String nodeIp = ccmBridge.getNodeIpAddress(nodeId);
-      targets.add(new InetSocketAddress(nodeIp, 9042));
-    }
-
-    // Create new proxy before closing old one to avoid inconsistent state on failure
-    discoveryProxy = new RoundRobinProxy(bindAddress, basePort, targets);
-    if (oldProxy != null) {
-      oldProxy.close();
-    }
-    LOG.info(
-        "NLB: discovery proxy on port {} -> {} nodes: {}", basePort, targets.size(), activeNodes);
+    targets.add(firstTarget);
+    RoundRobinProxy proxy = new RoundRobinProxy(bindAddress, basePort, targets);
+    LOG.info("NLB: discovery proxy on port {} -> {}", basePort, firstTarget);
+    return proxy;
   }
 
   @Override

integration-tests/src/test/java/com/datastax/oss/driver/core/clientroutes/RoundRobinProxy.java

@@ -47,7 +47,7 @@ public class RoundRobinProxy implements Closeable {
   private static final int BUFFER_SIZE = 8192;
 
   private final ServerSocket serverSocket;
-  private final List<InetSocketAddress> targets;
+  private final CopyOnWriteArrayList<InetSocketAddress> targets;
   private final AtomicInteger counter = new AtomicInteger(0);
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final CopyOnWriteArrayList<Socket> activeSockets = new CopyOnWriteArrayList<>();
@@ -59,7 +59,7 @@ public RoundRobinProxy(String bindAddress, int listenPort, List<InetSocketAddres
     if (targets.isEmpty()) {
       throw new IllegalArgumentException("At least one target required");
     }
-    this.targets = targets;
+    this.targets = new CopyOnWriteArrayList<>(targets);
     this.serverSocket = new ServerSocket();
     this.serverSocket.setReuseAddress(true);
     this.serverSocket.bind(new InetSocketAddress(bindAddress, listenPort));
@@ -75,6 +75,18 @@ public int getLocalPort() {
     return serverSocket.getLocalPort();
   }
 
+  /** Adds a target to the round-robin pool. */
+  public void addTarget(InetSocketAddress target) {
+    targets.add(target);
+    LOG.debug("RoundRobinProxy on port {} added target: {}", serverSocket.getLocalPort(), target);
+  }
+
+  /** Removes a target from the round-robin pool. */
+  public void removeTarget(InetSocketAddress target) {
+    targets.remove(target);
+    LOG.debug("RoundRobinProxy on port {} removed target: {}", serverSocket.getLocalPort(), target);
+  }
+
   private void acceptLoop() {
     while (!closed.get()) {
       try {