Browse Source

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.
pull/15853/head
dshvaika 1 month ago
parent
commit
879936b03f
  1. 10
      application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
  2. 10
      application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java
  3. 4
      application/src/main/resources/thingsboard.yml
  4. 8
      application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java
  5. 41
      common/util/src/main/java/org/thingsboard/common/util/HashPartitioner.java
  6. 3
      dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java
  7. 8
      dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java

10
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);
}

10
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<Rpc> findAllByDeviceIdAndStatus(TenantId tenantId, DeviceId deviceId, RpcStatus rpcStatus, PageLink pageLink) {
return rpcService.findAllByDeviceIdAndStatus(tenantId, deviceId, rpcStatus, pageLink);
}

4
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

8
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())

41
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)}.
* <p>
* 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;
}
}

3
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<E, R> {
}
public ListenableFuture<R> 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);
}

8
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<Boolean> saveOrUpdate(List<RpcQueueEntry> entries) {
List<RpcEntity> inserts = entries.stream().filter(RpcQueueEntry::insert).map(RpcQueueEntry::entity).toList();
List<RpcEntity> updates = entries.stream().filter(entry -> !entry.insert()).map(RpcQueueEntry::entity).toList();
Map<Boolean, List<RpcEntity>> byIntent = entries.stream().collect(Collectors.partitioningBy(
RpcQueueEntry::insert, Collectors.mapping(RpcQueueEntry::entity, Collectors.toList())));
List<RpcEntity> inserts = byIntent.get(true);
List<RpcEntity> 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.

Loading…
Cancel
Save