feat(integration): deliver client lifecycle events to integrations

application/src/main/java/org/thingsboard/mqtt/broker/actors/client/service/ActorProcessorImpl.java

@@ -47,6 +47,7 @@
 import org.thingsboard.mqtt.broker.service.security.authorization.AuthRulePatterns;
 import org.thingsboard.mqtt.broker.session.ClientMqttActorManager;
 import org.thingsboard.mqtt.broker.session.ClientSessionCtx;
+import org.thingsboard.mqtt.broker.service.integration.IntegrationLifecycleEventPublisher;
 import org.thingsboard.mqtt.broker.session.DisconnectReason;
 import org.thingsboard.mqtt.broker.session.DisconnectReasonType;
 import org.thingsboard.mqtt.broker.util.MqttReasonCodeResolver;
@@ -81,6 +82,7 @@ public class ActorProcessorImpl implements ActorProcessor {
     private final UnauthorizedClientManager unauthorizedClientManager;
     private final BlockedClientService blockedClientService;
     private final AuthorizationRoutingService authorizationRoutingService;
+    private final IntegrationLifecycleEventPublisher integrationLifecycleEventPublisher;
 
     @Override
     public void onInit(ClientActorState state, SessionInitMsg sessionInitMsg) {
@@ -96,11 +98,13 @@ public void onInit(ClientActorState state, SessionInitMsg sessionInitMsg) {
         }
 
         AuthContext authContext = buildAuthContext(state, sessionInitMsg);
+        sessionCtx.setUsername(authContext.getUsername());
         AuthResponse authResponse = authorizationRoutingService.executeAuthFlow(authContext);
 
         if (authResponse.notSuccess()) {
             log.warn("[{}] Connection is not established due to: {}", state.getClientId(), CONNECTION_REFUSED_NOT_AUTHORIZED);
             unauthorizedClientManager.persistClientUnauthorized(state, sessionInitMsg, authResponse.getReason());
+            integrationLifecycleEventPublisher.publishAuthenticationFailed(sessionCtx, state.getClientId(), authResponse.getReason());
             sendConnectionRefusedNotAuthorizedMsgAndCloseChannel(sessionCtx);
             return;
         }
@@ -203,10 +207,14 @@ public void onEnhancedReAuth(ClientActorState state, MqttAuthMsg authMsg) {
     private void processAuth(ClientActorState state, MqttAuthMsg authMsg, ClientSessionCtx sessionCtx) {
         EnhancedAuthContext authContext = buildEnhancedAuthContext(state, authMsg);
         EnhancedAuthFinalResponse authResponse = enhancedAuthenticationService.onAuthContinue(sessionCtx, authContext);
+        sessionCtx.setUsername(authResponse.username());
         if (!authResponse.success()) {
             resetStateToDisconnected(state);
             MqttConnectReturnCode returnCode = getFailureReturnCode(authResponse);
             unauthorizedClientManager.persistClientUnauthorized(state, sessionCtx, authResponse);
+            // Carry the enhanced-auth failure cause (e.g. AUTH_METHOD_MISMATCH) so CLIENT_AUTHENTICATION_FAILED.reason
+            // describes why authentication failed on both paths, rather than leaking the MQTT CONNACK return-code name here.
+            integrationLifecycleEventPublisher.publishAuthenticationFailed(sessionCtx, state.getClientId(), authResponse.enhancedAuthFailure().name());
             sendConnectionRefusedMsgAndCloseChannel(sessionCtx, returnCode);
             return;
         }

application/src/main/java/org/thingsboard/mqtt/broker/actors/client/service/BrokerInitializer.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.mqtt.broker.actors.client.service;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
@@ -24,11 +25,15 @@
 import org.thingsboard.mqtt.broker.actors.client.service.session.ClientSessionService;
 import org.thingsboard.mqtt.broker.actors.client.service.subscription.ClientSubscriptionService;
 import org.thingsboard.mqtt.broker.common.data.ClientSessionInfo;
+import org.thingsboard.mqtt.broker.common.data.integration.ClientLifecycleEventType;
+import org.thingsboard.mqtt.broker.common.data.integration.ClientLifecycleEventTypeUtil;
+import org.thingsboard.mqtt.broker.common.data.integration.Integration;
 import org.thingsboard.mqtt.broker.common.data.subscription.TopicSubscription;
 import org.thingsboard.mqtt.broker.config.ClientsLimitProperties;
 import org.thingsboard.mqtt.broker.dao.integration.IntegrationService;
 import org.thingsboard.mqtt.broker.exception.QueuePersistenceException;
 import org.thingsboard.mqtt.broker.queue.cluster.ServiceInfoProvider;
+import org.thingsboard.mqtt.broker.service.integration.IntegrationLifecycleEventTypeCache;
 import org.thingsboard.mqtt.broker.service.limits.RateLimitService;
 import org.thingsboard.mqtt.broker.service.mqtt.client.blocked.BlockedClientService;
 import org.thingsboard.mqtt.broker.service.mqtt.client.blocked.consumer.BlockedClientConsumerService;
@@ -50,6 +55,7 @@
 import org.thingsboard.mqtt.broker.service.subscription.data.SubscriptionsSourceKey;
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -74,6 +80,7 @@ public class BrokerInitializer {
     private final ServiceInfoProvider serviceInfoProvider;
     private final RateLimitService rateLimitService;
     private final IntegrationService integrationService;
+    private final IntegrationLifecycleEventTypeCache lifecycleEventTypeCache;
     private final ClientsLimitProperties clientsLimitProperties;
 
     private final ClientSessionEventConsumer clientSessionEventConsumer;
@@ -89,6 +96,7 @@ public class BrokerInitializer {
     public void onApplicationEvent(ApplicationReadyEvent event) {
         log.info("Initializing Client Sessions and Subscriptions.");
         try {
+            initIntegrationLifecycleEventCache();
             Map<String, ClientSessionInfo> allClientSessions = initClientSessions();
             initClientSubscriptions(allClientSessions);
 
@@ -109,6 +117,25 @@ public void onApplicationEvent(ApplicationReadyEvent event) {
         }
     }
 
+    void initIntegrationLifecycleEventCache() {
+        List<Integration> integrations = integrationService.findAllIntegrations();
+        int cached = 0;
+        for (Integration integration : integrations) {
+            JsonNode configuration = integration.getConfiguration();
+            if (configuration == null || !configuration.has(ClientLifecycleEventTypeUtil.LIFECYCLE_EVENT_TYPES_KEY)) {
+                continue;
+            }
+            Set<ClientLifecycleEventType> eventTypes = ClientLifecycleEventTypeUtil.parse(
+                    configuration.get(ClientLifecycleEventTypeUtil.LIFECYCLE_EVENT_TYPES_KEY),
+                    name -> log.warn("[{}] Unknown lifecycle event type: {}", integration.getId(), name));
+            if (!eventTypes.isEmpty()) {
+                lifecycleEventTypeCache.put(integration.getIdStr(), eventTypes);
+                cached++;
+            }
+        }
+        log.info("Loaded lifecycle event type cache: cached {} of {} integrations.", cached, integrations.size());
+    }
+
     Map<String, ClientSessionInfo> initClientSessions() throws QueuePersistenceException {
         Map<String, ClientSessionInfo> allClientSessions = clientSessionConsumer.initLoad();
         log.info("Loaded {} stored client sessions from Kafka.", allClientSessions.size());

application/src/main/java/org/thingsboard/mqtt/broker/actors/client/service/connect/ConnectServiceImpl.java

@@ -57,6 +57,7 @@
 import org.thingsboard.mqtt.broker.service.mqtt.persistence.MsgPersistenceManager;
 import org.thingsboard.mqtt.broker.service.mqtt.validation.PublishMsgValidationService;
 import org.thingsboard.mqtt.broker.service.mqtt.will.LastWillService;
+import org.thingsboard.mqtt.broker.service.integration.IntegrationLifecycleEventPublisher;
 import org.thingsboard.mqtt.broker.service.stats.StatsManager;
 import org.thingsboard.mqtt.broker.service.subscription.ClientSubscriptionCache;
 import org.thingsboard.mqtt.broker.service.subscription.shared.TopicSharedSubscription;
@@ -94,6 +95,7 @@ public class ConnectServiceImpl implements ConnectService {
     private final PublishMsgValidationService publishMsgValidationService;
     private final MqttPublishMsgDeliveryService mqttPublishMsgDeliveryService;
     private final StatsManager statsManager;
+    private final IntegrationLifecycleEventPublisher integrationLifecycleEventPublisher;
 
     private ExecutorService connectHandlerExecutor;
 
@@ -209,6 +211,7 @@ public void acceptConnection(ClientActorStateInfo actorState, ConnectionAccepted
         log.debug("[{}] [{}] Client connected!", actorState.getClientId(), actorState.getCurrentSessionId());
 
         clientSessionCtxService.registerSession(sessionCtx);
+        integrationLifecycleEventPublisher.publishConnected(sessionCtx);
 
         if (sessionCtx.getSessionInfo().isPersistent()) {
             msgPersistenceManager.startProcessingPersistedMessages(actorState);
@@ -243,12 +246,16 @@ private void pushConnAckMsg(ClientActorStateInfo actorState, ConnectionAcceptedM
     void refuseConnection(ClientSessionCtx clientSessionCtx, ClientSessionFailureReason reason, Throwable t) {
         logConnectionRefused(clientSessionCtx, reason, t);
 
-        sendConnectionRefusedMsgAndDisconnect(clientSessionCtx, reason);
+        MqttConnectReturnCode returnCode = reason.toMqttReturnCode(clientSessionCtx);
+        // Emit the same MQTT CONNACK reason-code name the client receives, matching the pre-connection validation
+        // path (which emits MqttConnectReturnCode.name()) so CLIENT_CONNECTION_FAILED speaks a single vocabulary.
+        integrationLifecycleEventPublisher.publishConnectionFailed(clientSessionCtx, clientSessionCtx.getSessionInfo(), returnCode.name());
+
+        sendConnectionRefusedMsgAndDisconnect(clientSessionCtx, returnCode);
     }
 
-    private void sendConnectionRefusedMsgAndDisconnect(ClientSessionCtx ctx, ClientSessionFailureReason reason) {
+    private void sendConnectionRefusedMsgAndDisconnect(ClientSessionCtx ctx, MqttConnectReturnCode mqttReturnCode) {
         try {
-            MqttConnectReturnCode mqttReturnCode = reason.toMqttReturnCode(ctx);
             createAndSendConnAckMsg(mqttReturnCode, ctx);
         } catch (Exception e) {
             log.warn("[{}][{}] Failed to send CONN_ACK response.", ctx.getClientId(), ctx.getSessionId());
@@ -295,6 +302,9 @@ boolean shouldProceedWithConnection(ClientActorStateInfo actorState, MqttConnect
             validateLastWillMessage(ctx, clientId, msg);
         } catch (ConnectionValidationException e) {
             log.warn("[{}] Connection validation failed: {}", ctx.getSessionId(), e.getMessage());
+            // ctx.getSessionInfo() is not set yet at this point, so pass the already-built sessionInfo explicitly.
+            // reason is the MQTT connect return code the client received, e.g. CONNECTION_REFUSED_CLIENT_IDENTIFIER_NOT_VALID.
+            integrationLifecycleEventPublisher.publishConnectionFailed(ctx, sessionInfo, e.getMqttConnectReturnCode().name());
             createAndSendConnAckMsg(e.getMqttConnectReturnCode(), ctx);
             disconnect(clientId, ctx.getSessionId());
             return false;

application/src/main/java/org/thingsboard/mqtt/broker/actors/client/service/disconnect/DisconnectServiceImpl.java

@@ -27,6 +27,7 @@
 import org.thingsboard.mqtt.broker.common.data.ClientInfo;
 import org.thingsboard.mqtt.broker.common.data.SessionInfo;
 import org.thingsboard.mqtt.broker.service.auth.AuthorizationRuleService;
+import org.thingsboard.mqtt.broker.service.integration.IntegrationLifecycleEventPublisher;
 import org.thingsboard.mqtt.broker.service.historical.stats.TbMessageStatsReportClient;
 import org.thingsboard.mqtt.broker.service.limits.RateLimitService;
 import org.thingsboard.mqtt.broker.service.mqtt.MqttMessageGenerator;
@@ -60,6 +61,7 @@ public class DisconnectServiceImpl implements DisconnectService {
     private final FlowControlService flowControlService;
     private final TbMessageStatsReportClient tbMessageStatsReportClient;
     private final ChannelBackpressureManager channelBackpressureManager;
+    private final IntegrationLifecycleEventPublisher integrationLifecycleEventPublisher;
 
     @Override
     public void disconnect(ClientActorStateInfo actorState, MqttDisconnectMsg disconnectMsg) {
@@ -85,6 +87,15 @@ public void disconnect(ClientActorStateInfo actorState, MqttDisconnectMsg discon
         if (reasonType.isNotClusterConflictingSession()) {
             notifyClientDisconnected(actorState, sessionExpiryInterval, reasonType);
         }
+        // Emit the lifecycle CLIENT_DISCONNECTED on every disconnect, including cross-node session takeover
+        // (ON_CLUSTER_CONFLICTING_SESSIONS). The session-event notification above is intentionally suppressed
+        // on takeover, but the lifecycle event must still fire to pair with the CLIENT_CONNECTED this node emitted.
+        // Exception: a broker-refused connection (ON_CONNECTION_FAILURE) never established a session and never
+        // emitted CLIENT_CONNECTED, so its teardown must not emit a phantom CLIENT_DISCONNECTED; the dedicated
+        // CLIENT_CONNECTION_FAILED event covers that case instead.
+        if (reasonType != DisconnectReasonType.ON_CONNECTION_FAILURE) {
+            integrationLifecycleEventPublisher.publishDisconnected(sessionCtx, reasonType);
+        }
         cleanupClientSession(actorState, disconnectMsg, sessionExpiryInterval);
     }
 

application/src/main/java/org/thingsboard/mqtt/broker/actors/client/service/handlers/MqttSubscribeHandler.java

@@ -34,6 +34,8 @@
 import org.thingsboard.mqtt.broker.dao.topic.TopicValidationService;
 import org.thingsboard.mqtt.broker.exception.DataValidationException;
 import org.thingsboard.mqtt.broker.service.auth.AuthorizationRuleService;
+import org.thingsboard.mqtt.broker.service.integration.AuthorizationAction;
+import org.thingsboard.mqtt.broker.service.integration.IntegrationLifecycleEventPublisher;
 import org.thingsboard.mqtt.broker.service.limits.RateLimitService;
 import org.thingsboard.mqtt.broker.service.mqtt.MqttMessageGenerator;
 import org.thingsboard.mqtt.broker.service.mqtt.MqttMsgDeliveryService;
@@ -74,6 +76,7 @@ public class MqttSubscribeHandler {
     private final MsgPersistenceManager msgPersistenceManager;
     private final ApplicationPersistenceProcessor applicationPersistenceProcessor;
     private final RateLimitService rateLimitService;
+    private final IntegrationLifecycleEventPublisher integrationLifecycleEventPublisher;
 
     public void process(ClientSessionCtx ctx, MqttSubscribeMsg msg) {
         Set<TopicSharedSubscription> currentSharedSubscriptions = clientSubscriptionService.getClientSharedSubscriptions(ctx.getClientId());
@@ -132,6 +135,7 @@ List<MqttReasonCodes.SubAck> collectMqttReasonCodes(ClientSessionCtx ctx, MqttSu
                 if (!isClientAuthorized) {
                     log.warn("[{}][{}] Client is not authorized to subscribe to the topic {}",
                             ctx.getClientId(), ctx.getSessionId(), topic);
+                    integrationLifecycleEventPublisher.publishAuthorizationDenied(ctx, AuthorizationAction.SUBSCRIBE, topic);
                     codes.add(MqttReasonCodeResolver.notAuthorizedSubscribe(ctx));
                     continue;
                 }
@@ -162,6 +166,7 @@ private void subscribeAndPersist(ClientSessionCtx ctx, List<TopicSubscription> n
                 CallbackUtil.createCallback(
                         () -> {
                             sendSubAck(ctx, subAckMessage);
+                            integrationLifecycleEventPublisher.publishSubscribed(ctx, newSubscriptions);
                             processRetainedMessages(ctx, newSubscriptions, currentSubscriptions);
                         },
                         t -> log.warn("[{}][{}] Failed to process client subscription.", clientId, ctx.getSessionId(), t))

application/src/main/java/org/thingsboard/mqtt/broker/actors/client/service/handlers/MqttUnsubscribeHandler.java

@@ -24,14 +24,19 @@
 import org.thingsboard.mqtt.broker.actors.client.messages.mqtt.MqttUnsubscribeMsg;
 import org.thingsboard.mqtt.broker.actors.client.service.subscription.ClientSubscriptionService;
 import org.thingsboard.mqtt.broker.adaptor.NettyMqttConverter;
+import org.thingsboard.mqtt.broker.common.data.subscription.TopicSubscription;
 import org.thingsboard.mqtt.broker.common.data.util.CallbackUtil;
+import org.thingsboard.mqtt.broker.service.integration.IntegrationLifecycleEventPublisher;
 import org.thingsboard.mqtt.broker.service.mqtt.MqttMessageGenerator;
 import org.thingsboard.mqtt.broker.service.mqtt.persistence.application.ApplicationPersistenceProcessor;
 import org.thingsboard.mqtt.broker.service.subscription.shared.TopicSharedSubscription;
 import org.thingsboard.mqtt.broker.session.ClientSessionCtx;
 import org.thingsboard.mqtt.broker.util.MqttReasonCodeResolver;
 
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -43,20 +48,59 @@ public class MqttUnsubscribeHandler {
     private final MqttMessageGenerator mqttMessageGenerator;
     private final ClientSubscriptionService clientSubscriptionService;
     private final ApplicationPersistenceProcessor applicationPersistenceProcessor;
+    private final IntegrationLifecycleEventPublisher integrationLifecycleEventPublisher;
 
     public void process(ClientSessionCtx ctx, MqttUnsubscribeMsg msg) {
         log.trace("[{}][{}] Processing unsubscribe, messageId - {}, topic filters - {}", ctx.getClientId(), ctx.getSessionId(), msg.getMessageId(), msg.getTopics());
 
         MqttMessage unSubAckMessage = mqttMessageGenerator.createUnSubAckMessage(msg.getMessageId(), getCodes(ctx, msg));
+        // MQTT allows UNSUBSCRIBE for filters the client never subscribed to. Emit CLIENT_UNSUBSCRIBED only for
+        // the subscriptions actually removed (symmetric with CLIENT_SUBSCRIBED, which emits only the granted ones).
+        List<TopicSubscription> removedSubscriptions = getRemovedSubscriptions(ctx.getClientId(), msg.getTopics());
         clientSubscriptionService.unsubscribeAndPersist(ctx.getClientId(), msg.getTopics(),
                 CallbackUtil.createCallback(
-                        () -> ctx.getChannel().writeAndFlush(unSubAckMessage),
+                        () -> {
+                            ctx.getChannel().writeAndFlush(unSubAckMessage);
+                            if (!removedSubscriptions.isEmpty()) {
+                                integrationLifecycleEventPublisher.publishUnsubscribed(ctx, removedSubscriptions);
+                            }
+                        },
                         t -> log.warn("[{}][{}] Failed to process client unsubscription", ctx.getClientId(), ctx.getSessionId(), t)
                 ));
 
         stopProcessingApplicationSharedSubscriptions(ctx, msg.getTopics());
     }
 
+    private List<TopicSubscription> getRemovedSubscriptions(String clientId, List<String> requestedTopics) {
+        Set<TopicSubscription> currentSubscriptions = clientSubscriptionService.getClientSubscriptions(clientId);
+        if (CollectionUtils.isEmpty(currentSubscriptions)) {
+            return List.of();
+        }
+        // Index current subscriptions by (shareName, bare topic filter) so a requested $share/<group>/<filter>
+        // resolves to that exact shared subscription (carrying its shareName), and a bare filter to the regular one.
+        Map<SubKey, TopicSubscription> byKey = new HashMap<>();
+        for (TopicSubscription sub : currentSubscriptions) {
+            byKey.putIfAbsent(new SubKey(sub.getShareName(), sub.getTopicFilter()), sub);
+        }
+        List<TopicSubscription> removed = new ArrayList<>();
+        for (String requested : requestedTopics) {
+            String shareName = NettyMqttConverter.isSharedTopic(requested) ? NettyMqttConverter.getShareName(requested) : null;
+            TopicSubscription sub = byKey.get(new SubKey(shareName, toTopicFilter(requested)));
+            if (sub != null) {
+                removed.add(sub);
+            }
+        }
+        return removed;
+    }
+
+    // Composite lookup key; shareName is null for a regular subscription, the group name for a shared one.
+    private record SubKey(String shareName, String topicFilter) {
+    }
+
+    private static String toTopicFilter(String topic) {
+        return NettyMqttConverter.isSharedTopic(topic) ? NettyMqttConverter.getTopicFilter(topic) : topic;
+    }
+
     private List<UnsubAck> getCodes(ClientSessionCtx ctx, MqttUnsubscribeMsg msg) {
         return msg
                 .getTopics()