From 6f7c165ff5d78082a831a6dfd53324fd915a7db3 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Tue, 7 Jul 2026 13:17:48 +0300 Subject: [PATCH] refactor(rpc): drop unused batched-insert path The initial RPC create is now persisted synchronously before sending (persist-before-send); only status updates remain on the async batched queue. The batched-insert plumbing that was retained for reversibility is now dead code and is removed: - remove createAsync from RpcService/RpcDao/BaseRpcService/JpaRpcDao - delete RpcQueueEntry (insert/update discriminator no longer needed) - rename RpcInsertRepository -> RpcUpdateRepository; the queue now carries RpcEntity directly and batches UPDATEs only - re-aim JpaRpcDaoTest / TbRpcServiceTest off the removed insert API No behavior change: update batching, uuid partitioning/sort, and the response COALESCE preservation are all unchanged. --- .../src/main/resources/thingsboard.yml | 2 +- .../server/service/rpc/TbRpcServiceTest.java | 3 +- .../server/dao/rpc/RpcService.java | 2 - .../server/dao/rpc/BaseRpcService.java | 6 -- .../thingsboard/server/dao/rpc/RpcDao.java | 2 - .../server/dao/sql/rpc/JpaRpcDao.java | 17 ++-- .../dao/sql/rpc/RpcInsertRepository.java | 99 ------------------- .../server/dao/sql/rpc/RpcQueueEntry.java | 29 ------ .../dao/sql/rpc/RpcUpdateRepository.java | 64 ++++++++++++ .../server/dao/sql/rpc/JpaRpcDaoTest.java | 74 ++++++-------- 10 files changed, 100 insertions(+), 198 deletions(-) delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java delete mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 5cda04a33d..c0ad755176 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -425,7 +425,7 @@ sql: batch_threads: "${SQL_TS_LATEST_BATCH_THREADS:3}" # batch thread count has to be a prime number like 3 or 5 to gain perfect hash distribution update_by_latest_ts: "${SQL_TS_UPDATE_BY_LATEST_TIMESTAMP:true}" # Update latest values only if the timestamp of the new record is greater or equals the timestamp of the previously saved latest value. The latest values are stored separately from historical values for fast lookup from DB. Insert of historical value happens in any case rpc: - batch_size: "${SQL_RPC_BATCH_SIZE:1000}" # Batch size for persisting RPC inserts/updates + batch_size: "${SQL_RPC_BATCH_SIZE:1000}" # Batch size for persisting RPC status 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. A prime value (e.g. 3 or 5) keeps the hash distribution even 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 513b667c3c..bdb194e1de 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 @@ -78,9 +78,8 @@ public class TbRpcServiceTest { tbRpcService.create(rpc.getTenantId(), rpc); - // create must persist synchronously via save(...), never via the async batch path + // create must persist synchronously via save(...) verify(rpcService).save(rpc); - verify(rpcService, never()).createAsync(any()); verify(clusterService, timeout(5000)) .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); } diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java index aaac66c86b..19defc8eea 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/rpc/RpcService.java @@ -29,8 +29,6 @@ public interface RpcService extends EntityDaoService { Rpc save(Rpc rpc); - ListenableFuture createAsync(Rpc rpc); - ListenableFuture updateAsync(Rpc rpc); void deleteRpc(TenantId tenantId, RpcId id); diff --git a/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java b/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java index 759b90f3a2..44a265b26d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/rpc/BaseRpcService.java @@ -54,12 +54,6 @@ public class BaseRpcService implements RpcService { return rpcDao.save(rpc.getTenantId(), rpc); } - @Override - public ListenableFuture createAsync(Rpc rpc) { - log.trace("Executing createAsync, [{}]", rpc); - return rpcDao.createAsync(rpc); - } - @Override public ListenableFuture updateAsync(Rpc rpc) { log.trace("Executing updateAsync, [{}]", rpc); diff --git a/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java index b0dde0d969..c3999b2a11 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/rpc/RpcDao.java @@ -26,8 +26,6 @@ import org.thingsboard.server.dao.Dao; public interface RpcDao extends Dao { - ListenableFuture createAsync(Rpc rpc); - ListenableFuture updateAsync(Rpc rpc); PageData findAllByDeviceId(TenantId tenantId, DeviceId deviceId, PageLink pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java index 8752d58a78..2340e4b64a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDao.java @@ -54,7 +54,7 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, @Autowired private RpcRepository rpcRepository; @Autowired - private RpcInsertRepository rpcInsertRepository; + private RpcUpdateRepository rpcUpdateRepository; @Autowired private ScheduledLogExecutorComponent logExecutor; @Autowired @@ -71,7 +71,7 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, @Value("${sql.batch_sort:true}") private boolean batchSortEnabled; - private TbSqlBlockingQueueWrapper queue; + private TbSqlBlockingQueueWrapper queue; @PostConstruct private void init() { @@ -84,10 +84,10 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, .batchSortEnabled(batchSortEnabled) .withResponse(true) .build(); - Function hashcodeFunction = entry -> entry.entity().getUuid().hashCode(); + Function hashcodeFunction = entity -> entity.getUuid().hashCode(); queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory); - queue.init(logExecutor, entries -> rpcInsertRepository.saveOrUpdate(entries), - Comparator.comparing((RpcQueueEntry entry) -> entry.entity().getUuid()), + queue.init(logExecutor, entries -> rpcUpdateRepository.update(entries), + Comparator.comparing(RpcEntity::getUuid), Function.identity()); } @@ -98,14 +98,9 @@ public class JpaRpcDao extends JpaAbstractDao implements RpcDao, } } - @Override - public ListenableFuture createAsync(Rpc rpc) { - return queue.add(RpcQueueEntry.forInsert(new RpcEntity(rpc))); - } - @Override public ListenableFuture updateAsync(Rpc rpc) { - return queue.add(RpcQueueEntry.forUpdate(new RpcEntity(rpc))); + return queue.add(new RpcEntity(rpc)); } @Override 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 deleted file mode 100644 index 6b39f990ae..0000000000 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcInsertRepository.java +++ /dev/null @@ -1,99 +0,0 @@ -/** - * 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.dao.sql.rpc; - -import com.fasterxml.jackson.databind.JsonNode; -import org.springframework.jdbc.core.BatchPreparedStatementSetter; -import org.springframework.stereotype.Repository; -import org.thingsboard.common.util.JacksonUtil; -import org.thingsboard.server.dao.model.sql.RpcEntity; -import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository; - -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 { - - private static final String INSERT = - "INSERT INTO rpc (id, created_time, tenant_id, device_id, expiration_time, request, response, additional_info, status) " + - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " + - "ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, response = COALESCE(EXCLUDED.response, rpc.response);"; - - private static final String UPDATE = - "UPDATE rpc SET status = ?, response = COALESCE(?, response) WHERE id = ?;"; - - @FunctionalInterface - private interface ColumnBinder { - void bind(PreparedStatement ps, RpcEntity rpc) throws SQLException; - } - - List saveOrUpdate(List entries) { - 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); - return transactionTemplate.execute(status -> { - if (!inserts.isEmpty()) { - batch(INSERT, inserts, (ps, rpc) -> { - ps.setObject(1, rpc.getUuid()); - ps.setLong(2, rpc.getCreatedTime()); - ps.setObject(3, rpc.getTenantId()); - ps.setObject(4, rpc.getDeviceId()); - ps.setLong(5, rpc.getExpirationTime()); - ps.setString(6, toJsonStr(rpc.getRequest())); - ps.setString(7, toJsonStr(rpc.getResponse())); - ps.setString(8, toJsonStr(rpc.getAdditionalInfo())); - ps.setString(9, rpc.getStatus().name()); - }); - } - int[] updateCounts = updates.isEmpty() ? new int[0] : batch(UPDATE, updates, (ps, rpc) -> { - ps.setString(1, rpc.getStatus().name()); - ps.setString(2, toJsonStr(rpc.getResponse())); - ps.setObject(3, rpc.getUuid()); - }); - - List persisted = new ArrayList<>(entries.size()); - int updateIdx = 0; - for (RpcQueueEntry entry : entries) { - persisted.add(entry.insert() || updateCounts[updateIdx++] > 0); - } - return persisted; - }); - } - - private int[] batch(String sql, List entities, ColumnBinder binder) { - return jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() { - @Override - public void setValues(PreparedStatement ps, int i) throws SQLException { - binder.bind(ps, entities.get(i)); - } - - @Override - public int getBatchSize() { - return entities.size(); - } - }); - } - - private String toJsonStr(JsonNode node) { - return node == null ? null : replaceNullChars(JacksonUtil.toString(node)); - } -} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java deleted file mode 100644 index a74fe7a2b4..0000000000 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcQueueEntry.java +++ /dev/null @@ -1,29 +0,0 @@ -/** - * 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.dao.sql.rpc; - -import org.thingsboard.server.dao.model.sql.RpcEntity; - -record RpcQueueEntry(RpcEntity entity, boolean insert) { - - static RpcQueueEntry forInsert(RpcEntity entity) { - return new RpcQueueEntry(entity, true); - } - - static RpcQueueEntry forUpdate(RpcEntity entity) { - return new RpcQueueEntry(entity, false); - } -} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java new file mode 100644 index 0000000000..9403e551fb --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rpc/RpcUpdateRepository.java @@ -0,0 +1,64 @@ +/** + * 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.dao.sql.rpc; + +import com.fasterxml.jackson.databind.JsonNode; +import org.springframework.jdbc.core.BatchPreparedStatementSetter; +import org.springframework.stereotype.Repository; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.dao.model.sql.RpcEntity; +import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository; + +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; + +@Repository +public class RpcUpdateRepository extends AbstractInsertRepository { + + private static final String UPDATE = + "UPDATE rpc SET status = ?, response = COALESCE(?, response) WHERE id = ?;"; + + List update(List updates) { + return transactionTemplate.execute(status -> { + int[] updateCounts = jdbcTemplate.batchUpdate(UPDATE, new BatchPreparedStatementSetter() { + @Override + public void setValues(PreparedStatement ps, int i) throws SQLException { + RpcEntity rpc = updates.get(i); + ps.setString(1, rpc.getStatus().name()); + ps.setString(2, toJsonStr(rpc.getResponse())); + ps.setObject(3, rpc.getUuid()); + } + + @Override + public int getBatchSize() { + return updates.size(); + } + }); + + List persisted = new ArrayList<>(updateCounts.length); + for (int updateCount : updateCounts) { + persisted.add(updateCount > 0); + } + return persisted; + }); + } + + private String toJsonStr(JsonNode node) { + return node == null ? null : replaceNullChars(JacksonUtil.toString(node)); + } +} diff --git a/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java b/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java index e2df9db08e..35dbfabfb3 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/sql/rpc/JpaRpcDaoTest.java @@ -39,7 +39,7 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { JpaRpcDao rpcDao; @Autowired - RpcInsertRepository rpcInsertRepository; + RpcUpdateRepository rpcUpdateRepository; @Test public void deleteOutdated() { @@ -66,19 +66,19 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { } @Test - public void saveAsyncInsertThenUpsert() throws Exception { + public void syncCreateThenAsyncUpdateConvergesToUpdatedStatus() throws Exception { UUID id = UUID.randomUUID(); DeviceId deviceId = new DeviceId(UUID.randomUUID()); - // A create always persists (INSERT ... ON CONFLICT), so the future resolves true. - assertThat(rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS)).isTrue(); + // Production create path: the QUEUED row is persisted synchronously (persist-before-send). + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); Rpc stored = rpcDao.findById(TenantId.SYS_TENANT_ID, id); assertThat(stored).isNotNull(); assertThat(stored.getStatus()).isEqualTo(RpcStatus.QUEUED); assertThat(stored.getResponse()).isNull(); - // The update matches the existing row, so the future resolves true. + // The async status update matches the existing row, so the future resolves true. assertThat(rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.DELIVERED, JacksonUtil.toJsonNode("{\"ok\":true}"))) .get(5, TimeUnit.SECONDS)).isTrue(); @@ -88,52 +88,34 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { } @Test - public void saveAsyncSequentialWritesConvergeToLatestStatus() throws Exception { - UUID id = UUID.randomUUID(); + public void updateBatchAppliesInOrderAndAlignsResults() { DeviceId deviceId = new DeviceId(UUID.randomUUID()); + UUID idA = UUID.randomUUID(); + UUID idB = UUID.randomUUID(); // never created -> its update must report no match - // A create followed by a status update for the same rpcId, both via the async API. Whether the - // two land in one flush batch or two is up to the queue's timing and not asserted here (see - // saveOrUpdateCoalescedBatchAppliesInOrderAndAlignsResults for the deterministic coalesced case); - // either way the final persisted row must converge to DELIVERED, never get stuck at QUEUED. - var queuedFuture = rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)); - var deliveredFuture = rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.DELIVERED, JacksonUtil.toJsonNode("{\"ok\":true}"))); - queuedFuture.get(5, TimeUnit.SECONDS); - deliveredFuture.get(5, TimeUnit.SECONDS); + // Row A exists (persisted synchronously at create time); B was never created. + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(idA, deviceId, RpcStatus.QUEUED, null)); - 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}")); - } + // Drive the persist logic directly with a single, deterministically-coalesced update batch. + // This is exactly what "coalescing" means: one update() call carrying several writes for the + // same partition in submission order. No queue timing involved. + // index 0: update A (SUCCESSFUL, {ok:true}) -> UPDATE hits the existing row -> true + // index 1: update B (SUCCESSFUL, {x:1}) -> UPDATE for a missing row -> false + // index 2: update A (EXPIRED, null response) -> UPDATE hits A, keeps response -> true + List batch = List.of( + new RpcEntity(rpc(idA, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))), + new RpcEntity(rpc(idB, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"x\":1}"))), + new RpcEntity(rpc(idA, deviceId, RpcStatus.EXPIRED, null))); - @Test - public void saveOrUpdateCoalescedBatchAppliesInOrderAndAlignsResults() { - DeviceId deviceId = new DeviceId(UUID.randomUUID()); - UUID idA = UUID.randomUUID(); - UUID idB = UUID.randomUUID(); // never inserted -> its update must report no match - - // Drive the persist logic directly with a single, deterministically-coalesced flush batch. - // This is exactly what "coalescing" means: one saveOrUpdate call carrying several writes for - // the same partition in submission order. No queue timing involved. - // index 0: create A (QUEUED) -> INSERT, always persists -> true - // index 1: update B (SUCCESSFUL) -> UPDATE for a missing row -> false - // index 2: update A (DELIVERED, {ok:true}) -> UPDATE on the row inserted at index 0 -> true - List batch = List.of( - RpcQueueEntry.forInsert(new RpcEntity(rpc(idA, deviceId, RpcStatus.QUEUED, null))), - RpcQueueEntry.forUpdate(new RpcEntity(rpc(idB, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"x\":1}")))), - RpcQueueEntry.forUpdate(new RpcEntity(rpc(idA, deviceId, RpcStatus.DELIVERED, JacksonUtil.toJsonNode("{\"ok\":true}"))))); - - List persisted = rpcInsertRepository.saveOrUpdate(batch); - - // Booleans are aligned positionally to submission order even though saveOrUpdate runs all - // inserts before all updates internally - this guards the updateIdx cursor alignment. + List persisted = rpcUpdateRepository.update(batch); + + // Booleans are aligned positionally to submission order. assertThat(persisted).containsExactly(true, false, true); - // Inserts run before updates, so A ends DELIVERED (never stuck at QUEUED)... + // Updates apply in order, so A ends EXPIRED, and the null-response update kept the stored response. Rpc storedA = rpcDao.findById(TenantId.SYS_TENANT_ID, idA); assertThat(storedA).isNotNull(); - assertThat(storedA.getStatus()).isEqualTo(RpcStatus.DELIVERED); + assertThat(storedA.getStatus()).isEqualTo(RpcStatus.EXPIRED); assertThat(storedA.getResponse()).isEqualTo(JacksonUtil.toJsonNode("{\"ok\":true}")); // ...and the update for a never-created row neither persisted nor resurrected it. assertThat(rpcDao.findById(TenantId.SYS_TENANT_ID, idB)).isNull(); @@ -145,7 +127,7 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { DeviceId deviceId = new DeviceId(UUID.randomUUID()); // Initial create. - rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); // Delivery timeout with closeTransportSessionOnRpcDeliveryTimeout=true re-queues the RPC: the // device actor persists status=QUEUED again as a status update so init() can re-pick it up. @@ -162,7 +144,7 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { UUID id = UUID.randomUUID(); DeviceId deviceId = new DeviceId(UUID.randomUUID()); - rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); // A successful response is stored. rpcDao.updateAsync(rpc(id, deviceId, RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode("{\"ok\":true}"))) @@ -181,7 +163,7 @@ public class JpaRpcDaoTest extends AbstractJpaDaoTest { UUID id = UUID.randomUUID(); DeviceId deviceId = new DeviceId(UUID.randomUUID()); - rpcDao.createAsync(rpc(id, deviceId, RpcStatus.QUEUED, null)).get(5, TimeUnit.SECONDS); + rpcDao.saveAndFlush(TenantId.SYS_TENANT_ID, rpc(id, deviceId, RpcStatus.QUEUED, null)); // RPC is removed (TTL cleanup / manual delete) while a response is still in flight. rpcDao.removeById(TenantId.SYS_TENANT_ID, id);