test(rpc): cover same-rpcId coalesced-batch ordering; clarify callback_threads comment

application/src/main/resources/thingsboard.yml

@@ -429,7 +429,7 @@ sql:
     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}" # batch thread count has to be a prime number like 3 or 5 to gain perfect hash distribution
-    callback_threads: "${SQL_RPC_CALLBACK_THREADS:3}" # striped threads for post-persist rule-engine notifications; keep equal to batch_threads for aligned ordering
+    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
   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

dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java

@@ -93,4 +93,42 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest {
         assertThat(afterUpdate.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}"));
     }
 
+    @Test
+    public void saveAsyncSameRpcIdCoalescedBatchKeepsLatestStatus() throws Exception {
+        UUID id = UUID.randomUUID();
+        DeviceId deviceId = new DeviceId(UUID.randomUUID());
+        long createdTime = System.currentTimeMillis();
+        long expirationTime = createdTime + 60_000;
+
+        Rpc queued = new Rpc(new RpcId(id));
+        queued.setCreatedTime(createdTime);
+        queued.setTenantId(TenantId.SYS_TENANT_ID);
+        queued.setDeviceId(deviceId);
+        queued.setExpirationTime(expirationTime);
+        queued.setRequest(JacksonUtil.toJsonNode("{\"method\":\"x\"}"));
+        queued.setStatus(RpcStatus.QUEUED);
+
+        Rpc delivered = new Rpc(new RpcId(id));
+        delivered.setCreatedTime(createdTime);
+        delivered.setTenantId(TenantId.SYS_TENANT_ID);
+        delivered.setDeviceId(deviceId);
+        delivered.setExpirationTime(expirationTime);
+        delivered.setRequest(queued.getRequest());
+        delivered.setStatus(RpcStatus.DELIVERED);
+        delivered.setResponse(JacksonUtil.toJsonNode("{\"ok\":true}"));
+
+        // Enqueue both writes for the same rpcId back-to-back so they coalesce into one flush batch.
+        // Same rpcId -> same partition; the queue's stable sort must keep submission order
+        // (QUEUED before DELIVERED), so the final persisted row must be DELIVERED, never QUEUED.
+        var queuedFuture = rpcDao.saveAsync(TenantId.SYS_TENANT_ID, queued);
+        var deliveredFuture = rpcDao.saveAsync(TenantId.SYS_TENANT_ID, delivered);
+        queuedFuture.get(5, TimeUnit.SECONDS);
+        deliveredFuture.get(5, TimeUnit.SECONDS);
+
+        Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id);
+        assertThat(stored).isNotNull();
+        assertThat(stored.getStatus()).isEqualTo(RpcStatus.DELIVERED);
+        assertThat(stored.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}"));
+    }
+
 }