From 6de4b30e5ff22848cbab5cb26eb951c5c6b685c5 Mon Sep 17 00:00:00 2001 From: Nikita Mazurenko Date: Tue, 4 Aug 2026 16:01:34 +0300 Subject: [PATCH] Unblock and interrupt edge event consumer on session destroy --- .../server/service/edge/rpc/EdgeGrpcSession.java | 2 +- .../server/service/edge/rpc/KafkaEdgeGrpcSession.java | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) 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 fb420d080a..1e9a0ade87 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 @@ -760,7 +760,7 @@ public abstract class EdgeGrpcSession implements Closeable { ctx.getClusterService().onEdgeEventUpdate(new EdgeEventUpdateMsg(edge.getTenantId(), edge.getId())); } - private void stopCurrentSendDownlinkMsgsTask(Boolean isInterrupted) { + protected void stopCurrentSendDownlinkMsgsTask(Boolean isInterrupted) { if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone()) { sessionState.getSendDownlinkMsgsFuture().set(isInterrupted); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java index 67c0a19e66..e1acc08ba3 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java @@ -156,6 +156,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession { @Override public boolean destroy() { + stopCurrentSendDownlinkMsgsTask(true); try { if (consumer != null) { log.info("[{}][{}] Stopping edge event consumer...", tenantId, edge != null ? edge.getId() : null); @@ -180,9 +181,13 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession { private void awaitConsumerTermination() { try { - consumerExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS); + if (!consumerExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) { + // without this the thread is leaked for the whole remaining delivery-retry window (or forever, if the future is never completed). + consumerExecutor.shutdownNow(); + } } catch (InterruptedException ie) { log.warn("[{}][{}] Interrupted while awaiting consumer executor termination", tenantId, edge.getId()); + Thread.currentThread().interrupt(); } }