From ad270c81cb0410ba2c8aaf792936422ba0f04f64 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 11 Jun 2026 16:07:25 +0300 Subject: [PATCH] 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). --- .../edge/rpc/KafkaEdgeGrpcSession.java | 11 +- .../common/consumer/QueueConsumerManager.java | 31 ++- .../consumer/QueueConsumerManagerTest.java | 251 ++++++++++++++++++ 3 files changed, 290 insertions(+), 3 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/KafkaEdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java index f7e729c6b7..082452e944 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 @@ -70,7 +70,9 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession { private void processMsgs(List> msgs, TbQueueConsumer> 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 migrateEdgeEvents() throws Exception { return super.processEdgeEvents(); @@ -103,7 +109,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession { @Override public ListenableFuture 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(); 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(); + } + + } + +}