From b1b0e44c7fda9118cea95b6715430cf1641943bc Mon Sep 17 00:00:00 2001
From: Dmytro Skarzhynets <d.skarzh@protonmail.com>
Date: Fri, 26 Jan 2024 17:04:49 +0200
Subject: [PATCH] Add executor to process activity events in a dedicated thread
 not to avoid blocking main consumer

---
 .../queue/DefaultTbCoreConsumerService.java   | 64 +++++++++++--------
 .../DefaultTbCoreConsumerServiceTest.java     | 23 +++++++
 2 files changed, 59 insertions(+), 28 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
index 5c93351d0a..9942476c01 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
@@ -15,6 +15,9 @@
  */
 package org.thingsboard.server.service.queue;
 
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
@@ -148,6 +151,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
     protected volatile ExecutorService consumersExecutor;
     protected volatile ExecutorService usageStatsExecutor;
     private volatile ExecutorService firmwareStatesExecutor;
+    private volatile ListeningExecutorService deviceActivityEventsExecutor;
 
     public DefaultTbCoreConsumerService(TbCoreQueueFactory tbCoreQueueFactory,
                                         ActorSystemContext actorContext,
@@ -195,6 +199,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         this.consumersExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("tb-core-consumer"));
         this.usageStatsExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-core-usage-stats-consumer"));
         this.firmwareStatesExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-core-firmware-notifications-consumer"));
+        this.deviceActivityEventsExecutor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-core-device-activity-events-consumer")));
     }
 
     @PreDestroy
@@ -209,6 +214,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         if (firmwareStatesExecutor != null) {
             firmwareStatesExecutor.shutdownNow();
         }
+        if (deviceActivityEventsExecutor != null) {
+            deviceActivityEventsExecutor.shutdownNow();
+        }
     }
 
     @AfterStartUp(order = AfterStartUp.REGULAR_SERVICE)
@@ -665,13 +673,13 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         }
         var tenantId = toTenantId(deviceConnectMsg.getTenantIdMSB(), deviceConnectMsg.getTenantIdLSB());
         var deviceId = new DeviceId(new UUID(deviceConnectMsg.getDeviceIdMSB(), deviceConnectMsg.getDeviceIdLSB()));
-        try {
-            stateService.onDeviceConnect(tenantId, deviceId, deviceConnectMsg.getLastConnectTime());
-            callback.onSuccess();
-        } catch (Exception e) {
-            log.warn("[{}] Failed to process device connect message for device [{}]", tenantId.getId(), deviceId.getId(), e);
-            callback.onFailure(e);
-        }
+        ListenableFuture<?> future = deviceActivityEventsExecutor.submit(() -> stateService.onDeviceConnect(tenantId, deviceId, deviceConnectMsg.getLastConnectTime()));
+        DonAsynchron.withCallback(future,
+                __ -> callback.onSuccess(),
+                t -> {
+                    log.warn("[{}] Failed to process device connect message for device [{}]", tenantId.getId(), deviceId.getId(), t);
+                    callback.onFailure(t);
+                });
     }
 
     void forwardToStateService(TransportProtos.DeviceActivityProto deviceActivityMsg, TbCallback callback) {
@@ -680,13 +688,13 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         }
         var tenantId = toTenantId(deviceActivityMsg.getTenantIdMSB(), deviceActivityMsg.getTenantIdLSB());
         var deviceId = new DeviceId(new UUID(deviceActivityMsg.getDeviceIdMSB(), deviceActivityMsg.getDeviceIdLSB()));
-        try {
-            stateService.onDeviceActivity(tenantId, deviceId, deviceActivityMsg.getLastActivityTime());
-            callback.onSuccess();
-        } catch (Exception e) {
-            log.warn("[{}] Failed to process device activity message for device [{}]", tenantId.getId(), deviceId.getId(), e);
-            callback.onFailure(new RuntimeException("Failed to update device activity for device [" + deviceId.getId() + "]!", e));
-        }
+        ListenableFuture<?> future = deviceActivityEventsExecutor.submit(() -> stateService.onDeviceActivity(tenantId, deviceId, deviceActivityMsg.getLastActivityTime()));
+        DonAsynchron.withCallback(future,
+                __ -> callback.onSuccess(),
+                t -> {
+                    log.warn("[{}] Failed to process device activity message for device [{}]", tenantId.getId(), deviceId.getId(), t);
+                    callback.onFailure(new RuntimeException("Failed to update device activity for device [" + deviceId.getId() + "]!", t));
+                });
     }
 
     void forwardToStateService(TransportProtos.DeviceDisconnectProto deviceDisconnectMsg, TbCallback callback) {
@@ -695,13 +703,13 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         }
         var tenantId = toTenantId(deviceDisconnectMsg.getTenantIdMSB(), deviceDisconnectMsg.getTenantIdLSB());
         var deviceId = new DeviceId(new UUID(deviceDisconnectMsg.getDeviceIdMSB(), deviceDisconnectMsg.getDeviceIdLSB()));
