From 879936b03f51bc6b32b33d77cb5f71a02ddc12a4 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Fri, 26 Jun 2026 18:18:15 +0300 Subject: [PATCH] refactor(rpc): address review feedback on batched RPC persistence - Rename TbRpcService.save -> update so the create/update pairing is self-documenting (matches createAsync/updateAsync); update all actor call sites and unit tests. - Extract the (hash & 0x7FFFFFFF) % n striping into a shared HashPartitioner helper used by both TbSqlBlockingQueueWrapper and TbRpcService, so the end-to-end submission-order invariant has a single source of truth. - Collapse the insert/update split in RpcInsertRepository.saveOrUpdate into a single partitioningBy pass. - Remove the unused TbRpcService.findRpcById (and its now-orphaned import). - Restore the prime-thread-count guidance on the RPC batch_threads / callback_threads yml comments. --- .../device/DeviceActorMessageProcessor.java | 10 ++--- .../server/service/rpc/TbRpcService.java | 10 ++--- .../src/main/resources/thingsboard.yml | 4 +- .../server/service/rpc/TbRpcServiceTest.java | 8 ++-- .../common/util/HashPartitioner.java | 41 +++++++++++++++++++ .../dao/sql/TbSqlBlockingQueueWrapper.java | 3 +- .../dao/sql/rpc/RpcInsertRepository.java | 8 +++- 7 files changed, 63 insertions(+), 21 deletions(-) create mode 100644 common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 6f257405af..6d0f07e507 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -356,7 +356,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso UUID rpcId = toDeviceRpcRequest.getId(); log.debug("[{}][{}][{}] RPC request timeout detected!", deviceId, rpcId, requestId); if (toDeviceRpcRequest.isPersisted()) { - systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.EXPIRED, null)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.EXPIRED, null)); } systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION)); @@ -667,7 +667,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } catch (IllegalArgumentException e) { response = JacksonUtil.newObjectNode().put("error", payload); } - systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRequestMsg, status, response)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRequestMsg, status, response)); } } finally { if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE)) { @@ -731,7 +731,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } if (persisted) { - systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRpcRequest, status, response)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, status, response)); } if (rpcSubmitStrategy.equals(RpcSubmitStrategy.SEQUENTIAL_ON_RESPONSE_FROM_DEVICE) && status.equals(RpcStatus.DELIVERED) && !oneWayRpc) { @@ -824,7 +824,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso var toDeviceRpcRequest = md.getMsg().getMsg(); if (toDeviceRpcRequest.isPersisted()) { var responseAwaitTimeout = JacksonUtil.newObjectNode().put("error", "There was a timeout awaiting for RPC response from device."); - systemContext.getTbRpcService().save(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout)); + systemContext.getTbRpcService().update(tenantId, buildRpc(toDeviceRpcRequest, RpcStatus.FAILED, responseAwaitTimeout)); } }, systemContext.getRpcResponseTimeout(), TimeUnit.MILLISECONDS); } @@ -1035,7 +1035,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso long timeout = rpc.getExpirationTime() - System.currentTimeMillis(); if (timeout <= 0) { rpc.setStatus(RpcStatus.EXPIRED); - systemContext.getTbRpcService().save(tenantId, rpc); + systemContext.getTbRpcService().update(tenantId, rpc); } else { registerPendingRpcRequest(ctx, new ToDeviceRpcRequestActorMsg(systemContext.getServiceId(), msg), false, createToDeviceRpcRequestMsg(msg), timeout); } diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java index da90db7846..e2f1e00fb1 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java @@ -21,11 +21,11 @@ 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.HashPartitioner; 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; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.page.PageData; @@ -79,7 +79,7 @@ public class TbRpcService { persist(tenantId, rpc, rpcService.createAsync(rpc)); } - public void save(TenantId tenantId, Rpc rpc) { + public void update(TenantId tenantId, Rpc rpc) { persist(tenantId, rpc, rpcService.updateAsync(rpc)); } @@ -99,7 +99,7 @@ public class TbRpcService { } private Executor executorFor(UUID rpcId) { - return callbackExecutors[(rpcId.hashCode() & 0x7FFFFFFF) % callbackExecutors.length]; + return callbackExecutors[HashPartitioner.resolvePartition(rpcId.hashCode(), callbackExecutors.length)]; } private void pushRpcMsgToRuleEngine(TenantId tenantId, Rpc rpc) { @@ -112,10 +112,6 @@ public class TbRpcService { tbClusterService.pushMsgToRuleEngine(tenantId, rpc.getDeviceId(), msg, null); } - public Rpc findRpcById(TenantId tenantId, RpcId rpcId) { - return rpcService.findById(tenantId, rpcId); - } - public PageData findAllByDeviceIdAndStatus(TenantId tenantId, DeviceId deviceId, RpcStatus rpcStatus, PageLink pageLink) { return rpcService.findAllByDeviceIdAndStatus(tenantId, deviceId, rpcStatus, pageLink); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 19075e7305..af1b7d328d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -428,8 +428,8 @@ sql: batch_size: "${SQL_RPC_BATCH_SIZE:1000}" # Batch size for persisting RPC inserts/updates batch_max_delay: "${SQL_RPC_BATCH_MAX_DELAY_MS:50}" # Max timeout for RPC entries queue polling, in milliseconds stats_print_interval_ms: "${SQL_RPC_BATCH_STATS_PRINT_MS:10000}" # Interval in milliseconds for printing RPC persistence statistic - batch_threads: "${SQL_RPC_BATCH_THREADS:3}" # number of queue partitions for batched RPC persistence; writes for a given RPC always map to the same partition (by rpcId hash) so they persist in submission order - callback_threads: "${SQL_RPC_CALLBACK_THREADS:3}" # striped threads for post-persist rule-engine notifications; each rpcId maps to a single stripe, preserving per-command notification order + batch_threads: "${SQL_RPC_BATCH_THREADS:3}" # number of queue partitions for batched RPC persistence; writes for a given RPC always map to the same partition (by rpcId hash) so they persist in submission order. A prime value (e.g. 3 or 5) keeps the hash distribution even + callback_threads: "${SQL_RPC_CALLBACK_THREADS:3}" # striped threads for post-persist rule-engine notifications; each rpcId maps to a single stripe, preserving per-command notification order. A prime value (e.g. 3 or 5) keeps the hash distribution even events: batch_size: "${SQL_EVENTS_BATCH_SIZE:10000}" # Batch size for persisting latest telemetry updates batch_max_delay: "${SQL_EVENTS_BATCH_MAX_DELAY_MS:100}" # Max timeout for latest telemetry entries queue polling. The value set in milliseconds diff --git a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java index 87162c1123..cf11dfed48 100644 --- a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -51,11 +51,11 @@ public class TbRpcServiceTest { } @Test - public void savePersistsViaUpdateAsyncThenPushesToRuleEngine() { + public void updatePersistsViaUpdateAsyncThenPushesToRuleEngine() { Rpc rpc = newRpc(); when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(true)); - tbRpcService.save(rpc.getTenantId(), rpc); + tbRpcService.update(rpc.getTenantId(), rpc); verify(rpcService).updateAsync(rpc); verify(clusterService, timeout(5000)) @@ -75,13 +75,13 @@ public class TbRpcServiceTest { } @Test - public void saveDoesNotNotifyRuleEngineWhenRowMissing() { + public void updateDoesNotNotifyRuleEngineWhenRowMissing() { Rpc rpc = newRpc(); // updateAsync resolves false: the UPDATE matched no row (RPC was deleted), so the rule engine // must not be notified for a status change that never persisted. when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(false)); - tbRpcService.save(rpc.getTenantId(), rpc); + tbRpcService.update(rpc.getTenantId(), rpc); verify(rpcService).updateAsync(rpc); verify(clusterService, after(500).never()) diff --git a/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java b/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java new file mode 100644 index 0000000000..54555b2639 --- /dev/null +++ b/common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java @@ -0,0 +1,41 @@ +/** + * 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.common.util; + +/** + * Maps a hash code to a non-negative partition index in {@code [0, partitions)}. + *

+ * Centralises the {@code (hashCode & 0x7FFFFFFF) % partitions} striping so that components which must + * agree on a partition for the same key derive it identically. In particular, batched RPC persistence + * relies on this: {@code TbSqlBlockingQueueWrapper} picks the DB write partition and {@code TbRpcService} + * picks the post-persist callback stripe, both keyed off the rpcId hash. If those two ever disagreed on + * how the index is computed, the submission-order guarantee (e.g. RPC_QUEUED before RPC_DELIVERED) would + * silently break — keeping the math in one place removes that risk. + */ +public final class HashPartitioner { + + private HashPartitioner() { + } + + /** + * @param hashCode hash of the partition key + * @param partitions number of partitions; must be {@code > 0} + * @return a non-negative partition index in {@code [0, partitions)} + */ + public static int resolvePartition(int hashCode, int partitions) { + return (hashCode & 0x7FFFFFFF) % partitions; + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java b/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java index 4ae8661d77..1dab15b473 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java @@ -18,6 +18,7 @@ package org.thingsboard.server.dao.sql; import com.google.common.util.concurrent.ListenableFuture; import lombok.Data; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.HashPartitioner; import org.thingsboard.server.common.stats.MessagesStats; import org.thingsboard.server.common.stats.StatsFactory; @@ -58,7 +59,7 @@ public class TbSqlBlockingQueueWrapper { } public ListenableFuture add(E element) { - int queueIndex = element != null ? (hashCodeFunction.apply(element) & 0x7FFFFFFF) % maxThreads : 0; + int queueIndex = element != null ? HashPartitioner.resolvePartition(hashCodeFunction.apply(element), maxThreads) : 0; return queues.get(queueIndex).add(element); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java index 9d0eb45464..08fa1c7969 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java @@ -26,6 +26,8 @@ import java.sql.PreparedStatement; import java.sql.SQLException; import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; @Repository public class RpcInsertRepository extends AbstractInsertRepository { @@ -50,8 +52,10 @@ public class RpcInsertRepository extends AbstractInsertRepository { } public List saveOrUpdate(List entries) { - List inserts = entries.stream().filter(RpcQueueEntry::insert).map(RpcQueueEntry::entity).toList(); - List updates = entries.stream().filter(entry -> !entry.insert()).map(RpcQueueEntry::entity).toList(); + Map> byIntent = entries.stream().collect(Collectors.partitioningBy( + RpcQueueEntry::insert, Collectors.mapping(RpcQueueEntry::entity, Collectors.toList()))); + List inserts = byIntent.get(true); + List updates = byIntent.get(false); int[] updateCounts = transactionTemplate.execute(status -> { // Inserts run first so a create and a status update for the same rpcId coalesced into one // batch still apply in create -> update order.