diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 8ada917234..9571319226 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -187,10 +187,7 @@ public final class EdgeGrpcSession implements Closeable { public void startSyncProcess(TenantId tenantId, EdgeId edgeId, boolean fullSync) { log.trace("[{}][{}] Staring edge sync process", tenantId, edgeId); syncCompleted = false; - if (sessionState.getSendDownlinkMsgsFuture() != null && sessionState.getSendDownlinkMsgsFuture().isDone()) { - String errorMsg = String.format("[%s][%s] Sync process started. General processing interrupted!", tenantId, edgeId); - sessionState.getSendDownlinkMsgsFuture().setException(new RuntimeException(errorMsg)); - } + interruptGeneralProcessingOnSync(tenantId, edgeId); doSync(new EdgeSyncCursor(ctx, edge, fullSync)); } @@ -265,10 +262,7 @@ public final class EdgeGrpcSession implements Closeable { } if (sessionState.getPendingMsgsMap().isEmpty()) { log.debug("[{}] Pending msgs map is empty. Stopping current iteration", edge.getRoutingKey()); - if (sessionState.getScheduledSendDownlinkTask() != null) { - sessionState.getScheduledSendDownlinkTask().cancel(false); - } - sessionState.getSendDownlinkMsgsFuture().set(null); + stopCurrentSendDownlinkMsgsTask(null); } } catch (Exception e) { log.error("[{}] Can't process downlink response message [{}]", this.sessionId, msg, e); @@ -391,15 +385,14 @@ public final class EdgeGrpcSession implements Closeable { } private ListenableFuture sendDownlinkMsgsPack(List downlinkMsgsPack) { - if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone()) { - String errorMsg = "[" + this.sessionId + "] Previous send downlink future was not properly completed, stopping it now"; - log.error(errorMsg); - sessionState.getSendDownlinkMsgsFuture().setException(new RuntimeException(errorMsg)); - } + interruptPreviousSendDownlinkMsgsTask(); + sessionState.setSendDownlinkMsgsFuture(SettableFuture.create()); sessionState.getPendingMsgsMap().clear(); + downlinkMsgsPack.forEach(msg -> sessionState.getPendingMsgsMap().put(msg.getDownlinkMsgId(), msg)); scheduleDownlinkMsgsPackSend(1); + return sessionState.getSendDownlinkMsgsFuture(); } @@ -422,13 +415,13 @@ public final class EdgeGrpcSession implements Closeable { } else { log.warn("[{}] Failed to deliver the batch after {} attempts. Next messages are going to be discarded {}", this.sessionId, MAX_DOWNLINK_ATTEMPTS, copy); - sessionState.getSendDownlinkMsgsFuture().set(null); + stopCurrentSendDownlinkMsgsTask(null); } } else { - sessionState.getSendDownlinkMsgsFuture().set(null); + stopCurrentSendDownlinkMsgsTask(null); } } catch (Exception e) { - sessionState.getSendDownlinkMsgsFuture().setException(e); + stopCurrentSendDownlinkMsgsTask(e); } }; @@ -673,4 +666,29 @@ public final class EdgeGrpcSession implements Closeable { log.debug("[{}] Failed to close output stream: {}", sessionId, e.getMessage()); } } + + private void interruptPreviousSendDownlinkMsgsTask() { + String msg = String.format("[%s] Previous send downlink future was not properly completed, stopping it now!", this.sessionId); + stopCurrentSendDownlinkMsgsTask(new RuntimeException(msg)); + } + + private void interruptGeneralProcessingOnSync(TenantId tenantId, EdgeId edgeId) { + String msg = String.format("[%s][%s] Sync process started. General processing interrupted!", tenantId, edgeId); + stopCurrentSendDownlinkMsgsTask(new RuntimeException(msg)); + } + + public void stopCurrentSendDownlinkMsgsTask(Exception e) { + if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone()) { + if (e != null) { + log.warn(e.getMessage(), e); + sessionState.getSendDownlinkMsgsFuture().setException(e); + } else { + sessionState.getSendDownlinkMsgsFuture().set(null); + } + } + if (sessionState.getScheduledSendDownlinkTask() != null) { + sessionState.getScheduledSendDownlinkTask().cancel(true); + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSessionState.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSessionState.java index 98673c9cf5..5a3d8dc0f8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSessionState.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeSessionState.java @@ -19,6 +19,7 @@ import com.google.common.util.concurrent.SettableFuture; import lombok.Data; import org.thingsboard.server.gen.edge.v1.DownlinkMsg; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; import java.util.concurrent.ScheduledFuture; @@ -26,7 +27,7 @@ import java.util.concurrent.ScheduledFuture; @Data public class EdgeSessionState { - private final Map pendingMsgsMap = new LinkedHashMap<>(); + private final Map pendingMsgsMap = Collections.synchronizedMap(new LinkedHashMap<>()); private SettableFuture sendDownlinkMsgsFuture; private ScheduledFuture scheduledSendDownlinkTask; } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 2b0adf0557..6a00775a38 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -930,7 +930,7 @@ edges: storage: max_read_records_count: "${EDGES_STORAGE_MAX_READ_RECORDS_COUNT:50}" no_read_records_sleep: "${EDGES_NO_READ_RECORDS_SLEEP:1000}" - sleep_between_batches: "${EDGES_SLEEP_BETWEEN_BATCHES:1000}" + sleep_between_batches: "${EDGES_SLEEP_BETWEEN_BATCHES:10000}" scheduler_pool_size: "${EDGES_SCHEDULER_POOL_SIZE:1}" send_scheduler_pool_size: "${EDGES_SEND_SCHEDULER_POOL_SIZE:1}" grpc_callback_thread_pool_size: "${EDGES_GRPC_CALLBACK_POOL_SIZE:1}" diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java index ffe21ae961..a3162ebc5f 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java @@ -439,9 +439,9 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest { Assert.assertTrue(edgeImitator.waitForResponses()); Assert.assertTrue(edgeImitator.waitForMessages()); - AbstractMessage latestMessage = edgeImitator.getMessageFromTail(2); - Assert.assertTrue(latestMessage instanceof DeviceUpdateMsg); - DeviceUpdateMsg latestDeviceUpdateMsg = (DeviceUpdateMsg) latestMessage; + Optional deviceUpdateMsgOpt = edgeImitator.findMessageByType(DeviceUpdateMsg.class); + Assert.assertTrue(deviceUpdateMsgOpt.isPresent()); + DeviceUpdateMsg latestDeviceUpdateMsg = deviceUpdateMsgOpt.get(); Assert.assertNotEquals(deviceOnCloudName, latestDeviceUpdateMsg.getName()); Assert.assertEquals(deviceOnCloudName, latestDeviceUpdateMsg.getConflictName()); @@ -453,9 +453,9 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest { Assert.assertNotNull(device); Assert.assertNotEquals(deviceOnCloudName, device.getName()); - latestMessage = edgeImitator.getLatestMessage(); - Assert.assertTrue(latestMessage instanceof DeviceCredentialsRequestMsg); - DeviceCredentialsRequestMsg latestDeviceCredentialsRequestMsg = (DeviceCredentialsRequestMsg) latestMessage; + Optional deviceCredentialsUpdateMsgOpt = edgeImitator.findMessageByType(DeviceCredentialsRequestMsg.class); + Assert.assertTrue(deviceCredentialsUpdateMsgOpt.isPresent()); + DeviceCredentialsRequestMsg latestDeviceCredentialsRequestMsg = deviceCredentialsUpdateMsgOpt.get(); Assert.assertEquals(uuid.getMostSignificantBits(), latestDeviceCredentialsRequestMsg.getDeviceIdMSB()); Assert.assertEquals(uuid.getLeastSignificantBits(), latestDeviceCredentialsRequestMsg.getDeviceIdLSB()); diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java index 25280de666..8930166396 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java @@ -34,7 +34,7 @@ abstract public class BaseTelemetryEdgeTest extends AbstractEdgeTest { @Test public void testTimeseriesWithFailures() throws Exception { - int numberOfTimeseriesToSend = 1000; + int numberOfTimeseriesToSend = 333; Device device = findDeviceByName("Edge Device 1"); diff --git a/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java b/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java index 668f702a87..4fa3b5e723 100644 --- a/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java +++ b/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java @@ -365,11 +365,7 @@ public class EdgeImitator { } public AbstractMessage getLatestMessage() { - return getMessageFromTail(1); - } - - public AbstractMessage getMessageFromTail(int offset) { - return downlinkMsgs.get(downlinkMsgs.size() - offset); + return downlinkMsgs.get(downlinkMsgs.size() - 1); } public void ignoreType(Class type) {