This closes #698

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/ProtonProtocolManager.java

@@ -106,8 +106,9 @@ public class ProtonProtocolManager implements ProtocolManager<Interceptor>, Noti
          ttl = server.getConfiguration().getConnectionTTLOverride();
       }
 
+      String id = server.getConfiguration().getName();
       AMQPServerConnectionContext amqpConnection = ProtonServerConnectionContextFactory.getFactory().
-         createConnection(connectionCallback, (int) ttl, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX, server.getExecutorFactory().getExecutor(), server.getScheduledPool());
+         createConnection(connectionCallback, id, (int) ttl, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX, server.getExecutorFactory().getExecutor(), server.getScheduledPool());
 
       Executor executor = server.getExecutorFactory().getExecutor();
 

artemis-protocols/artemis-proton-plug/src/main/java/org/proton/plug/AMQPConnectionContextFactory.java

@@ -25,6 +25,7 @@ public abstract class AMQPConnectionContextFactory {
     * @return
     */
    public abstract AMQPConnectionContext createConnection(AMQPConnectionCallback connectionCallback,
+                                                          String containerId,
                                                           int idleTimeout,
                                                           int maxFrameSize,
                                                           int channelMax,

artemis-protocols/artemis-proton-plug/src/main/java/org/proton/plug/context/AbstractConnectionContext.java

@@ -17,6 +17,7 @@
 package org.proton.plug.context;
 
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
@@ -47,10 +48,12 @@ public abstract class AbstractConnectionContext extends ProtonInitializable impl
    private static final Logger log = Logger.getLogger(AbstractConnectionContext.class);
 
    public static final Symbol CONNECTION_OPEN_FAILED = Symbol.valueOf("amqp:connection-establishment-failed");
+   public static final String AMQP_CONTAINER_ID = "amqp-container-id";
 
    protected final ProtonHandler handler;
 
    protected AMQPConnectionCallback connectionCallback;
+   private final String containerId;
    private final ScheduledExecutorService scheduledPool;
 
    private final Map<Session, AbstractProtonSessionContext> sessions = new ConcurrentHashMap<>();
@@ -58,16 +61,18 @@ public abstract class AbstractConnectionContext extends ProtonInitializable impl
    protected LocalListener listener = new LocalListener();
 
    public AbstractConnectionContext(AMQPConnectionCallback connectionCallback, Executor dispatchExecutor, ScheduledExecutorService scheduledPool) {
-      this(connectionCallback, DEFAULT_IDLE_TIMEOUT, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX, dispatchExecutor, scheduledPool);
+      this(connectionCallback, null, DEFAULT_IDLE_TIMEOUT, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX, dispatchExecutor, scheduledPool);
    }
 
    public AbstractConnectionContext(AMQPConnectionCallback connectionCallback,
+                                    String containerId,
                                     int idleTimeout,
                                     int maxFrameSize,
                                     int channelMax,
                                     Executor dispatchExecutor,
                                     ScheduledExecutorService scheduledPool) {
       this.connectionCallback = connectionCallback;
+      this.containerId = (containerId != null) ? containerId : UUID.randomUUID().toString();
       this.scheduledPool = scheduledPool;
       connectionCallback.setConnection(this);
       this.handler =   ProtonHandler.Factory.create(dispatchExecutor);
@@ -190,6 +195,7 @@ public abstract class AbstractConnectionContext extends ProtonInitializable impl
       public void onRemoteOpen(Connection connection) throws Exception {
          synchronized (getLock()) {
             connection.setContext(AbstractConnectionContext.this);
+            connection.setContainer(containerId);
             connection.open();
          }
          initialise();

artemis-protocols/artemis-proton-plug/src/main/java/org/proton/plug/context/client/ProtonClientConnectionContext.java

@@ -39,12 +39,13 @@ public class ProtonClientConnectionContext extends AbstractConnectionContext imp
    }
 
    public ProtonClientConnectionContext(AMQPConnectionCallback connectionCallback,
+                                        String containerId,
                                         int idleTimeout,
                                         int maxFrameSize,
                                         int channelMax,
                                         Executor dispatchExecutor,
                                         ScheduledExecutorService scheduledPool) {
-      super(connectionCallback, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
+      super(connectionCallback, containerId, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
    }
 
    // Maybe a client interface?

artemis-protocols/artemis-proton-plug/src/main/java/org/proton/plug/context/client/ProtonClientConnectionContextFactory.java

@@ -39,11 +39,12 @@ public class ProtonClientConnectionContextFactory extends AMQPConnectionContextF
 
    @Override
    public AMQPConnectionContext createConnection(AMQPConnectionCallback connectionCallback,
+                                                 String containerId,
                                                  int idleTimeout,
                                                  int maxFrameSize,
                                                  int channelMax,
                                                  Executor dispatchExecutor,
                                                  ScheduledExecutorService scheduledPool) {
-      return new ProtonClientConnectionContext(connectionCallback, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
+      return new ProtonClientConnectionContext(connectionCallback, containerId, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
    }
 }

artemis-protocols/artemis-proton-plug/src/main/java/org/proton/plug/context/server/ProtonServerConnectionContext.java

@@ -38,12 +38,13 @@ public class ProtonServerConnectionContext extends AbstractConnectionContext imp
    }
 
    public ProtonServerConnectionContext(AMQPConnectionCallback connectionSP,
+                                        String containerId,
                                         int idleTimeout,
                                         int maxFrameSize,
                                         int channelMax,
                                         Executor dispatchExecutor,
                                         ScheduledExecutorService scheduledPool) {
-      super(connectionSP, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
+      super(connectionSP, containerId, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
    }
 
    @Override

artemis-protocols/artemis-proton-plug/src/main/java/org/proton/plug/context/server/ProtonServerConnectionContextFactory.java

@@ -37,16 +37,17 @@ public class ProtonServerConnectionContextFactory extends AMQPConnectionContextF
 
    @Override
    public AMQPServerConnectionContext createConnection(AMQPConnectionCallback connectionCallback, Executor dispatchExecutor, ScheduledExecutorService scheduledPool) {
-      return createConnection(connectionCallback, DEFAULT_IDLE_TIMEOUT, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX, dispatchExecutor, scheduledPool);
+      return createConnection(connectionCallback, null, DEFAULT_IDLE_TIMEOUT, DEFAULT_MAX_FRAME_SIZE, DEFAULT_CHANNEL_MAX, dispatchExecutor, scheduledPool);
    }
 
    @Override
    public AMQPServerConnectionContext createConnection(AMQPConnectionCallback connectionCallback,
+                                                       String containerId,
                                                        int idleTimeout,
                                                        int maxFrameSize,
                                                        int channelMax,
                                                        Executor dispatchExecutor,
                                                        ScheduledExecutorService scheduledPool) {
-      return new ProtonServerConnectionContext(connectionCallback, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
+      return new ProtonServerConnectionContext(connectionCallback, containerId, idleTimeout, maxFrameSize, channelMax, dispatchExecutor, scheduledPool);
    }
 }

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java

@@ -95,6 +95,8 @@ public class ProtonTest extends ActiveMQTestBase {
 
    private static final String password = "guest";
 
+   private static final String brokerName = "my-broker";
+
    // this will ensure that all tests in this class are run twice,
    // once with "true" passed to the class' constructor and once with "false"
    @Parameterized.Parameters(name = "{0}")
@@ -137,6 +139,7 @@ public class ProtonTest extends ActiveMQTestBase {
       TransportConfiguration transportConfiguration = new TransportConfiguration(NETTY_ACCEPTOR_FACTORY, params);
 
       server.getConfiguration().getAcceptorConfigurations().add(transportConfiguration);
+      server.getConfiguration().setName(brokerName);
 
       // Default Page
       AddressSettings addressSettings = new AddressSettings();
@@ -186,6 +189,20 @@ public class ProtonTest extends ActiveMQTestBase {
       }
    }
 
+   @Test
+   public void testBrokerContainerId() throws Exception {
+      if (protocol != 0 && protocol != 3) return; // Only run this test for AMQP protocol
+
+      AmqpClient client = new AmqpClient(new URI(tcpAmqpConnectionUri), userName, password);
+      AmqpConnection amqpConnection = client.connect();
+      try {
+         assertTrue(brokerName.equals(amqpConnection.getEndpoint().getRemoteContainer()));
+      }
+      finally {
+         amqpConnection.close();
+      }
+   }
+
    @Test
    public void testCreditsAreAllocatedOnlyOnceOnLinkCreate() throws Exception {
       if (protocol != 0 && protocol != 3) return; // Only run this test for AMQP protocol