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).

application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java

@@ -70,7 +70,9 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
 
     private void processMsgs(List<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> msgs, TbQueueConsumer<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> consumer) {
         log.trace("[{}][{}] starting processing edge events", tenantId, edge.getId());
-        if (!isConnected() || 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 (!isReadyToProcessGeneralEvents()) {
             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, edge.getId(), isConnected(), isSyncInProgress(), isHighPriorityProcessing);
@@ -96,6 +98,10 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
         }
     }
 
+    private boolean isReadyToProcessGeneralEvents() {
+        return isConnected() && !isSyncInProgress() && !isHighPriorityProcessing;
+    }
+
     @Override
     public ListenableFuture<Boolean> migrateEdgeEvents() throws Exception {
         return super.processEdgeEvents();
@@ -103,7 +109,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
 
     @Override
     public ListenableFuture<Boolean> processEdgeEvents() {
-        if (!isConnected() || isSyncInProgress() || isHighPriorityProcessing) {
+        if (!isReadyToProcessGeneralEvents()) {
             log.warn("[{}][{}] Session is not ready (connected={}, syncInProgress={}, highPriority={}), skip starting edge event consumer",
                     tenantId, edge != null ? edge.getId() : null, isConnected(), isSyncInProgress(), isHighPriorityProcessing);
             return Futures.immediateFuture(Boolean.FALSE);
@@ -126,6 +132,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
                         .consumerCreator(() -> tbCoreQueueFactory.createEdgeEventMsgConsumer(tenantId, edge.getId()))
                         .consumerExecutor(consumerExecutor)
                         .threadPrefix("edge-events-" + edge.getId())
+                        .readinessCheck(this::isReadyToProcessGeneralEvents)
                         .build();
                 consumer.subscribe();
                 consumer.launch();

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<M extends TbQueueMsg> {
     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<M> consumer;
@@ -49,12 +52,13 @@ public class QueueConsumerManager<M extends TbQueueMsg> {
     @Builder
     public QueueConsumerManager(String name, MsgPackProcessor<M> msgPackProcessor,
                                 long pollInterval, Supplier<TbQueueConsumer<M>> 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<M extends TbQueueMsg> {
     private void consumerLoop(TbQueueConsumer<M> consumer) {
         while (!stopped && !consumer.isStopped()) {
             try {
+                if (!isReadyToProcess()) {
+                    if (!awaitNextReadinessCheck()) {
+                        return;
+                    }
+                    continue;
+                }
                 List<M> msgs = consumer.poll(pollInterval);
                 if (msgs.isEmpty()) {
                     continue;
@@ -102,6 +112,25 @@ public class QueueConsumerManager<M extends TbQueueMsg> {
         }
     }
 
+    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;

common/queue/src/test/java/org/thingsboard/server/queue/common/consumer/QueueConsumerManagerTest.java

@@ -0,0 +1,247 @@
+/**
+ * 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<TbQueueMsg> manager;
+
+    @AfterEach
+    void tearDown() {
+        if (manager != null) {
+            manager.stop();
+        }
+        if (consumerExecutor != null) {
+            consumerExecutor.shutdownNow();
+        }
+    }
+
+    @Test
+    void eventQueuedWhileNotReadyIsDeliveredAfterReadinessGateOpensInsteadOfBeingDropped() {
+        List<TbQueueMsg> 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<TbQueueMsg> 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<TbQueueMsg> launchManager(TestQueueConsumer consumer, BooleanSupplier readinessCheck,
+                                                           QueueConsumerManager.MsgPackProcessor<TbQueueMsg> processor) {
+        consumerExecutor = Executors.newSingleThreadExecutor();
+        QueueConsumerManager<TbQueueMsg> queueConsumerManager = QueueConsumerManager.<TbQueueMsg>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<TbQueueMsg> {
+
+        private final Queue<List<TbQueueMsg>> batches = new ConcurrentLinkedQueue<>();
+        private final AtomicInteger pollCount = new AtomicInteger();
+        private volatile boolean stopped;
+
+        void enqueue(List<TbQueueMsg> batch) {
+            batches.add(batch);
+        }
+
+        int getPollCount() {
+            return pollCount.get();
+        }
+
+        @Override
+        public List<TbQueueMsg> poll(long durationInMillis) {
+            pollCount.incrementAndGet();
+            List<TbQueueMsg> 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<TopicPartitionInfo> 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<String> getFullTopicNames() {
+            return Collections.emptyList();
+        }
+
+    }
+
+}