api to create or delete all kafka topics for app shared subs entities

Author: dmytro-landiak

application/src/main/java/org/thingsboard/mqtt/broker/controller/AppSharedSubscriptionController.java

@@ -28,6 +28,7 @@
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 import org.thingsboard.mqtt.broker.common.data.ApplicationSharedSubscription;
+import org.thingsboard.mqtt.broker.common.data.BrokerConstants;
 import org.thingsboard.mqtt.broker.common.data.exception.ThingsboardErrorCode;
 import org.thingsboard.mqtt.broker.common.data.exception.ThingsboardException;
 import org.thingsboard.mqtt.broker.common.data.page.PageData;
@@ -36,6 +37,8 @@
 import org.thingsboard.mqtt.broker.dao.topic.TopicValidationService;
 import org.thingsboard.mqtt.broker.service.mqtt.persistence.application.topic.ApplicationTopicService;
 
+import java.util.function.Consumer;
+
 @RestController
 @RequiredArgsConstructor
 @RequestMapping("/api/app/shared/subs")
@@ -106,4 +109,39 @@ public void deleteSharedSubscription(@PathVariable String id) throws Thingsboard
             applicationTopicService.deleteSharedTopic(sharedSubscription);
         }
     }
+
+    @PreAuthorize("hasAuthority('SYS_ADMIN')")
+    @PostMapping(value = "/createTopics")
+    public void createKafkaTopicsForAppSharedSubscriptions() {
+        processAppSharedSubsRequest(applicationTopicService::createSharedTopic, "create");
+    }
+
+    @PreAuthorize("hasAuthority('SYS_ADMIN')")
+    @DeleteMapping(value = "/deleteTopics")
+    public void deleteKafkaTopicsForAppSharedSubscriptions() {
+        if (!enableTopicDeletion) {
+            log.debug("Cannot delete topics due to TB_KAFKA_ENABLE_TOPIC_DELETION is set to false");
+            return;
+        }
+        processAppSharedSubsRequest(applicationTopicService::deleteSharedTopic, "delete");
+    }
+
+    private void processAppSharedSubsRequest(Consumer<ApplicationSharedSubscription> consumer, String request) {
+        PageLink pageLink = new PageLink(BrokerConstants.DEFAULT_PAGE_SIZE);
+        PageData<ApplicationSharedSubscription> pageData;
+        do {
+            pageData = applicationSharedSubscriptionService.getSharedSubscriptions(pageLink);
+            pageData.getData().forEach(sharedSubscription -> {
+                try {
+                    consumer.accept(sharedSubscription);
+                } catch (Exception e) {
+                    log.error("Failed to {} kafka topic for shared subscription {}", request, sharedSubscription, e);
+                }
+            });
+            if (pageData.hasNext()) {
+                pageLink = pageLink.nextPageLink();
+            }
+        } while (pageData.hasNext());
+    }
+
 }

application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/client/cleanup/ClientSessionCleanUpServiceImpl.java

