From 131ba74514db43ad037c0239adc06de97b35fa47 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 11 Jun 2026 16:29:37 +0300 Subject: [PATCH 1/3] Add an opt-in readinessCheck gate so the edge consumer pauses polling instead of polling and dropping events while not ready (sync/high-priority/disconnected). --- .../KafkaBasedEdgeGrpcSessionManager.java | 32 ++- .../common/consumer/QueueConsumerManager.java | 31 ++- .../consumer/QueueConsumerManagerTest.java | 251 ++++++++++++++++++ 3 files changed, 308 insertions(+), 6 deletions(-) create mode 100644 common/queue/src/test/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManagerTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java index ebc5531320..01174423c2 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java @@ -35,7 +35,6 @@ import org.thingsboard.server.queue.discovery.TopicService; import org.thingsboard.server.queue.kafka.KafkaAdmin; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.queue.util.TbKafkaComponent; - import org.thingsboard.server.service.edge.rpc.EdgeSessionState; import org.thingsboard.server.service.edge.rpc.processor.PostgresGeneralEdgeEventsDispatcher; import org.thingsboard.server.service.edge.rpc.session.EdgeSessionsHolder; @@ -143,8 +142,7 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan } } catch (Exception e) { log.warn("[{}] Failed to process edge events for edge [{}]!", tenantId, edgeId, e); - } - finally { + } finally { initLock.unlock(); } }, ctx.getEdgeEventStorageSettings().getNoRecordsSleepInterval(), TimeUnit.MILLISECONDS); @@ -176,7 +174,7 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan } private boolean initAndLaunchConsumer(TenantId tenantId, EdgeId edgeId, EdgeSessionState state) { - if (!state.isConnected() || state.isSyncInProgress() || isHighPriorityProcessing) { + if (!isSessionReady(state)) { log.warn("[{}][{}] Session is not ready (connected={}, syncInProgress={}, highPriority={}), skip starting edge event consumer", tenantId, edgeId, state.isConnected(), state.isSyncInProgress(), isHighPriorityProcessing); return false; @@ -205,6 +203,7 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan .consumerCreator(() -> tbCoreQueueFactory.createEdgeEventMsgConsumer(tenantId, edgeId)) .consumerExecutor(consumerExecutor) .threadPrefix("edge-events-" + edgeId) + .readinessCheck(this::isReadyToProcessGeneralEvents) .build(); consumer.subscribe(); consumer.launch(); @@ -224,7 +223,9 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan EdgeId edgeId = state.getEdgeId(); log.trace("[{}][{}] starting processing edge events", tenantId, edgeId); - if (!state.isConnected() || state.isSyncInProgress() || isHighPriorityProcessing) { + // Defensive backstop: the loop already gates polling on readiness; this only fires on the narrow race + // where readiness flips during poll(), and that already-polled batch is dropped here (can't rewind). + if (!isSessionReady(state)) { log.debug("[{}][{}] edge not connected, edge sync is not completed or high priority processing in progress, " + "connected = {}, sync in progress = {}, high priority in progress = {}. Skipping iteration", tenantId, edgeId, state.isConnected(), state.isSyncInProgress(), isHighPriorityProcessing); @@ -250,6 +251,26 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan } } + /** + * Readiness gate for the edge-event consumer (see {@link QueueConsumerManager}'s {@code readinessCheck}): it polls + * only while the session is connected and no sync or high-priority processing is running. + *

+ * Pausing polling in those windows is deliberate. A batch that is polled but then skipped advances the Kafka + * position without committing, so it is lost until a rebalance; keeping events queued instead matches the no-loss + * behaviour the Postgres-based manager gets by re-reading by seqId. This covers the common case - a batch already + * in flight when a sync starts is a known residual (closing it would need seek/rewind, which {@code TbQueueConsumer} + * lacks). If a sync ever outran {@code max.poll.interval.ms} (default 5 min) the consumer is rebalanced and resumes + * from the committed offset: still no loss, just a possible replay. Edge syncs are seconds, so this is acceptable. + */ + private boolean isReadyToProcessGeneralEvents() { + return isSessionReady(getState()); + } + + /** Single source of truth for "the session may process general edge events" - see {@link #isReadyToProcessGeneralEvents}. */ + private boolean isSessionReady(EdgeSessionState state) { + return state != null && state.isConnected() && !state.isSyncInProgress() && !isHighPriorityProcessing; + } + private void cancelMigrationAndProcessingInit() { EdgeSessionState state = getState(); log.trace("[{}] cancelling edge migration & processing init for edge", state != null ? state.getEdgeId() : "unknown"); @@ -299,4 +320,5 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan Thread.currentThread().interrupt(); } } + } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java index 4adc0354c4..9ee6bc0b50 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManager.java @@ -30,6 +30,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.BooleanSupplier; import java.util.function.Supplier; @Slf4j @@ -40,6 +41,8 @@ public class QueueConsumerManager { private final long pollInterval; private final ExecutorService consumerExecutor; private final String threadPrefix; + /** Optional poll gate: while {@code false} the loop skips polling so the position doesn't advance; {@code null} = always ready (default). */ + private final BooleanSupplier readinessCheck; @Getter private final TbQueueConsumer consumer; @@ -49,12 +52,13 @@ public class QueueConsumerManager { @Builder public QueueConsumerManager(String name, MsgPackProcessor msgPackProcessor, long pollInterval, Supplier> consumerCreator, - ExecutorService consumerExecutor, String threadPrefix) { + ExecutorService consumerExecutor, String threadPrefix, BooleanSupplier readinessCheck) { this.name = name; this.pollInterval = pollInterval; this.msgPackProcessor = msgPackProcessor; this.consumerExecutor = consumerExecutor; this.threadPrefix = threadPrefix; + this.readinessCheck = readinessCheck; this.consumer = consumerCreator.get(); } @@ -84,6 +88,12 @@ public class QueueConsumerManager { private void consumerLoop(TbQueueConsumer consumer) { while (!stopped && !consumer.isStopped()) { try { + if (!isReadyToProcess()) { + if (!awaitNextReadinessCheck()) { + return; + } + continue; + } List msgs = consumer.poll(pollInterval); if (msgs.isEmpty()) { continue; @@ -102,6 +112,25 @@ public class QueueConsumerManager { } } + private boolean isReadyToProcess() { + return readinessCheck == null || readinessCheck.getAsBoolean(); + } + + /** + * Waits one poll interval before readiness is re-checked. Returns {@code false} if interrupted, which is treated as + * a stop signal so the consumer loop exits. + */ + private boolean awaitNextReadinessCheck() { + log.trace("[{}] Consumer is not ready to process messages yet, skipping poll iteration", name); + try { + Thread.sleep(pollInterval); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + public void stop() { log.debug("[{}] Stopping consumer", name); stopped = true; diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManagerTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManagerTest.java new file mode 100644 index 0000000000..c987fb502e --- /dev/null +++ b/common/queue/src/test/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManagerTest.java @@ -0,0 +1,251 @@ +/** + * 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.queue.common.consumer; + +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; + +import java.util.Collections; +import java.util.List; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.mockito.Mockito.mock; + +@Slf4j +class QueueConsumerManagerTest { + + private static final long POLL_INTERVAL_MS = 20L; + // Before asserting the consumer never polled we wait until the loop has evaluated the readiness gate at least + // this many times. That proves the consumer thread is actually running and deciding not to poll, rather than the + // assertion passing vacuously because the thread simply has not started yet. + private static final int MIN_READINESS_CHECKS = 3; + + private final AtomicBoolean readyToProcess = new AtomicBoolean(false); + private final AtomicInteger readinessChecks = new AtomicInteger(); + private final TestQueueConsumer consumer = new TestQueueConsumer(); + private ExecutorService consumerExecutor; + private QueueConsumerManager manager; + + @AfterEach + void tearDown() { + if (manager != null) { + manager.stop(); + } + if (consumerExecutor != null) { + consumerExecutor.shutdownNow(); + } + } + + @Test + void eventQueuedWhileNotReadyIsDeliveredAfterReadinessGateOpensInsteadOfBeingDropped() { + List delivered = new CopyOnWriteArrayList<>(); + + consumer.enqueue(List.of(mock(TbQueueMsg.class))); + + // The processor is unconditional: only the readiness gate may hold the event back, so delivery proves the + // gate (not the processor) is what kept the event queued while not ready. + manager = launchManager(consumer, countingReadiness(readyToProcess, readinessChecks), (msgs, c) -> { + delivered.addAll(msgs); + c.commit(); + }); + + // The loop is running and repeatedly evaluating the gate during the not-ready (sync) window... + awaitReadinessGateEvaluated(readinessChecks); + // ...yet the queued event is neither polled nor delivered - it stays in the queue rather than being dropped. + assertThat(consumer.getPollCount()) + .as("consumer must not poll while not ready") + .isZero(); + assertThat(delivered) + .as("event must not be delivered while not ready") + .isEmpty(); + + // Sync completes - the processor becomes ready. + readyToProcess.set(true); + + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(delivered) + .as("event queued during the not-ready window must be delivered, not dropped") + .hasSize(1)); + } + + @Test + void consumerIsNotPolledWhileNotReadyToProcess() { + manager = launchManager(consumer, countingReadiness(readyToProcess, readinessChecks), (msgs, c) -> c.commit()); + + awaitReadinessGateEvaluated(readinessChecks); + assertThat(consumer.getPollCount()) + .as("consumer must not be polled while not ready to process") + .isZero(); + + readyToProcess.set(true); + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(consumer.getPollCount()) + .as("consumer resumes polling once ready") + .isPositive()); + } + + @Test + void consumerWithoutReadinessCheckPollsAndDeliversImmediately() { + List delivered = new CopyOnWriteArrayList<>(); + + consumer.enqueue(List.of(mock(TbQueueMsg.class))); + + // No readiness gate configured - the consumer must default to "always ready", preserving the behaviour every + // consumer that does not opt in relies on. + manager = launchManager(consumer, null, (msgs, c) -> { + delivered.addAll(msgs); + c.commit(); + }); + + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(delivered) + .as("consumer without a readiness gate must poll and deliver immediately") + .hasSize(1)); + } + + @Test + void consumerLoopExitsWhenInterruptedWhileNotReady() throws Exception { + manager = launchManager(consumer, countingReadiness(readyToProcess, readinessChecks), (msgs, c) -> c.commit()); + + // The loop is parked in the not-ready wait... + awaitReadinessGateEvaluated(readinessChecks); + + // ...interrupting the worker (as shutdownNow does on stop) must end the loop, not spin or hang. + consumerExecutor.shutdownNow(); + assertThat(consumerExecutor.awaitTermination(5, TimeUnit.SECONDS)) + .as("consumer loop must exit when interrupted while waiting to become ready") + .isTrue(); + } + + private static void awaitReadinessGateEvaluated(AtomicInteger readinessChecks) { + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(readinessChecks.get()) + .as("consumer loop must be running and repeatedly evaluating the readiness gate") + .isGreaterThanOrEqualTo(MIN_READINESS_CHECKS)); + } + + private static BooleanSupplier countingReadiness(AtomicBoolean ready, AtomicInteger readinessChecks) { + return () -> { + readinessChecks.incrementAndGet(); + return ready.get(); + }; + } + + private QueueConsumerManager launchManager(TestQueueConsumer consumer, BooleanSupplier readinessCheck, + QueueConsumerManager.MsgPackProcessor processor) { + consumerExecutor = Executors.newSingleThreadExecutor(); + QueueConsumerManager queueConsumerManager = QueueConsumerManager.builder() + .name("test-consumer") + .pollInterval(POLL_INTERVAL_MS) + .consumerCreator(() -> consumer) + .consumerExecutor(consumerExecutor) + .readinessCheck(readinessCheck) + .msgPackProcessor(processor) + .build(); + queueConsumerManager.subscribe(); + queueConsumerManager.launch(); + return queueConsumerManager; + } + + private static class TestQueueConsumer implements TbQueueConsumer { + + private final Queue> batches = new ConcurrentLinkedQueue<>(); + private final AtomicInteger pollCount = new AtomicInteger(); + private volatile boolean stopped; + + void enqueue(List batch) { + batches.add(batch); + } + + int getPollCount() { + return pollCount.get(); + } + + @Override + public List poll(long durationInMillis) { + pollCount.incrementAndGet(); + List batch = batches.poll(); + if (batch != null) { + return batch; + } + try { + Thread.sleep(durationInMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return Collections.emptyList(); + } + + @Override + public String getTopic() { + return "test-topic"; + } + + @Override + public void subscribe() { + } + + @Override + public void subscribe(Set partitions) { + } + + @Override + public void stop() { + stopped = true; + } + + @Override + public void unsubscribe() { + stopped = true; + } + + @Override + public void commit() { + } + + @Override + public boolean isStopped() { + return stopped; + } + + @Override + public Set getPartitions() { + return Collections.emptySet(); + } + + @Override + public List getFullTopicNames() { + return Collections.emptyList(); + } + + } + +} From 28f25d5331693cfb768fde7a700e084fd3f4aa8c Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 11 Jun 2026 16:29:37 +0300 Subject: [PATCH 2/3] KafkaBasedEdgeGrpcSessionManagerTest: added --- .../KafkaBasedEdgeGrpcSessionManagerTest.java | 322 ++++++++++++++++++ 1 file changed, 322 insertions(+) create mode 100644 application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java new file mode 100644 index 0000000000..0759a6b05e --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java @@ -0,0 +1,322 @@ +/** + * 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.service.edge.rpc.session.manager; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.id.EdgeId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.gen.transport.TransportProtos.ToEdgeEventNotificationMsg; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.common.consumer.QueueConsumerManager; +import org.thingsboard.server.queue.discovery.TopicService; +import org.thingsboard.server.queue.kafka.KafkaAdmin; +import org.thingsboard.server.queue.provider.TbCoreQueueFactory; +import org.thingsboard.server.service.edge.EdgeContextComponent; +import org.thingsboard.server.service.edge.rpc.DownlinkMessageMapper; +import org.thingsboard.server.service.edge.rpc.EdgeEventStorageSettings; +import org.thingsboard.server.service.edge.rpc.EdgeSessionState; +import org.thingsboard.server.service.edge.rpc.session.EdgeSession; +import org.thingsboard.server.service.edge.rpc.session.EdgeSessionsHolder; + +import java.util.Collections; +import java.util.List; +import java.util.Queue; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Unit tests for the readiness gate that {@link KafkaBasedEdgeGrpcSessionManager} feeds into its edge-event consumer. + *

+ * The generic poll-gate mechanism is covered by {@code QueueConsumerManagerTest}; these tests pin the edge-specific + * half of the fix: the manager's readiness predicate, and the fact that the consumer is actually wired with it - so + * dropping the {@code .readinessCheck(...)} builder line would silently reintroduce the event-loss bug. + */ +class KafkaBasedEdgeGrpcSessionManagerTest { + + private static final long POLL_INTERVAL_MS = 20L; + + private EdgeContextComponent ctx; + private TbCoreQueueFactory tbCoreQueueFactory; + private EdgeSessionState state; + private KafkaBasedEdgeGrpcSessionManager manager; + + @BeforeEach + void setUp() { + ctx = mock(EdgeContextComponent.class); + tbCoreQueueFactory = mock(TbCoreQueueFactory.class); + TopicService topicService = mock(TopicService.class); + KafkaAdmin kafkaAdmin = mock(KafkaAdmin.class); + EdgeSessionsHolder sessions = mock(EdgeSessionsHolder.class); + + manager = new KafkaBasedEdgeGrpcSessionManager(tbCoreQueueFactory, topicService, kafkaAdmin, sessions); + + Edge edge = new Edge(new EdgeId(UUID.randomUUID())); + edge.setTenantId(TenantId.fromUUID(UUID.randomUUID())); + state = new EdgeSessionState(); + state.setEdge(edge); + + EdgeSession session = mock(EdgeSession.class); + when(session.getState()).thenReturn(state); + + ReflectionTestUtils.setField(manager, "session", session); + ReflectionTestUtils.setField(manager, "ctx", ctx); + ReflectionTestUtils.setField(manager, "downlinkMessageMapper", mock(DownlinkMessageMapper.class)); + } + + @AfterEach + void tearDown() { + if (manager != null) { + manager.destroy(); + } + } + + @Test + void readyOnlyWhenConnectedNotSyncingNotHighPriority() { + setReadiness(true, false, false); + assertThat(isReadyToProcessGeneralEvents()) + .as("connected, not syncing, no high-priority work -> ready") + .isTrue(); + } + + @Test + void notReadyWhenDisconnected() { + setReadiness(false, false, false); + assertThat(isReadyToProcessGeneralEvents()) + .as("disconnected -> not ready") + .isFalse(); + } + + @Test + void notReadyWhileSyncInProgress() { + setReadiness(true, true, false); + assertThat(isReadyToProcessGeneralEvents()) + .as("sync in progress -> not ready (this is the window where events were being dropped)") + .isFalse(); + } + + @Test + void notReadyWhileHighPriorityProcessing() { + setReadiness(true, false, true); + assertThat(isReadyToProcessGeneralEvents()) + .as("high-priority processing -> not ready") + .isFalse(); + } + + @Test + void initConsumerWiresReadinessPredicateIntoConsumerGate() { + stubStorageSettings(); + + @SuppressWarnings("unchecked") + TbQueueConsumer> queueConsumer = mock(TbQueueConsumer.class); + // Report stopped so the launched consumer loop exits immediately - this test asserts on wiring, not polling. + when(queueConsumer.isStopped()).thenReturn(true); + when(tbCoreQueueFactory.createEdgeEventMsgConsumer(any(), any())).thenReturn(queueConsumer); + + setReadiness(true, false, false); + initConsumer(); + + QueueConsumerManager> consumer = manager.getConsumer(); + assertThat(consumer).as("consumer must be built").isNotNull(); + + BooleanSupplier readinessCheck = (BooleanSupplier) ReflectionTestUtils.getField(consumer, "readinessCheck"); + assertThat(readinessCheck) + .as("the edge consumer must be wired with a readinessCheck (the .readinessCheck(...) builder line)") + .isNotNull(); + + // It must be the live predicate, not a snapshot: flipping the session state must flip the gate. + assertThat(readinessCheck.getAsBoolean()).as("ready session -> gate open").isTrue(); + state.tryStartSync(); + assertThat(readinessCheck.getAsBoolean()).as("sync starts -> gate closes, consumer pauses polling").isFalse(); + } + + @Test + void eventArrivingDuringSyncIsHeldByTheEdgeConsumerUntilSyncCompletes() { + stubStorageSettings(); + + RecordingEdgeEventConsumer queueConsumer = new RecordingEdgeEventConsumer(); + when(tbCoreQueueFactory.createEdgeEventMsgConsumer(any(), any())).thenReturn(queueConsumer); + + // The consumer is launched only while the session is ready. + setReadiness(true, false, false); + initConsumer(); + + // Sync starts: the gate closes. Wait until the loop has actually parked on it (poll count stops advancing) + // before enqueuing - otherwise we would race an in-flight poll() and the test would be non-deterministic. + state.tryStartSync(); + awaitParkedOnClosedGate(queueConsumer); + + // An event lands in the edge-event topic during the sync window - exactly the case that used to be dropped. + @SuppressWarnings("unchecked") + TbProtoQueueMsg event = mock(TbProtoQueueMsg.class); + int pollsBeforeEvent = queueConsumer.getPollCount(); + queueConsumer.enqueue(List.of(event)); + + // While sync is in progress the consumer stays parked: it neither polls nor consumes the event. + sleepQuietly(POLL_INTERVAL_MS * 5); + assertThat(queueConsumer.getPolledEvents()) + .as("event must not be polled while sync is in progress (it must stay queued, not be dropped)") + .isEmpty(); + assertThat(queueConsumer.getPollCount()) + .as("consumer must not poll at all while the gate is closed") + .isEqualTo(pollsBeforeEvent); + + // Sync completes: the gate opens and the held event is finally picked up by the consumer. + state.finishSync(); + await().atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(queueConsumer.getPolledEvents()) + .as("event held during sync must be picked up once sync completes, not lost") + .hasSize(1)); + } + + private void stubStorageSettings() { + EdgeEventStorageSettings storageSettings = mock(EdgeEventStorageSettings.class); + when(storageSettings.getNoRecordsSleepInterval()).thenReturn(POLL_INTERVAL_MS); + when(ctx.getEdgeEventStorageSettings()).thenReturn(storageSettings); + } + + private void initConsumer() { + ReflectionTestUtils.invokeMethod(manager, "initConsumerAndExecutor", state.getTenantId(), state.getEdgeId(), state); + } + + private boolean isReadyToProcessGeneralEvents() { + return Boolean.TRUE.equals(ReflectionTestUtils.invokeMethod(manager, "isReadyToProcessGeneralEvents")); + } + + private void setReadiness(boolean connected, boolean syncInProgress, boolean highPriorityProcessing) { + state.setConnected(connected); + if (syncInProgress) { + state.tryStartSync(); + } else { + state.finishSync(); + } + ReflectionTestUtils.setField(manager, "isHighPriorityProcessing", highPriorityProcessing); + } + + private static void awaitParkedOnClosedGate(RecordingEdgeEventConsumer consumer) { + await().atMost(5, TimeUnit.SECONDS).until(() -> { + int before = consumer.getPollCount(); + // Several poll intervals with no new poll means the loop is parked on the readiness gate rather than + // blocked inside poll(), so it is safe to enqueue without racing an in-flight read. + sleepQuietly(POLL_INTERVAL_MS * 5); + return consumer.getPollCount() == before; + }); + } + + private static void sleepQuietly(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** + * Fake edge-event consumer that records what it was actually polled, so a test can prove an event stayed in the + * queue while the readiness gate was closed and was only read once it opened. + */ + private static class RecordingEdgeEventConsumer implements TbQueueConsumer> { + + private final Queue>> pending = new ConcurrentLinkedQueue<>(); + private final List> polledEvents = new CopyOnWriteArrayList<>(); + private final AtomicInteger pollCount = new AtomicInteger(); + private volatile boolean stopped; + + void enqueue(List> batch) { + pending.add(batch); + } + + int getPollCount() { + return pollCount.get(); + } + + List> getPolledEvents() { + return polledEvents; + } + + @Override + public List> poll(long durationInMillis) { + pollCount.incrementAndGet(); + List> batch = pending.poll(); + if (batch != null) { + polledEvents.addAll(batch); + return batch; + } + sleepQuietly(durationInMillis); + return Collections.emptyList(); + } + + @Override + public String getTopic() { + return "test-edge-event-topic"; + } + + @Override + public void subscribe() { + } + + @Override + public void subscribe(Set partitions) { + } + + @Override + public void stop() { + stopped = true; + } + + @Override + public void unsubscribe() { + stopped = true; + } + + @Override + public void commit() { + } + + @Override + public boolean isStopped() { + return stopped; + } + + @Override + public List getFullTopicNames() { + return Collections.emptyList(); + } + + @Override + public Set getPartitions() { + return Collections.emptySet(); + } + + } + +} From 4f8dcb3dceaff18716b6cda93ea0ca8b23b9df17 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 11 Jun 2026 16:43:58 +0300 Subject: [PATCH 3/3] Remove comments --- .../manager/KafkaBasedEdgeGrpcSessionManager.java | 11 ----------- .../manager/KafkaBasedEdgeGrpcSessionManagerTest.java | 7 ------- 2 files changed, 18 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java index 01174423c2..f42f48421a 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManager.java @@ -251,17 +251,6 @@ public class KafkaBasedEdgeGrpcSessionManager extends AbstractEdgeGrpcSessionMan } } - /** - * Readiness gate for the edge-event consumer (see {@link QueueConsumerManager}'s {@code readinessCheck}): it polls - * only while the session is connected and no sync or high-priority processing is running. - *

- * Pausing polling in those windows is deliberate. A batch that is polled but then skipped advances the Kafka - * position without committing, so it is lost until a rebalance; keeping events queued instead matches the no-loss - * behaviour the Postgres-based manager gets by re-reading by seqId. This covers the common case - a batch already - * in flight when a sync starts is a known residual (closing it would need seek/rewind, which {@code TbQueueConsumer} - * lacks). If a sync ever outran {@code max.poll.interval.ms} (default 5 min) the consumer is rebalanced and resumes - * from the committed offset: still no loss, just a possible replay. Edge syncs are seconds, so this is acceptable. - */ private boolean isReadyToProcessGeneralEvents() { return isSessionReady(getState()); } diff --git a/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java b/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java index 0759a6b05e..7ddc2af192 100644 --- a/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java +++ b/application/src/test/java/org/thingsboard/server/service/edge/rpc/session/manager/KafkaBasedEdgeGrpcSessionManagerTest.java @@ -54,13 +54,6 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; -/** - * Unit tests for the readiness gate that {@link KafkaBasedEdgeGrpcSessionManager} feeds into its edge-event consumer. - *

- * The generic poll-gate mechanism is covered by {@code QueueConsumerManagerTest}; these tests pin the edge-specific - * half of the fix: the manager's readiness predicate, and the fact that the consumer is actually wired with it - so - * dropping the {@code .readinessCheck(...)} builder line would silently reintroduce the event-loss bug. - */ class KafkaBasedEdgeGrpcSessionManagerTest { private static final long POLL_INTERVAL_MS = 20L;