ARTEMIS-5976 Pass clientID to session creation in ActiveMQResourceAdapter

The ActiveMQResourceAdapter.createSession() method was not passing the
clientID parameter when creating sessions via ClientSessionFactory,
resulting in remoting connection without the clientID property set on the
server side.

Author: brusdev

artemis-ra/src/main/java/org/apache/activemq/artemis/ra/ActiveMQResourceAdapter.java

@@ -1030,20 +1030,20 @@ public ClientSession createSession(final ClientSessionFactory parameterFactory,
          // If transacted we need to send the ack flush as soon as possible as if any transaction times out, we need
          // the ack on the server already
          if (useLocalTx) {
-            result = parameterFactory.createSession(user, pass, false, false, false, false, 0);
+            result = parameterFactory.createSession(user, pass, false, false, false, false, 0, getClientID());
          } else {
-            result = parameterFactory.createSession(user, pass, true, false, false, false, 0);
+            result = parameterFactory.createSession(user, pass, true, false, false, false, 0, getClientID());
          }
       } else {
          if (preAck != null && preAck) {
-            result = parameterFactory.createSession(user, pass, false, true, true, true, -1);
+            result = parameterFactory.createSession(user, pass, false, true, true, true, -1, getClientID());
          } else {
             // only auto ack and dups ok are supported
             result = switch (ackMode) {
-               case Session.AUTO_ACKNOWLEDGE -> parameterFactory.createSession(user, pass, false, true, true, false, 0);
+               case Session.AUTO_ACKNOWLEDGE -> parameterFactory.createSession(user, pass, false, true, true, false, 0, getClientID());
                case Session.DUPS_OK_ACKNOWLEDGE -> {
                   int actDupsOkBatchSize = dupsOkBatchSize != null ? dupsOkBatchSize : ActiveMQClient.DEFAULT_ACK_BATCH_SIZE;
-                  yield parameterFactory.createSession(user, pass, false, true, true, false, actDupsOkBatchSize);
+                  yield parameterFactory.createSession(user, pass, false, true, true, false, actDupsOkBatchSize, getClientID());
                }
                default -> throw new IllegalArgumentException("Invalid ackmode: " + ackMode);
             };

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/ra/ResourceAdapterTest.java

@@ -891,6 +891,69 @@ public void testConnectionFactoryPropertiesApplyToRecoveryConfig() throws Except
 
    }
 
+   @Test
+   public void testClientIDPassedToSessionWithUseLocalTx() throws Exception {
+      testClientIDPassedToSession("testClientID", ra -> ra.setUseLocalTx(true), spec -> { }, false);
+   }
+
+   @Test
+   public void testClientIDPassedToSessionWithDeliveryTransacted() throws Exception {
+      testClientIDPassedToSession("testClientIDTransacted", ra -> { }, spec -> { }, true);
+   }
+
+   @Test
+   public void testClientIDPassedToSessionWithPreAck() throws Exception {
+      testClientIDPassedToSession("testClientIDPreAck", ra -> ra.setPreAcknowledge(true), spec -> { }, false);
+   }
+
+   @Test
+   public void testClientIDPassedToSessionWithDupsOkAck() throws Exception {
+      testClientIDPassedToSession("testClientIDDupsOk",
+         ra -> ra.setDupsOKBatchSize(100),
+         spec -> spec.setAcknowledgeMode("Dups-ok-acknowledge"),
+         false);
+   }
+
+   @Test
+   public void testClientIDPassedToSessionWithAutoAck() throws Exception {
+      testClientIDPassedToSession("testClientIDAutoAck",
+         ra -> { },
+         spec -> spec.setAcknowledgeMode("Auto-acknowledge"),
+         false);
+   }
+
+   private void testClientIDPassedToSession(String clientID,
+                                             java.util.function.Consumer<ActiveMQResourceAdapter> raConfigurator,
+                                             java.util.function.Consumer<ActiveMQActivationSpec> specConfigurator,
+                                             boolean deliveryTransacted) throws Exception {
+      ActiveMQResourceAdapter ra = new ActiveMQResourceAdapter();
+      ra.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+      ra.setClientID(clientID);
+      raConfigurator.accept(ra);
+      ra.start(new BootstrapContext());
+
+      ActiveMQActivationSpec spec = new ActiveMQActivationSpec();
+      spec.setResourceAdapter(ra);
+      spec.setUseJNDI(false);
+      spec.setDestinationType("javax.jms.Queue");
+      spec.setDestination(MDBQUEUE);
+      specConfigurator.accept(spec);
+
+      CountDownLatch latch = new CountDownLatch(1);
+      DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+      DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, deliveryTransacted);
+
+      ra.endpointActivation(endpointFactory, spec);
+
+      // Verify the clientID is set on the server-side remoting connection
+      assertFalse(server.getRemotingService().getConnections().isEmpty());
+      assertTrue(server.getRemotingService().getConnections().stream().allMatch(
+         remotingConnection -> clientID.equals(remotingConnection.getClientID())));
+
+      ra.stop();
+      assertTrue(endpoint.released);
+   }
+
    @Override
    public boolean useSecurity() {
       return false;