-        try {
-            stateService.onDeviceDisconnect(tenantId, deviceId, deviceDisconnectMsg.getLastDisconnectTime());
-            callback.onSuccess();
-        } catch (Exception e) {
-            log.warn("[{}] Failed to process device activity message for device [{}]", tenantId.getId(), deviceId.getId(), e);
-            callback.onFailure(e);
-        }
+        ListenableFuture<?> future = deviceActivityEventsExecutor.submit(() -> stateService.onDeviceDisconnect(tenantId, deviceId, deviceDisconnectMsg.getLastDisconnectTime()));
+        DonAsynchron.withCallback(future,
+                __ -> callback.onSuccess(),
+                t -> {
+                    log.warn("[{}] Failed to process device disconnect message for device [{}]", tenantId.getId(), deviceId.getId(), t);
+                    callback.onFailure(t);
+                });
     }
 
     void forwardToStateService(TransportProtos.DeviceInactivityProto deviceInactivityMsg, TbCallback callback) {
@@ -710,13 +718,13 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         }
         var tenantId = toTenantId(deviceInactivityMsg.getTenantIdMSB(), deviceInactivityMsg.getTenantIdLSB());
         var deviceId = new DeviceId(new UUID(deviceInactivityMsg.getDeviceIdMSB(), deviceInactivityMsg.getDeviceIdLSB()));
-        try {
-            stateService.onDeviceInactivity(tenantId, deviceId, deviceInactivityMsg.getLastInactivityTime());
-            callback.onSuccess();
-        } catch (Exception e) {
-            log.warn("[{}] Failed to process device inactivity message for device [{}]", tenantId.getId(), deviceId.getId(), e);
-            callback.onFailure(e);
-        }
+        ListenableFuture<?> future = deviceActivityEventsExecutor.submit(() -> stateService.onDeviceInactivity(tenantId, deviceId, deviceInactivityMsg.getLastInactivityTime()));
+        DonAsynchron.withCallback(future,
+                __ -> callback.onSuccess(),
+                t -> {
+                    log.warn("[{}] Failed to process device inactivity message for device [{}]", tenantId.getId(), deviceId.getId(), t);
+                    callback.onFailure(t);
+                });
     }
 
     private void forwardToNotificationSchedulerService(TransportProtos.NotificationSchedulerServiceMsg msg, TbCallback callback) {
diff --git a/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerServiceTest.java b/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerServiceTest.java
index 19166f75d2..621d4c5a91 100644
--- a/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerServiceTest.java
+++ b/application/src/test/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerServiceTest.java
@@ -15,6 +15,9 @@
  */
 package org.thingsboard.server.service.queue;
 
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -29,6 +32,7 @@ import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.service.state.DeviceStateService;
 
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -52,12 +56,31 @@ public class DefaultTbCoreConsumerServiceTest {
     private final DeviceId deviceId = new DeviceId(UUID.randomUUID());
     private final long time = System.currentTimeMillis();
 
+    private ListeningExecutorService executor;
+
     @Mock
     private DefaultTbCoreConsumerService defaultTbCoreConsumerServiceMock;
 
     @BeforeEach
     public void setup() {
+        executor = MoreExecutors.newDirectExecutorService();
         ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stateService", stateServiceMock);
+        ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "deviceActivityEventsExecutor", executor);
+    }
+
+    @AfterEach
+    public void cleanup() {
+        if (executor != null) {
+            executor.shutdown();
+            try {
+                if (!executor.awaitTermination(10L, TimeUnit.SECONDS)) {
+                    executor.shutdownNow();
+                }
+            } catch (InterruptedException e) {
+                executor.shutdownNow();
+                Thread.currentThread().interrupt();
+            }
+        }
     }
 
     @Test