feat(rpc): async RPC persistence with striped post-flush rule-engine notification

application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java

@@ -16,10 +16,16 @@
 package org.thingsboard.server.service.rpc;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.util.concurrent.ListenableFuture;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.DonAsynchron;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.RpcId;
@@ -34,6 +40,11 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.dao.rpc.RpcService;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 
+import java.util.UUID;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 @TbCoreComponent
 @Service
 @RequiredArgsConstructor
@@ -42,10 +53,40 @@ public class TbRpcService {
     private final RpcService rpcService;
     private final TbClusterService tbClusterService;
 
-    public Rpc save(TenantId tenantId, Rpc rpc) {
+    @Value("${sql.rpc.callback_threads:3}")
-        Rpc saved = rpcService.save(rpc);
+    private int callbackThreads;
-        pushRpcMsgToRuleEngine(tenantId, saved);
+
-        return saved;
+    private ExecutorService[] callbackExecutors;
+
+    @PostConstruct
+    private void init() {
+        callbackExecutors = new ExecutorService[callbackThreads];
+        for (int i = 0; i < callbackThreads; i++) {
+            callbackExecutors[i] = Executors.newSingleThreadExecutor(
+                    ThingsBoardThreadFactory.forName("rpc-persist-callback-" + i));
+        }
+    }
+
+    @PreDestroy
+    private void destroy() {
+        if (callbackExecutors != null) {
+            for (ExecutorService executor : callbackExecutors) {
+                executor.shutdownNow();
+            }
+        }
+    }
+
+    public void save(TenantId tenantId, Rpc rpc) {
+        ListenableFuture<Void> future = rpcService.saveAsync(rpc);
+        DonAsynchron.withCallback(future,
+                v -> pushRpcMsgToRuleEngine(tenantId, rpc),
+                t -> log.error("[{}][{}][{}] Failed to persist RPC with status [{}]",
+                        tenantId, rpc.getDeviceId(), rpc.getId(), rpc.getStatus(), t),
+                executorFor(rpc.getUuidId()));
+    }
+
+    private Executor executorFor(UUID rpcId) {
+        return callbackExecutors[(rpcId.hashCode() & 0x7FFFFFFF) % callbackThreads];
     }
 
     public void save(TenantId tenantId, RpcId rpcId, RpcStatus newStatus, JsonNode response) {

application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java

@@ -0,0 +1,70 @@
+/**
+ * Copyright © 2016-2026 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.service.rpc;
+
+import com.google.common.util.concurrent.Futures;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.cluster.TbClusterService;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.RpcId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.rpc.Rpc;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.dao.rpc.RpcService;
+
+import java.util.UUID;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TbRpcServiceTest {
+
+    private final RpcService rpcService = mock(RpcService.class);
+    private final TbClusterService clusterService = mock(TbClusterService.class);
+    private TbRpcService tbRpcService;
+
+    @Before
+    public void setUp() {
+        tbRpcService = new TbRpcService(rpcService, clusterService);
+        ReflectionTestUtils.setField(tbRpcService, "callbackThreads", 1);
+        ReflectionTestUtils.invokeMethod(tbRpcService, "init");
+    }
+
+    @Test
+    public void savePushesToRuleEngineAfterFlush() {
+        Rpc rpc = new Rpc(new RpcId(UUID.randomUUID()));
+        rpc.setTenantId(TenantId.SYS_TENANT_ID);
+        rpc.setDeviceId(new DeviceId(UUID.randomUUID()));
+        rpc.setStatus(RpcStatus.QUEUED);
+        rpc.setRequest(JacksonUtil.toJsonNode("{}"));
+
+        when(rpcService.saveAsync(rpc)).thenReturn(Futures.immediateFuture(null));
+
+        tbRpcService.save(rpc.getTenantId(), rpc);
+
+        verify(clusterService, timeout(5000))
+                .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull());
+    }
+}