Fix for https://issues.apache.org/jira/browse/AMQ-4719

activemq-broker/src/main/java/org/apache/activemq/broker/ConnectionContext.java

@@ -332,4 +332,8 @@ public class ConnectionContext {
     public XATransactionId getXid() {
         return xid;
     }
+
+    public boolean isAllowLinkStealing(){
+       return connector != null && connector.isAllowLinkStealing();
+    }
 }

activemq-broker/src/main/java/org/apache/activemq/broker/Connector.java

@@ -60,4 +60,10 @@ public interface Connector extends Service {
     public boolean  isUpdateClusterClientsOnRemove();
 
     int connectionCount();
+
+    /**
+     * If enabled, older connections with the same clientID are stopped
+     * @return true/false if link stealing is enabled
+     */
+    boolean isAllowLinkStealing();
 }

activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnector.java

@@ -74,6 +74,7 @@ public class TransportConnector implements Connector, BrokerServiceAware {
     private int maximumProducersAllowedPerConnection = Integer.MAX_VALUE;
     private int maximumConsumersAllowedPerConnection  = Integer.MAX_VALUE;
     private PublishedAddressPolicy publishedAddressPolicy = new PublishedAddressPolicy();
+    private boolean allowLinkStealing;
 
     LinkedList<String> peerBrokers = new LinkedList<String>();
 
@@ -573,6 +574,15 @@ public class TransportConnector implements Connector, BrokerServiceAware {
         return connections.size();
     }
 
+    @Override
+    public boolean isAllowLinkStealing() {
+        return allowLinkStealing;
+    }
+
+    public void setAllowLinkStealing (boolean allowLinkStealing) {
+        this.allowLinkStealing=allowLinkStealing;
+    }
+
     public boolean isAuditNetworkProducers() {
         return auditNetworkProducers;
     }

activemq-broker/src/main/java/org/apache/activemq/broker/region/RegionBroker.java

@@ -231,8 +231,18 @@ public class RegionBroker extends EmptyBroker {
         synchronized (clientIdSet) {
             ConnectionContext oldContext = clientIdSet.get(clientId);
             if (oldContext != null) {
-                throw new InvalidClientIDException("Broker: " + getBrokerName() + " - Client: " + clientId + " already connected from "
+                if (context.isAllowLinkStealing()){
-                    + oldContext.getConnection().getRemoteAddress());
+                     clientIdSet.remove(clientId);
+                     if (oldContext.getConnection() != null) {
+                        LOG.warn("Stealing link for clientId " + clientId + " From Connection " + oldContext.getConnection());
+                        oldContext.getConnection().stop();
+                     }else{
+                         LOG.error("Not Connection for " + oldContext);
+                     }
+                }else{
+                    throw new InvalidClientIDException("Broker: " + getBrokerName() + " - Client: " + clientId + " already connected from "
+                            + oldContext.getConnection().getRemoteAddress());
+                }
             } else {
                 clientIdSet.put(clientId, context);
             }