Merge pull request eclipse-kapua#4087 from eclipse/feat-upgrade_artemis

⬆️  [Dependencies] Upgrade Artemis from 2.31.2 to 2.38.0

assembly/broker-artemis/configurations/broker.xml

@@ -91,6 +91,18 @@
         -->
         <!-- allowLinkStealing=true -->
 
+        <!-- CLUSTER CONFIGURATION NOTES -->
+        <!-- The current Kapua deployment deploys a broker with 2 consumers (Lifecycle and Telemetry) plus the Authentication service.
+             If a cluster is needed, the following steps could be followed:
+                - change the run-broker script, section "/opt/artemis/bin/artemis create \", (of the broker container) to create the cluster simply adding "-clustered" option and changing cluster username and password.
+                  (see https://activemq.apache.org/components/artemis/documentation/latest/using-server.html#creating-a-broker-instance)
+                - change this configuration file adding the cluster connectors as per cluster topology
+                  (see https://activemq.apache.org/components/artemis/documentation/latest/clusters.html)
+                - change the consumer/service connection string according to the documentation
+                  (see section 3.2.2 of the link https://activemq.apache.org/components/artemis/documentation/latest/clusters.html)
+             NOTE: The consumers/service could be scaled up also and independently each from the others.
+             IMPORTANT NOTE: while clustering pay attention to the performances as described in chapter 2 of the link https://activemq.apache.org/components/artemis/documentation/latest/clusters.html.
+        -->
         <!-- disable caching at all, otherwise we'll have issues while managing stealing link:
             the remove connection (ServerPlugin#afterDestroyConnection) is called without calling the authenticate (SecurityPlugin#authenticate).
         TODO do further tests to check the performance impact and try to enable authorization cache only -->
@@ -162,5 +174,20 @@
                 </anycast>
             </address>
         </addresses>
+
+        <!-- Uncomment the following if you want to use the Standard LoggingActiveMQServerPlugin pluging to log in events
+        <broker-plugins>
+            <broker-plugin class-name="org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin">
+                <property key="LOG_ALL_EVENTS" value="true"/>
+                <property key="LOG_CONNECTION_EVENTS" value="true"/>
+                <property key="LOG_SESSION_EVENTS" value="true"/>
+                <property key="LOG_CONSUMER_EVENTS" value="true"/>
+                <property key="LOG_DELIVERING_EVENTS" value="true"/>
+                <property key="LOG_SENDING_EVENTS" value="true"/>
+                <property key="LOG_INTERNAL_EVENTS" value="true"/>
+            </broker-plugin>
+        </broker-plugins>
+        -->
+
     </core>
 </configuration>

assembly/broker-artemis/configurations/logback.xml

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Copyright (c) 2020, 2022 Red Hat Inc and others
+
+    This program and the accompanying materials are made
+    available under the terms of the Eclipse Public License 2.0
+    which is available at https://www.eclipse.org/legal/epl-2.0/
+
+    SPDX-License-Identifier: EPL-2.0
+
+    Contributors:
+        Red Hat Inc - initial API and implementation
+        Eurotech
+ -->
+<!DOCTYPE xml>
+<configuration>
+    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <logger name="liquibase" level="WARN"/>
+    <logger name="org.apache.activemq.audit.base" level="WARN"/>
+    <logger name="org.apache.activemq.audit.message" level="WARN"/>
+    <logger name="org.apache.activemq.audit.resource" level="WARN"/>
+
+    <root level="INFO">
+        <appender-ref ref="console"/>
+    </root>
+</configuration>

assembly/broker-artemis/descriptors/kapua-broker-artemis.xml

@@ -36,10 +36,16 @@
             <outputDirectory>/opt/artemis</outputDirectory>
             <directory>${project.build.directory}/dependencies/artemis/apache-artemis-${artemis.version}</directory>
             <fileMode>0644</fileMode>
-            <!-- <excludes>
+            <excludes>
+                <exclude>lib/log4j-over-slf4j-*.jar</exclude>
+                <exclude>lib/log4j-core-*.jar</exclude>
+                <exclude>lib/log4j-slf4j-impl-*.jar</exclude>
+                <exclude>lib/log4j-slf4j2-impl-*.jar</exclude>
+                <!--
                 <exclude>activemq-all-*</exclude>
                 <exclude>webapps-demo/</exclude>
-            </excludes> -->
+                -->
+            </excludes>
         </fileSet>
 
         <fileSet>
@@ -65,8 +71,7 @@
             <includes>
                 <include>ch.qos.logback:logback-core</include>
                 <include>ch.qos.logback:logback-classic</include>
-                <include>org.slf4j:log4j-over-slf4j</include>
-                <include>org.slf4j:slf4j-api</include>
+                <include>org.apache.logging.log4j:log4j-to-slf4j</include>
                 <include>commons-io:commons-io</include>
                 <include>commons-configuration:commons-configuration</include>
                 <include>commons-lang:commons-lang</include>

assembly/broker-artemis/pom.xml

@@ -195,10 +195,6 @@
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>log4j-over-slf4j</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-api</artifactId>

broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/plugin/security/ActivateCallback.java

@@ -0,0 +1,49 @@
+/*******************************************************************************
+ * Copyright (c) 2022 Eurotech and/or its affiliates and others
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Eurotech
+ *******************************************************************************/
+package org.eclipse.kapua.broker.artemis.plugin.security;
+
+import org.eclipse.kapua.broker.artemis.plugin.security.connector.AcceptorHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class to intercept Broker lifecycle events
+ */
+public class ActivateCallback implements org.apache.activemq.artemis.core.server.ActivateCallback {
+
+    private final Logger logger = LoggerFactory.getLogger(ActivateCallback.class);
+
+    private final AcceptorHandler acceptorHandler;
+
+    public ActivateCallback(AcceptorHandler acceptorHandler) {
+        this.acceptorHandler = acceptorHandler;
+    }
+
+    @Override
+    /**
+     * Use this callback to start Acceptors for now (but could be used to do whatever needs to have a fully running broker
+     */
+    public void activationComplete() {
+        logger.info("Broker activation completed!");
+        org.apache.activemq.artemis.core.server.ActivateCallback.super.activationComplete();
+        logger.info("Creating acceptors...");
+        try {
+            acceptorHandler.syncAcceptors();
+        } catch (Exception e) {
+            logger.error("Creating acceptors... ERROR: {}", e.getMessage(), e);
+            //TODO throw runtime? the broker doesn't work properly if not all the acceptors are created
+        }
+        logger.info("Creating acceptors... DONE");
+    }
+
+}

broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/plugin/security/SecurityPlugin.java

@@ -150,15 +150,15 @@ private Subject authenticateInternalConn(ConnectionInfo connectionInfo, String c
                     username, connectionInfo.getClientId(), connectionInfo.getClientIp(), remotingConnection.getID(),
                     remotingConnection.getTransportConnection().getRemoteAddress(), remotingConnection.getTransportConnection().isOpen());
             //TODO double check why the client id is null once coming from AMQP connection (the Kapua connection factory with custom client id generation is called)
-            KapuaPrincipal kapuaPrincipal = buildInternalKapuaPrincipal(getAdminAccountInfo().getId(), username, connectionInfo.getClientId());
-            //auto generate client id if null. It shouldn't be null but in some case the one from JMS connection is.
             String clientId = connectionInfo.getClientId();
             //set a random client id value if not set by the client
             //from JMS 2 specs "Although setting client ID remains mandatory when creating an unshared durable subscription, it is optional when creating a shared durable subscription."
             if (Strings.isNullOrEmpty(clientId)) {
                 clientId = clientIdPrefix + INDEX.getAndIncrement();
                 logger.info("Updated empty client id to: {}", clientId);
             }
+            KapuaPrincipal kapuaPrincipal = buildInternalKapuaPrincipal(getAdminAccountInfo().getId(), username, clientId);
+            //auto generate client id if null. It shouldn't be null but in some case the one from JMS connection is.
             //update client id with account|clientId (see pattern)
             String fullClientId = Utils.getFullClientId(getAdminAccountInfo().getId(), clientId);
             remotingConnection.setClientID(fullClientId);

broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/plugin/security/ServerPlugin.java

@@ -137,8 +137,8 @@ public void registered(ActiveMQServer server) {
             serverContext.init(server);
             acceptorHandler = new AcceptorHandler(server,
                     brokerSetting.getMap(String.class, BrokerSettingKey.ACCEPTORS));
-            //init acceptors
-            acceptorHandler.syncAcceptors();
+            //init activateCallback to handle acceptor initialization instead of calling it from here
+            server.registerActivateCallback(new ActivateCallback(acceptorHandler));
 
             deviceConnectionEventListenerService.addReceiver(serviceEvent -> processDeviceConnectionEvent(serviceEvent));
 

broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/plugin/security/connector/AcceptorHandler.java

@@ -83,32 +83,29 @@ public String removeAcceptor(String name) throws Exception {
      * @throws Exception
      */
     public void syncAcceptors() throws Exception {
-        logger.info("Init acceptors... server started: {} - {}", server.isStarted(), server.getState());
-        if (server.isStarted()) {
-            List<String> acceptorToRemove = new ArrayList<>();
-            server.getConfiguration().getAcceptorConfigurations().forEach(tc -> {
-                String acceptorName = tc.getName();
-                logger.info("Checking acceptor {}", acceptorName);
-                if (definedAcceptors.get(acceptorName) == null) {
-                    acceptorToRemove.add(acceptorName);
-                    logger.info("Adding acceptor {} to the remove list", acceptorName);
-                } else {
-                    logger.info("Leaving acceptor {} running", acceptorName);
-                }
-            });
-            acceptorToRemove.forEach(acceptorName -> {
-                logger.info("Stopping acceptor {}...", acceptorName);
-                try {
-                    server.getRemotingService().getAcceptor(acceptorName).stop();
-                    server.getRemotingService().destroyAcceptor(acceptorName);
-                    TransportConfiguration tc = getByName(acceptorName);
-                    server.getConfiguration().getAcceptorConfigurations().remove(tc);
-                } catch (Exception e) {
-                    logger.error("Error stopping acceptor {}... Error: {}", acceptorName, e.getMessage(), e);
-                }
-                logger.info("Stopping acceptor {}... DONE", acceptorName);
-            });
-        }
+        List<String> acceptorToRemove = new ArrayList<>();
+        server.getConfiguration().getAcceptorConfigurations().forEach(tc -> {
+            String acceptorName = tc.getName();
+            logger.info("Checking acceptor {}", acceptorName);
+            if (definedAcceptors.get(acceptorName) == null) {
+                acceptorToRemove.add(acceptorName);
+                logger.info("Adding acceptor {} to the remove list", acceptorName);
+            } else {
+                logger.info("Leaving acceptor {} running", acceptorName);
+            }
+        });
+        acceptorToRemove.forEach(acceptorName -> {
+            logger.info("Stopping acceptor {}...", acceptorName);
+            try {
+                server.getRemotingService().getAcceptor(acceptorName).stop();
+                server.getRemotingService().destroyAcceptor(acceptorName);
+                TransportConfiguration tc = getByName(acceptorName);
+                server.getConfiguration().getAcceptorConfigurations().remove(tc);
+            } catch (Exception e) {
+                logger.error("Error stopping acceptor {}... Error: {}", acceptorName, e.getMessage(), e);
+            }
+            logger.info("Stopping acceptor {}... DONE", acceptorName);
+        });
 //        server.getConfiguration().clearAcceptorConfigurations();
 
         definedAcceptors.forEach((name, uri) -> {

broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/plugin/security/context/Acl.java

@@ -33,18 +33,16 @@ public class Acl {
     private static final char ANY_WORDS = '#';
     private static final char SEPARATOR = '/';
 
-    private final LoginMetric loginMetric;
     private final WildcardConfiguration wildcardConfiguration;
     private final HierarchicalRepository<KapuaPrincipal> read;
     private final HierarchicalRepository<KapuaPrincipal> write;
     private final HierarchicalRepository<KapuaPrincipal> admin;
 
     public Acl(LoginMetric loginMetric, KapuaPrincipal principal, List<AuthAcl> authAcls) throws KapuaIllegalArgumentException {
-        this.loginMetric = loginMetric;
-        wildcardConfiguration = new WildcardConfiguration();
-        wildcardConfiguration.setSingleWord(SINGLE_WORD);
-        wildcardConfiguration.setAnyWords(ANY_WORDS);
-        wildcardConfiguration.setDelimiter(SEPARATOR);
+        wildcardConfiguration = new WildcardConfiguration().
+            setSingleWord(SINGLE_WORD).
+            setAnyWords(ANY_WORDS).
+            setDelimiter(SEPARATOR);
         if (principal == null) {
             throw new KapuaIllegalArgumentException("principal", null);
         }

client/security/src/main/java/org/eclipse/kapua/client/security/bean/KapuaPrincipalImpl.java

@@ -23,11 +23,10 @@
  *
  * @since 1.0
  */
-public class KapuaPrincipalImpl implements KapuaPrincipal {
+public class KapuaPrincipalImpl extends org.apache.activemq.artemis.spi.core.security.jaas.UserPrincipal implements KapuaPrincipal {
 
     private static final long serialVersionUID = -2120917772921438176L;
 
-    private String name;
     private String clientId;
     private boolean internal;
     /**
@@ -45,7 +44,7 @@ public class KapuaPrincipalImpl implements KapuaPrincipal {
      * @param authResponse
      */
     public KapuaPrincipalImpl(AuthResponse authResponse) {
-        name = authResponse.getUsername() + "@" + authResponse.getClientId();
+        super(authResponse.getUsername() + "@" + authResponse.getClientId());
         clientId = authResponse.getClientId();
         accessTokenId = authResponse.getAccessTokenId();
         userId = KapuaEid.parseCompactId(authResponse.getUserId());
@@ -61,17 +60,12 @@ public KapuaPrincipalImpl(AuthResponse authResponse) {
      * @param clientId
      */
     public KapuaPrincipalImpl(KapuaId accountId, String username, String clientId) {
+        super(username + "@" + clientId);
         internal = true;
         this.accountId = accountId;
-        name = username + "@" + clientId;
         this.clientId = clientId;
     }
 
-    @Override
-    public String getName() {
-        return name;
-    }
-
     @Override
     public String getTokenId() {
         return accessTokenId;
@@ -110,7 +104,7 @@ public boolean isInternal() {
     @Override
     public int hashCode() {
         //name is unique so let's use it for hashing
-        return (name == null ? 0 : name.hashCode());
+        return (getName() == null ? 0 : getName().hashCode());
     }
 
     @Override
@@ -125,11 +119,11 @@ public boolean equals(Object obj) {
             return false;
         }
         KapuaPrincipalImpl other = (KapuaPrincipalImpl) obj;
-        if (name == null) {
-            if (other.name != null) {
+        if (getName() == null) {
+            if (other.getName() != null) {
                 return false;
             }
-        } else if (!name.equals(other.name)) {
+        } else if (!getName().equals(other.getName())) {
             return false;
         }
         return true;

consumer/lifecycle-app/pom.xml

@@ -469,13 +469,6 @@
                                 </filter>
                             </filters>
                             <transformers>
-                                <!-- added by 2.7 -->
-                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
-                                    <resource>META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports</resource>
-                                </transformer>
-                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
-                                    <resource>META-INF/spring/org.springframework.boot.actuate.autoconfigure.web.ManagementContextConfiguration.imports</resource>
-                                </transformer>
                                 <transformer implementation="org.apache.maven.plugins.shade.resource.XmlAppendingTransformer">
                                     <resource>META-INF/persistence.xml</resource>
                                 </transformer>
@@ -507,6 +500,14 @@
                                 <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                     <resource>META-INF/services/org.eclipse.kapua.translator.Translator</resource>
                                 </transformer>
+
+                                <!-- add by 2.7 -->
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports</resource>
+                                </transformer>
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>META-INF/spring/org.springframework.boot.actuate.autoconfigure.web.ManagementContextConfiguration.imports</resource>
+                                </transformer>
                                 <!--
                                                     <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                                         <resource>META-INF/spring.handlers</resource>

consumer/lifecycle-app/src/main/resources/spring/applicationContext.xml

@@ -34,6 +34,7 @@
     </bean>
 
     <bean id="connectionFactory" class="org.eclipse.kapua.service.client.amqp.ServiceConnectionFactoryImpl">
+        <constructor-arg name="schema" type="java.lang.String" value="amqp"/>
         <constructor-arg name="host" type="java.lang.String" value="${BROKER_HOST}"/>
         <constructor-arg name="port" type="int" value="${BROKER_PORT}"/>
         <constructor-arg name="username" type="java.lang.String" value="${BROKER_INTERNAL_USERNAME}"/>