@@ -51,27 +51,21 @@ public class ClientSessionCleanUpServiceImpl implements ClientSessionCleanUpServ
 
     @Override
     public void removeClientSession(String clientId, UUID sessionId) throws ThingsboardException {
-        if (log.isTraceEnabled()) {
-            log.trace("[{}] Removing ClientSession.", clientId);
-        }
+        log.trace("[{}] Removing ClientSession.", clientId);
         ClientSessionInfo clientSessionInfo = clientSessionCache.getClientSessionInfo(clientId);
         if (clientSessionInfo == null || differentSession(sessionId, clientSessionInfo)) {
             throw new ThingsboardException("No such client session", ThingsboardErrorCode.ITEM_NOT_FOUND);
         }
         if (clientSessionInfo.isConnected()) {
             throw new ThingsboardException("Client is currently connected", ThingsboardErrorCode.GENERAL);
         }
-        if (log.isDebugEnabled()) {
-            log.debug("[{}] Cleaning up client session.", clientId);
-        }
+        log.debug("[{}] Cleaning up client session.", clientId);
         clientSessionEventService.requestSessionCleanup(ClientSessionInfoFactory.clientSessionInfoToSessionInfo(clientSessionInfo));
     }
 
     @Override
     public void disconnectClientSession(String clientId, UUID sessionId) throws ThingsboardException {
-        if (log.isTraceEnabled()) {
-            log.trace("[{}][{}] Disconnecting ClientSession.", clientId, sessionId);
-        }
+        log.trace("[{}][{}] Disconnecting ClientSession.", clientId, sessionId);
         ClientSessionInfo clientSessionInfo = clientSessionCache.getClientSessionInfo(clientId);
         if (clientSessionInfo == null || differentSession(sessionId, clientSessionInfo)) {
             throw new ThingsboardException("No such client session", ThingsboardErrorCode.ITEM_NOT_FOUND);
@@ -85,9 +79,7 @@ public void disconnectClientSession(String clientId, UUID sessionId) throws Thin
 
     @Override
     public void disconnectClientSession(String clientId) {
-        if (log.isTraceEnabled()) {
-            log.trace("[{}] Disconnecting ClientSession", clientId);
-        }
+        log.trace("[{}] Disconnecting ClientSession", clientId);
         ClientSessionInfo clientSessionInfo = clientSessionCache.getClientSessionInfo(clientId);
         if (clientSessionInfo == null || !clientSessionInfo.isConnected()) {
             return;

application/src/main/java/org/thingsboard/mqtt/broker/service/mqtt/persistence/application/topic/ApplicationTopicServiceImpl.java

@@ -46,9 +46,7 @@ public class ApplicationTopicServiceImpl implements ApplicationTopicService {
 
     @Override
     public String createTopic(String clientId) {
-        if (log.isDebugEnabled()) {
-            log.debug("[{}] Creating APPLICATION topic", clientId);
-        }
+        log.debug("[{}] Creating APPLICATION topic", clientId);
         String clientTopic = appClientHelperService.getAppTopic(clientId, validateClientId);
         queueAdmin.createTopic(clientTopic, applicationPersistenceMsgQueueFactory.getTopicConfigs());
         return clientTopic;
@@ -57,9 +55,7 @@ public String createTopic(String clientId) {
     @Override
     public void createSharedTopic(ApplicationSharedSubscription subscription) {
         String topic = subscription.getTopicFilter();
-        if (log.isDebugEnabled()) {
-            log.debug("[{}] Creating shared APPLICATION topic", topic);
-        }
+        log.debug("[{}] Creating shared APPLICATION topic", topic);
 
         final var topicToCreate = appClientHelperService.getSharedAppTopic(topic, validateSharedTopicFilter);
 
@@ -70,9 +66,7 @@ public void createSharedTopic(ApplicationSharedSubscription subscription) {
 
     @Override
     public void deleteTopic(String clientId, BasicCallback callback) {
-        if (log.isDebugEnabled()) {
-            log.debug("[{}] Deleting APPLICATION topic", clientId);
-        }
+        log.debug("[{}] Deleting APPLICATION topic", clientId);
         String clientTopic = appClientHelperService.getAppTopic(clientId, validateClientId);
         queueAdmin.deleteTopic(clientTopic, callback);
         String consumerGroup = appClientHelperService.getAppConsumerGroup(clientId);
@@ -82,9 +76,7 @@ public void deleteTopic(String clientId, BasicCallback callback) {
     @Override
     public void deleteSharedTopic(ApplicationSharedSubscription subscription) {
         String topic = subscription.getTopicFilter();
-        if (log.isDebugEnabled()) {
-            log.debug("[{}] Deleting shared APPLICATION topic", topic);
-        }
+        log.debug("[{}] Deleting shared APPLICATION topic", topic);
 
         final var topicToDelete = appClientHelperService.getSharedAppTopic(topic, validateSharedTopicFilter);