> deviceTelemetryTimers = Collections.synchronizedList(new ArrayList<>());
+
+ private boolean staggered() {
+ return "STAGGERED".equalsIgnoreCase(onboardMode);
+ }
+
@Override
public void createDevices() throws Exception {
createDevices(true);
@@ -53,9 +83,194 @@ public void removeDevices() throws Exception {
removeEntities(devices.stream().map(IdBased::getId).collect(Collectors.toList()), "devices");
}
+ @Override
+ public void warmUpDevices() throws InterruptedException {
+ if (staggered()) {
+ // STAGGERED: no separate warm-up phase; each device's own periodic telemetry timer (started
+ // by EntityLifecycle.onboard(), driven from runApiTests()) sends its own first message.
+ return;
+ }
+ super.warmUpDevices();
+ }
+
@Override
public void runApiTests() throws InterruptedException {
- super.runApiTests(mqttClients.size());
+ if (!staggered()) {
+ super.runApiTests(mqttClients.size());
+ return;
+ }
+ runStaggeredApiTests();
+ }
+
+ /**
+ * STAGGERED: ramp devices in through the engine (each onboard connects and schedules its own
+ * telemetry timer — see {@link #connectAndScheduleDevice(int)}), then hold for the test duration.
+ * Direct-device mode has no RPC subscribe step (RPC is a gateway-mode concept) — STAGGERED here is
+ * telemetry-only, matching what {@link MqttDeviceAPITest} already supports in PHASED.
+ */
+ private void runStaggeredApiTests() throws InterruptedException {
+ // Register the THROUGHPUT stats source BEFORE starting the reporter (same order + same block
+ // AbstractAPITest.runApiTests(int) uses for PHASED) — StatsReporter.start() snapshots
+ // sources.isEmpty() once at call time and never reschedules if it was empty then, so without this
+ // the reporter would log "no active sources" and stay inert for the whole run (direct-device
+ // STAGGERED registers nothing else).
+ if (testMessagesPerSecond > 0) {
+ statsReporter().register(StatsBlock.THROUGHPUT,
+ new ThroughputStats(totalSuccessPublishedCount, totalFailedPublishedCount)::summaryAndReset);
+ }
+ statsReporter().start();
+ AtomicBoolean rampCompleted = new AtomicBoolean(false);
+ onboardingEngine = new StaggeredOnboardingEngine(
+ deviceLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed);
+ onboardingEngine.start((onboarded, failed) -> {
+ rampCompleted.set(true);
+ log.info("STAGGERED device ramp complete: {} onboarded, {} failed", onboarded, failed);
+ });
+ try {
+ Thread.sleep(testDurationInSec * 1000L);
+ } finally {
+ if (!rampCompleted.get()) {
+ log.warn("STAGGERED: test.duration ({}s) elapsed before the onboarding ramp completed — "
+ + "not every device may have onboarded or started publishing telemetry. "
+ + "Consider raising DURATION_IN_SECONDS or lowering ONBOARD_MAX_CONCURRENT/ONBOARD_FIRST_JITTER_SEC.",
+ testDurationInSec);
+ }
+ for (ScheduledFuture timer : deviceTelemetryTimers) {
+ timer.cancel(false);
+ }
+ if (onboardingEngine != null) {
+ onboardingEngine.stop();
+ }
+ }
+ }
+
+ private EntityLifecycle deviceLifecycle() {
+ return new EntityLifecycle() {
+ @Override
+ public int entityCount() {
+ return deviceEndIdx - deviceStartIdx;
+ }
+
+ @Override
+ public void onboard(int idx) throws Exception {
+ int devIdx = deviceStartIdx + idx;
+ // 1) connect this device's client (persistent, autoReconnect via createClient)
+ // 2) schedule this device's telemetry timer
+ connectAndScheduleDevice(devIdx);
+ }
+ };
+ }
+
+ /**
+ * One STAGGERED device's full onboarding step, composed entirely from existing pieces: the same
+ * connect sequence {@link #initClientBlocking} performs for PHASED's bulk connect. Synchronous:
+ * throws on any failure so the engine counts this device as failed rather than onboarded.
+ * Nothing is registered into the shared {@code mqttClients}/{@code deviceClients} collections —
+ * which the telemetry scheduler reads from — until the connect above has succeeded; a device that
+ * fails to connect is closed (by {@link #initClientBlocking}) and left out of every shared collection
+ * entirely.
+ */
+ private void connectAndScheduleDevice(int devIdx) throws Exception {
+ int localIdx = devIdx - deviceStartIdx;
+ String deviceName = staggeredDeviceNames.get(localIdx);
+
+ // 1) connect (persistent; createClient() applies autoReconnect() same as every other device client)
+ MqttClient client = initClientBlocking(deviceName);
+
+ // Onboarding succeeded: only now commit this device's client into the shared collections and
+ // start its telemetry timer.
+ mqttClients.add(client);
+ clientNames.put(client, deviceName);
+ DeviceClient dc = new DeviceClient();
+ dc.setMqttClient(client);
+ dc.setDeviceName(deviceName);
+ deviceClients.add(dc);
+
+ // 2) schedule this device's own telemetry timer, starting immediately (its onboarding time is
+ // already spread by the engine's ramp jitter; an extra small per-device startup offset avoids
+ // every device's tick landing on the exact same millisecond).
+ scheduleDeviceTelemetry(devIdx, client, deviceName);
+ }
+
+ /** Blocking connect for one device token, identical in behavior to the private {@code initClient} used
+ * by PHASED's {@code connectDevices} — reconstructed here (rather than reused) only because that method
+ * is {@code private} in {@link org.thingsboard.tools.service.shared.BaseMqttAPITest}. */
+ private MqttClient initClientBlocking(String token) throws Exception {
+ MqttClient client = createClient(token);
+ Future connectFuture = connectAsync(client);
+ MqttConnectResult result;
+ try {
+ result = connectFuture.get(CONNECT_TIMEOUT, TimeUnit.SECONDS);
+ } catch (TimeoutException ex) {
+ connectFuture.cancel(true);
+ client.disconnect();
+ throw new RuntimeException(String.format("STAGGERED: timed out connecting device [%s]", token), ex);
+ }
+ if (!result.isSuccess()) {
+ connectFuture.cancel(true);
+ client.disconnect();
+ throw new RuntimeException(String.format("STAGGERED: failed to connect device [%s]. Result code: %s", token, result.getReturnCode()));
+ }
+ return client;
+ }
+
+ /** Schedules this device's own periodic telemetry publish (one MQTT publish per device, same message
+ * construction {@link org.thingsboard.tools.service.shared.BaseMqttAPITest#nextPublishTask} uses for
+ * PHASED's per-device publish), independent of every other device's timer. No-op when publishing is
+ * disabled (MESSAGES_PER_SECOND=0).
+ * Period is MPS-derived so STAGGERED's steady-state aggregate matches PHASED's: today's metronome
+ * does {@code testMessagesPerSecond} single-device publishes/sec by sweeping the whole fleet, i.e.
+ * each device publishes once every {@code entityCount / testMessagesPerSecond} seconds — so each
+ * independent per-device timer here fires on that same period, jittered so the first fires aren't
+ * synchronized across devices.
+ *
Package-private (not private): exercised directly by MqttDeviceAPITestTest. */
+ void scheduleDeviceTelemetry(int devIdx, MqttClient client, String deviceName) {
+ if (testMessagesPerSecond <= 0) {
+ return;
+ }
+ DeviceClient logClient = new DeviceClient();
+ logClient.setMqttClient(client);
+ logClient.setDeviceName(deviceName);
+ AtomicInteger tick = new AtomicInteger();
+ int entityCount = deviceEndIdx - deviceStartIdx;
+ long periodMs = Math.max(1L, (entityCount * 1000L) / testMessagesPerSecond);
+ long initialJitterMs = EphemeralSchedule.firstOffsetMillis(new Random(seed + devIdx), periodMs);
+ ScheduledFuture timer = restClientService.getScheduler().scheduleAtFixedRate(() -> {
+ try {
+ Msg message = getNextMessage(deviceName, false);
+ int iteration = tick.incrementAndGet();
+ client.publish(getTestTopic(), Unpooled.wrappedBuffer(message.getData()), MqttQoS.AT_MOST_ONCE)
+ .addListener(f -> {
+ if (f.isSuccess()) {
+ totalSuccessPublishedCount.incrementAndGet();
+ logSuccessTestMessage(iteration, logClient);
+ } else {
+ totalFailedPublishedCount.incrementAndGet();
+ logFailureTestMessage(iteration, logClient, f);
+ }
+ });
+ } catch (Exception e) {
+ log.warn("STAGGERED telemetry publish failed for device [{}]", deviceName, e);
+ }
+ }, initialJitterMs, periodMs, TimeUnit.MILLISECONDS);
+ deviceTelemetryTimers.add(timer);
+ }
+
+ /**
+ * Fails fast on an unsupported STAGGERED configuration, instead of silently diverging from it.
+ * {@link #scheduleDeviceTelemetry} always publishes one plain per-device message and never injects an
+ * alarm (needs {@code test.alarms.aps <= 0}). Deliberately narrow: STAGGERED currently supports
+ * exactly the no-alarms scenario this mode targets.
+ */
+ void checkStaggeredSupported() {
+ if (alarmsPerSecond > 0) {
+ String msg = String.format(
+ "onboard.mode=STAGGERED currently supports NO alarms for direct-device mode; "
+ + "test.alarms.aps=%d (> 0) is not supported yet. Set ALARMS_PER_SECOND=0, or use onboard.mode=PHASED.",
+ alarmsPerSecond);
+ log.error(msg);
+ throw new IllegalStateException(msg);
+ }
}
@Override
@@ -90,6 +305,13 @@ protected void logFailureTestMessage(int iteration, DeviceClient client, Future<
@Override
public void connectDevices() throws InterruptedException {
+ if (staggered()) {
+ // STAGGERED: no bulk connect here. Build the device name model only; the engine (driven from
+ // runApiTests()) connects + schedules each device's telemetry timer on its own paced schedule.
+ checkStaggeredSupported();
+ prepareStaggeredModel();
+ return;
+ }
AtomicInteger totalConnectedCount = new AtomicInteger();
List pack = null;
List devicesNames;
@@ -124,6 +346,28 @@ public void generationX509() {
}
+ /**
+ * STAGGERED-only: same device-name resolution as this method's PHASED body above, but index-based
+ * (0-based, local to this instance's [deviceStartIdx, deviceEndIdx)) instead of derived from each
+ * already-connected {@link MqttClient}'s username — so the assignment exists before any device has
+ * connected, and {@code onboard(idx)} can look up its own name deterministically regardless of
+ * connect order/timing.
+ */
+ // Package-private (not private): exercised directly by MqttDeviceAPITestTest.
+ void prepareStaggeredModel() {
+ List devicesNames;
+ if (!devices.isEmpty()) {
+ devicesNames = devices.stream().map(Device::getName).collect(Collectors.toList());
+ } else {
+ devicesNames = new ArrayList<>();
+ for (int i = deviceStartIdx; i < deviceEndIdx; i++) {
+ devicesNames.add(getToken(false, i));
+ }
+ }
+ this.staggeredDeviceNames = devicesNames;
+ log.info("STAGGERED model prepared: {} devices", devicesNames.size());
+ }
+
private void mapDevicesToDeviceClientConnections() {
for (MqttClient mqttClient : mqttClients) {
DeviceClient client = new DeviceClient();
diff --git a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java
index 51701045..2ceae940 100644
--- a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java
+++ b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java
@@ -17,6 +17,7 @@
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.concurrent.Future;
@@ -27,14 +28,19 @@
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.id.IdBased;
import org.thingsboard.mqtt.MqttClient;
+import org.thingsboard.mqtt.MqttConnectResult;
import org.thingsboard.tools.service.gateway.rpc.GatewayRpcReceiver;
import org.thingsboard.tools.service.gateway.rpc.RpcBurstSender;
import org.thingsboard.tools.service.gateway.rpc.RpcLatencyStats;
import org.thingsboard.tools.service.gateway.rpc.RpcMessageProcessor;
import org.thingsboard.tools.service.gateway.rpc.RpcResponseTemplate;
+import org.thingsboard.tools.service.msg.NodeMsg;
import org.thingsboard.tools.service.mqtt.DeviceClient;
import org.thingsboard.tools.service.shared.BaseMqttAPITest;
import org.thingsboard.tools.service.shared.StatsBlock;
+import org.thingsboard.tools.service.shared.ThroughputStats;
+import org.thingsboard.tools.service.shared.onboarding.EntityLifecycle;
+import org.thingsboard.tools.service.shared.onboarding.StaggeredOnboardingEngine;
import jakarta.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
@@ -45,6 +51,10 @@
import java.util.Map;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -60,6 +70,12 @@ public class MqttGatewayAPITest extends BaseMqttAPITest implements GatewayAPITes
@Value("${gateway.count}")
int gatewayCount;
+ // STAGGERED-only support check (see checkStaggeredSupported()): mirrors the same property that
+ // selects which bean loads (this class when gateway.batch!=true, MqttGatewayBatchAPITest when it's
+ // true), read directly here rather than inferred from the bean type.
+ @Value("${gateway.batch:false}")
+ boolean gatewayBatchEnabled;
+
@Value("${gateway.rpc.enabled:false}")
boolean rpcEnabled;
@Value("${gateway.rpc.topic:v1/gateway/rpc}")
@@ -119,11 +135,33 @@ public class MqttGatewayAPITest extends BaseMqttAPITest implements GatewayAPITes
int rpcSenderTimeoutMs;
@Value("${gateway.rpc.sender.mode:BURST}")
String rpcSenderMode;
+ @Value("${gateway.rpc.sender.fireThreads:48}")
+ int rpcSenderFireThreads;
+ @Value("${gateway.rpc.sender.maxInFlight:512}")
+ int rpcSenderMaxInFlight;
@Value("${rest.url}")
String restUrl;
private RpcBurstSender rpcBurstSender;
+ // STAGGERED onboarding (onboard.mode=STAGGERED): PHASED (default) never touches any of these.
+ private StaggeredOnboardingEngine onboardingEngine;
+ // Precomputed gateway-index -> name / sub-device-names model, built once by prepareStaggeredModel().
+ // Unlike PHASED's mapDevicesToGatewayClientConnections (which derives the mapping from mqttClients'
+ // connect order, established only after ALL gateways connect), this is index-based so onboard(idx)
+ // can look up its own assignment before any connection exists.
+ // Package-private (not private): MqttGatewayAPITestTest reads/exercises these directly, following
+ // the existing broker-free unit-test idiom (test class extends the SUT and touches its own members).
+ List staggeredGatewayNames;
+ Map> staggeredGatewayDeviceNames;
+ // Per-gateway telemetry timers started by the STAGGERED path (one per onboarded gateway); cancelled
+ // when the test duration elapses.
+ final List> gatewayTelemetryTimers = Collections.synchronizedList(new ArrayList<>());
+
+ private boolean staggered() {
+ return "STAGGERED".equalsIgnoreCase(onboardMode);
+ }
+
@Override
protected boolean isInboundHandlingEnabled() {
return rpcEnabled;
@@ -140,7 +178,8 @@ protected boolean isInboundHandlingEnabled() {
// Gateway client -> its sub-device names, for re-announcing sub-devices on reconnect.
// ConcurrentHashMap (not HashMap): populated on the main thread before start, then read from reconnect
// callbacks on netty event-loop threads — safe-publish without relying on incidental happens-before.
- private final Map> gatewayDeviceNames = new ConcurrentHashMap<>();
+ // Package-private (not private): MqttGatewayAPITestTest asserts on this directly (commit-on-success-only).
+ final Map> gatewayDeviceNames = new ConcurrentHashMap<>();
@PostConstruct
@@ -169,6 +208,14 @@ public void createGateways() throws Exception {
@Override
public void connectGateways() throws InterruptedException {
+ if (staggered()) {
+ // STAGGERED: no bulk connect here. Build the gateway/device name model only; the engine
+ // (driven from runApiTests()) connects + announces + subscribes each gateway on its own
+ // paced schedule.
+ checkStaggeredSupported();
+ prepareStaggeredModel();
+ return;
+ }
AtomicInteger totalConnectedCount = new AtomicInteger();
List pack = null;
List gatewayNames;
@@ -203,7 +250,9 @@ public void connectGateways() throws InterruptedException {
}
}
- private void mapDevicesToGatewayClientConnections() {
+ // Package-private (not private): exercised directly by MqttGatewayAPITestTest for parity with
+ // prepareStaggeredModel().
+ void mapDevicesToGatewayClientConnections() {
int gatewayCount = mqttClients.size();
for (int i = deviceStartIdx; i < deviceEndIdx; i++) {
int deviceIdx = i - deviceStartIdx;
@@ -219,6 +268,61 @@ private void mapDevicesToGatewayClientConnections() {
}
}
+ /**
+ * STAGGERED-only: same gateway-name resolution and deviceIdx % gatewayCount assignment as
+ * {@link #mapDevicesToGatewayClientConnections()}, but keyed by gateway INDEX (0-based, local to
+ * this instance's [gatewayStartIdx, gatewayEndIdx)) instead of by connected {@link MqttClient} —
+ * so the assignment exists before any gateway has connected, and {@code onboard(idx)} can look up
+ * its own sub-device names deterministically regardless of connect order/timing.
+ */
+ // Package-private (not private): exercised directly by MqttGatewayAPITestTest.
+ void prepareStaggeredModel() {
+ List gatewayNames;
+ if (!gateways.isEmpty()) {
+ gatewayNames = gateways.stream().map(Device::getName).collect(Collectors.toList());
+ } else {
+ gatewayNames = new ArrayList<>();
+ for (int i = gatewayStartIdx; i < gatewayEndIdx; i++) {
+ gatewayNames.add(getToken(true, i));
+ }
+ }
+ this.staggeredGatewayNames = gatewayNames;
+ int gatewayCount = gatewayNames.size();
+ Map> byGatewayIdx = new ConcurrentHashMap<>();
+ for (int i = deviceStartIdx; i < deviceEndIdx; i++) {
+ int deviceIdx = i - deviceStartIdx;
+ int gatewayIdx = deviceIdx % gatewayCount;
+ byGatewayIdx.computeIfAbsent(gatewayIdx, k -> Collections.synchronizedList(new ArrayList<>()))
+ .add(getToken(false, i));
+ }
+ this.staggeredGatewayDeviceNames = byGatewayIdx;
+ log.info("STAGGERED model prepared: {} gateways, {} devices", gatewayCount, deviceEndIdx - deviceStartIdx);
+ }
+
+ /**
+ * Fails fast on an unsupported STAGGERED configuration, instead of silently diverging from it.
+ * {@link #scheduleGatewayTelemetry} always publishes a whole-gateway batch (needs {@code
+ * gateway.batch=true}) and never injects an alarm (needs {@code test.alarms.aps <= 0}). Deliberately
+ * narrow: STAGGERED currently supports exactly the persistent gateway-batch, no-alarms scenario this
+ * mode targets.
+ */
+ void checkStaggeredSupported() {
+ if (!gatewayBatchEnabled) {
+ String msg = "onboard.mode=STAGGERED currently supports gateway.batch=true only (with no alarms); "
+ + "gateway.batch=false is not supported yet. Set GATEWAY_BATCH=true, or use onboard.mode=PHASED.";
+ log.error(msg);
+ throw new IllegalStateException(msg);
+ }
+ if (alarmsPerSecond > 0) {
+ String msg = String.format(
+ "onboard.mode=STAGGERED currently supports gateway.batch=true with NO alarms; "
+ + "test.alarms.aps=%d (> 0) is not supported yet. Set ALARMS_PER_SECOND=0, or use onboard.mode=PHASED.",
+ alarmsPerSecond);
+ log.error(msg);
+ throw new IllegalStateException(msg);
+ }
+ }
+
/** Re-announce a reconnected gateway's sub-devices so the server re-routes their RPC through it.
* Reuses the gateway connect topic and payload used at warm-up. */
private void reannounceDevices(MqttClient gatewayClient) {
@@ -246,15 +350,102 @@ protected Future warmUpPublish(DeviceClient deviceClient) {
return super.warmUpPublish(deviceClient); // RPC off: legacy QoS-0 warm-up
}
+ @Override
+ public void warmUpDevices() throws InterruptedException {
+ if (staggered()) {
+ return; // STAGGERED: announcement happens inside EntityLifecycle.onboard(), driven from runApiTests()
+ }
+ super.warmUpDevices();
+ }
+
@Override
public void runApiTests() throws InterruptedException {
- if (rpcSenderEnabled) {
- startRpcBurstSender();
+ if (!staggered()) {
+ if (rpcSenderEnabled) {
+ startRpcBurstSender();
+ }
+ try {
+ super.runApiTests(deviceClients.size());
+ } finally {
+ // Stop firing new bursts BEFORE draining so the tail can settle without fresh inbound.
+ if (rpcBurstSender != null) {
+ rpcBurstSender.stop();
+ }
+ if (rpcEnabled && rpcReceiver != null) {
+ long quietMs = rpcDrainQuietSec * 1000L;
+ long maxMs = GatewayRpcReceiver.resolveDrainMaxMs(
+ rpcDrainMaxSecConfig, rpcSenderEnabled, rpcSenderTimeoutMs, rpcResponseDelayMs, rpcDrainQuietSec);
+ log.info("Gateway RPC drain: waiting for in-flight RPCs to settle (quietSec={}, maxSec={})...",
+ rpcDrainQuietSec, maxMs / 1000);
+ GatewayRpcReceiver.DrainResult result = rpcReceiver.drain(quietMs, maxMs, rpcRespond);
+ rpcReceiver.finalizeLostReplies(); // replies still buffered (client never reconnected) are lost
+ rpcReceiver.logPending(); // name the distinct still-unanswered RPCs for DB EXPIRED correlation
+ String drainLine = String.format("Gateway RPC drain complete [drained %.1fs, quiesced=%b]",
+ result.elapsedMs / 1000.0, result.quiesced);
+ if (result.quiesced) {
+ log.info(drainLine);
+ } else {
+ log.warn(drainLine);
+ }
+ log.info(rpcReceiver.inTotalSummary()); // RPC In [total]: publish=… (new …, redelivered …)
+ log.info(rpcReceiver.outTotalSummary()); // RPC Out [total]: publish=…, pubAck=…, failed=…, recovered=…, lost=…
+
+ }
+ }
+ return;
+ }
+ runStaggeredApiTests();
+ }
+
+ /**
+ * STAGGERED: ramp gateways in through the engine (each onboard connects + announces + subscribes +
+ * schedules its own telemetry timer — see {@link #connectAnnounceSubscribeAndSchedule(int)}), start
+ * the RPC sender only once the ramp completes (PHASED starts it up-front, before any connection
+ * exists), then hold for the test duration. Reuses the same RPC drain/summary block as the PHASED
+ * {@code finally} above.
+ */
+ private void runStaggeredApiTests() throws InterruptedException {
+ // Register stats sources BEFORE starting the reporter — StatsReporter.start() snapshots
+ // sources.isEmpty() once at call time; if it's empty then, it logs "no active sources" and never
+ // schedules, so a source registered afterward would never print periodically (matches PHASED's
+ // order: connectGateways()'s attachRpcReceiver()/initRpcReceiver() always runs, via connectGateways(),
+ // before runApiTests() reaches statsReporter().start()).
+ if (rpcEnabled) {
+ initRpcReceiver();
}
+ // Same THROUGHPUT registration AbstractAPITest.runApiTests(int) does for PHASED — otherwise
+ // STAGGERED's per-gateway telemetry timers (which feed the same totalSuccess/totalFailed counters)
+ // have no periodic reporter block at all.
+ if (testMessagesPerSecond > 0) {
+ statsReporter().register(StatsBlock.THROUGHPUT,
+ new ThroughputStats(totalSuccessPublishedCount, totalFailedPublishedCount)::summaryAndReset);
+ }
+ statsReporter().start();
+ AtomicBoolean rampCompleted = new AtomicBoolean(false);
+ onboardingEngine = new StaggeredOnboardingEngine(
+ gatewayLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed);
+ onboardingEngine.start((onboarded, failed) -> {
+ rampCompleted.set(true);
+ log.info("STAGGERED gateway ramp complete: {} onboarded, {} failed — starting RPC sender", onboarded, failed);
+ if (rpcSenderEnabled) {
+ startRpcBurstSender(); // existing method, unchanged: full device list
+ }
+ });
try {
- super.runApiTests(deviceClients.size());
+ Thread.sleep(testDurationInSec * 1000L);
} finally {
- // Stop firing new bursts BEFORE draining so the tail can settle without fresh inbound.
+ if (!rampCompleted.get()) {
+ log.warn("STAGGERED: test.duration ({}s) elapsed before the onboarding ramp completed — "
+ + "not every gateway may have onboarded, and the RPC sender (if enabled) never started. "
+ + "Consider raising DURATION_IN_SECONDS or lowering ONBOARD_MAX_CONCURRENT/ONBOARD_FIRST_JITTER_SEC.",
+ testDurationInSec);
+ }
+ for (ScheduledFuture timer : gatewayTelemetryTimers) {
+ timer.cancel(false);
+ }
+ if (onboardingEngine != null) {
+ onboardingEngine.stop();
+ }
if (rpcBurstSender != null) {
rpcBurstSender.stop();
}
@@ -265,8 +456,8 @@ public void runApiTests() throws InterruptedException {
log.info("Gateway RPC drain: waiting for in-flight RPCs to settle (quietSec={}, maxSec={})...",
rpcDrainQuietSec, maxMs / 1000);
GatewayRpcReceiver.DrainResult result = rpcReceiver.drain(quietMs, maxMs, rpcRespond);
- rpcReceiver.finalizeLostReplies(); // replies still buffered (client never reconnected) are lost
- rpcReceiver.logPending(); // name the distinct still-unanswered RPCs for DB EXPIRED correlation
+ rpcReceiver.finalizeLostReplies();
+ rpcReceiver.logPending();
String drainLine = String.format("Gateway RPC drain complete [drained %.1fs, quiesced=%b]",
result.elapsedMs / 1000.0, result.quiesced);
if (result.quiesced) {
@@ -274,11 +465,177 @@ public void runApiTests() throws InterruptedException {
} else {
log.warn(drainLine);
}
- log.info(rpcReceiver.inTotalSummary()); // RPC In [total]: publish=… (new …, redelivered …)
- log.info(rpcReceiver.outTotalSummary()); // RPC Out [total]: publish=…, pubAck=…, failed=…, recovered=…, lost=…
+ log.info(rpcReceiver.inTotalSummary());
+ log.info(rpcReceiver.outTotalSummary());
+ }
+ }
+ }
+
+ private EntityLifecycle gatewayLifecycle() {
+ return new EntityLifecycle() {
+ @Override
+ public int entityCount() {
+ return gatewayEndIdx - gatewayStartIdx;
+ }
+
+ @Override
+ public void onboard(int idx) throws Exception {
+ int gwIdx = gatewayStartIdx + idx;
+ // 1) connect this gateway's client (persistent, autoReconnect via createClient)
+ // 2) announce its sub-devices through deviceAnnouncer.announce(...) (via the inherited warmUpPublish)
+ // 3) subscribe RPC for this client via rpcReceiver (single-client attach) + wire reconnect recovery
+ // 4) schedule this gateway's batch-telemetry timer
+ connectAnnounceSubscribeAndSchedule(gwIdx);
+ }
+ };
+ }
+
+ /**
+ * One STAGGERED gateway's full onboarding step, composed entirely from existing pieces: the same
+ * connect sequence {@link #initClientBlocking} performs for PHASED's bulk connect, the same
+ * per-device announce path ({@link #warmUpPublish}) PHASED's warm-up uses, and the same
+ * subscribe+reconnect wiring {@link #attachClientsToRpc} performs for PHASED's bulk attach — in that
+ * order (connect -> announce -> subscribe), per the {@link EntityLifecycle#onboard} contract.
+ * Synchronous: throws on any failure so the engine counts this gateway as failed rather than
+ * onboarded. The connect step is isolated here; everything after it (the part with a commit-on-success
+ * invariant to preserve) lives in {@link #onboardConnectedGateway}.
+ */
+ private void connectAnnounceSubscribeAndSchedule(int gwIdx) throws Exception {
+ int localIdx = gwIdx - gatewayStartIdx;
+ String gatewayName = staggeredGatewayNames.get(localIdx);
+ List deviceNames = staggeredGatewayDeviceNames.getOrDefault(localIdx, Collections.emptyList());
+
+ // 1) connect (persistent; createClient() applies autoReconnect() same as every other gateway client)
+ MqttClient client = initClientBlocking(gatewayName);
+ onboardConnectedGateway(gwIdx, client, gatewayName, deviceNames);
+ }
+
+ /**
+ * Everything after "the client is already connected": announce -> subscribe -> commit-on-success ->
+ * schedule telemetry. Split out of {@link #connectAnnounceSubscribeAndSchedule} purely so the
+ * commit-on-success invariant is unit-testable with a mocked, already-"connected" {@link MqttClient}
+ * (no real broker needed) — the production call site above still invokes this immediately after a
+ * real connect, so behavior is unchanged.
+ * Nothing is registered into the shared {@code mqttClients}/{@code deviceClients}/
+ * {@code gatewayDeviceNames}/{@code clientNames} collections — which {@link #startRpcBurstSender()},
+ * the telemetry scheduler, and reconnect recovery all read from wholesale — until the ENTIRE sequence
+ * below has succeeded. A gateway that fails partway (e.g. its subscribe throws after announce
+ * succeeded) is closed and left out of every shared collection entirely, so a partial onboarding can
+ * never leak un-announced/un-subscribed devices into the RPC-outcome measurement.
+ */
+ void onboardConnectedGateway(int gwIdx, MqttClient client, String gatewayName, List deviceNames) throws Exception {
+ try {
+ // 2) announce sub-devices through the same reliable (RPC on) / legacy QoS-0 (RPC off) path
+ // warm-up uses. No timeout here: an announce under retry can legitimately take longer than
+ // CONNECT_TIMEOUT; GatewayDeviceAnnouncer always eventually settles the future (acked or
+ // unconfirmed-after-retries).
+ for (String deviceName : deviceNames) {
+ DeviceClient dc = new DeviceClient();
+ dc.setMqttClient(client);
+ dc.setDeviceName(deviceName);
+ warmUpPublish(dc).get();
+ }
+ // 3) subscribe RPC for this client alone + wire its reconnect recovery (mirrors
+ // attachRpcReceiver's per-client wiring, done here per-gateway instead of once in bulk over
+ // mqttClients).
+ if (rpcEnabled) {
+ attachClientsToRpc(Collections.singletonList(client), 0);
}
+ } catch (Exception e) {
+ client.disconnect();
+ throw e;
}
+
+ // Onboarding succeeded end-to-end: only now commit this gateway's client + devices into the
+ // shared collections and start its telemetry timer.
+ mqttClients.add(client);
+ clientNames.put(client, gatewayName);
+ gatewayDeviceNames.put(client, Collections.synchronizedList(new ArrayList<>(deviceNames)));
+ List newDeviceClients = new ArrayList<>(deviceNames.size());
+ for (String deviceName : deviceNames) {
+ DeviceClient dc = new DeviceClient();
+ dc.setMqttClient(client);
+ dc.setDeviceName(deviceName);
+ dc.setGatewayName(gatewayName);
+ newDeviceClients.add(dc);
+ }
+ deviceClients.addAll(newDeviceClients);
+
+ // 4) schedule this gateway's own batch-telemetry timer, starting immediately (its onboarding time
+ // is already spread by the engine's ramp jitter; an extra small per-gateway startup offset avoids
+ // every gateway's tick landing on the exact same millisecond).
+ scheduleGatewayTelemetry(gwIdx, client, gatewayName, deviceNames);
+ }
+
+ /** Blocking connect for one gateway token, identical in behavior to the private {@code initClient} used
+ * by PHASED's {@code connectDevices} — reconstructed here (rather than reused) only because that method
+ * is {@code private} in {@link org.thingsboard.tools.service.shared.BaseMqttAPITest}. */
+ private MqttClient initClientBlocking(String token) throws Exception {
+ MqttClient client = createClient(token);
+ Future connectFuture = connectAsync(client);
+ MqttConnectResult result;
+ try {
+ result = connectFuture.get(CONNECT_TIMEOUT, TimeUnit.SECONDS);
+ } catch (TimeoutException ex) {
+ connectFuture.cancel(true);
+ client.disconnect();
+ throw new RuntimeException(String.format("STAGGERED: timed out connecting gateway [%s]", token), ex);
+ }
+ if (!result.isSuccess()) {
+ connectFuture.cancel(true);
+ client.disconnect();
+ throw new RuntimeException(String.format("STAGGERED: failed to connect gateway [%s]. Result code: %s", token, result.getReturnCode()));
+ }
+ return client;
+ }
+
+ /** Schedules this gateway's own periodic batch-telemetry publish (one MQTT publish carrying all of its
+ * sub-devices' next messages, same construction as {@link MqttGatewayBatchAPITest#nextPublishTask}),
+ * independent of every other gateway's timer. No-op when publishing is disabled (MESSAGES_PER_SECOND=0)
+ * or this gateway has no sub-devices.
+ * Period is MPS-derived so STAGGERED's steady-state aggregate matches PHASED's: today's metronome
+ * does {@code testMessagesPerSecond} gateway-batch publishes/sec by sweeping the whole fleet, i.e.
+ * each gateway publishes once every {@code entityCount / testMessagesPerSecond} seconds — so each
+ * independent per-gateway timer here fires on that same period, jittered so the first fires aren't
+ * synchronized across gateways.
+ *
Package-private (not private): exercised directly by MqttGatewayAPITestTest. */
+ void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatewayName, List deviceNames) {
+ if (testMessagesPerSecond <= 0 || deviceNames.isEmpty()) {
+ return;
+ }
+ DeviceClient logClient = new DeviceClient();
+ logClient.setMqttClient(client);
+ logClient.setGatewayName(gatewayName);
+ logClient.setDeviceName("batch[" + deviceNames.size() + " devices]");
+ AtomicInteger tick = new AtomicInteger();
+ int entityCount = gatewayEndIdx - gatewayStartIdx;
+ long periodMs = Math.max(1L, (entityCount * 1000L) / testMessagesPerSecond);
+ long initialJitterMs = EphemeralSchedule.firstOffsetMillis(new Random(seed + gwIdx), periodMs);
+ ScheduledFuture timer = restClientService.getScheduler().scheduleAtFixedRate(() -> {
+ try {
+ ObjectNode batch = mapper.createObjectNode();
+ for (String deviceName : deviceNames) {
+ NodeMsg nodeMsg = getNextNodeMessage(deviceName, false);
+ batch.setAll(nodeMsg.getNode());
+ }
+ byte[] data = mapper.writeValueAsBytes(batch);
+ int iteration = tick.incrementAndGet();
+ client.publish(getTestTopic(), Unpooled.wrappedBuffer(data), MqttQoS.AT_MOST_ONCE)
+ .addListener(f -> {
+ if (f.isSuccess()) {
+ totalSuccessPublishedCount.incrementAndGet();
+ logSuccessTestMessage(iteration, logClient);
+ } else {
+ totalFailedPublishedCount.incrementAndGet();
+ logFailureTestMessage(iteration, logClient, f);
+ }
+ });
+ } catch (Exception e) {
+ log.warn("STAGGERED telemetry publish failed for gateway [{}]", gatewayName, e);
+ }
+ }, initialJitterMs, periodMs, TimeUnit.MILLISECONDS);
+ gatewayTelemetryTimers.add(timer);
}
private void startRpcBurstSender() {
@@ -295,7 +652,8 @@ private void startRpcBurstSender() {
rpcBurstSender = new RpcBurstSender(
restClientService.getRestClient(), restUrl, deviceNames, template,
rpcSenderQueue, rpcSenderTimeoutMs, rpcSenderChunkSize,
- rpcSenderIntervalSec, rpcSenderStartDelaySec, RpcBurstSender.Mode.fromConfig(rpcSenderMode));
+ rpcSenderIntervalSec, rpcSenderStartDelaySec, RpcBurstSender.Mode.fromConfig(rpcSenderMode),
+ rpcSenderFireThreads, rpcSenderMaxInFlight);
rpcBurstSender.start();
}
@@ -332,6 +690,14 @@ protected void logFailureTestMessage(int iteration, DeviceClient client, Future<
}
protected void attachRpcReceiver() throws InterruptedException {
+ initRpcReceiver();
+ attachClientsToRpc(mqttClients, warmUpPackSize);
+ }
+
+ /** Builds {@code rpcReceiver}/{@code deviceAnnouncer} and registers their stats blocks. Split out of
+ * {@link #attachRpcReceiver()} (which still does exactly this + the bulk attach below, unchanged)
+ * so STAGGERED can construct these once, up front, before any gateway has connected. */
+ private void initRpcReceiver() {
ObjectMapper mapper = new ObjectMapper();
RpcResponseTemplate template = rpcRespond ? RpcResponseTemplate.load(rpcResponseTemplate) : null;
RpcMessageProcessor processor = new RpcMessageProcessor(mapper, rpcSendTsPath, rpcRespond, template);
@@ -346,11 +712,18 @@ protected void attachRpcReceiver() throws InterruptedException {
statsReporter().register(StatsBlock.RPC_SUBSCRIPTION, rpcReceiver::subscriptionSummary);
statsReporter().register(StatsBlock.RPC_IN, rpcReceiver::inSummary);
statsReporter().register(StatsBlock.RPC_OUT, rpcReceiver::outSummary);
- rpcReceiver.attach(mqttClients, warmUpPackSize);
+ }
+
+ /** Subscribes the given clients to the RPC topic and wires each one's reconnect recovery. Split out
+ * of {@link #attachRpcReceiver()} (unchanged for PHASED: called once with {@code mqttClients} +
+ * {@code warmUpPackSize}) so STAGGERED can call it per-gateway with a singleton list + packSize=0
+ * (pacing is meaningless for a single client). */
+ private void attachClientsToRpc(List clients, int packSize) throws InterruptedException {
+ rpcReceiver.attach(clients, packSize);
// On reconnect, a gateway loses its RPC subscription (cleanSession) and its server-side
// sub-device routing; restore both so RPC delivery resumes instead of silently dropping.
// Also flush any replies buffered while the channel was down so they land within the RPC expiry.
- for (MqttClient client : mqttClients) {
+ for (MqttClient client : clients) {
setReconnectAction(client, () -> {
rpcReceiver.resubscribe(client);
reannounceDevices(client);
diff --git a/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java b/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java
index 6419444e..757eaca8 100644
--- a/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java
+++ b/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java
@@ -71,6 +71,10 @@ public class GatewayRpcReceiver {
// Clients whose current v1/gateway/rpc subscription is not (yet) SUBACK-confirmed — a live gauge,
// so a slow-but-real SUBACK is never a false positive (it self-clears whenever the SUBACK lands).
private final java.util.Set unconfirmedSubscriptions = ConcurrentHashMap.newKeySet();
+ // The RPC stats legend is one-time ceremony. PHASED calls attach() once (bulk), but STAGGERED calls it
+ // per gateway (singleton list), so without this guard the multi-line legend would repeat ~once per
+ // gateway and flood the log during onboarding. Log it exactly once, on the first attach().
+ private final AtomicBoolean legendLogged = new AtomicBoolean(false);
private static final long DRAIN_POLL_MS = 500L;
@@ -98,7 +102,9 @@ public GatewayRpcReceiver(String topic, MqttQoS qos, RpcMessageProcessor process
}
public void attach(List clients, int packSize) throws InterruptedException {
- log.info("Gateway RPC stats key:\n{}", RpcLatencyStats.legend());
+ if (legendLogged.compareAndSet(false, true)) {
+ log.info("Gateway RPC stats key:\n{}", RpcLatencyStats.legend());
+ }
int n = 0;
for (MqttClient client : clients) {
subscribe(client);
@@ -109,7 +115,12 @@ public void attach(List clients, int packSize) throws InterruptedExc
Thread.sleep(100 + ThreadLocalRandom.current().nextInt(100));
}
}
- log.info("Subscribed {} gateways to RPC topic {}", clients.size(), topic);
+ // Only log the batch summary for a true bulk attach (PHASED). STAGGERED attaches one gateway at a
+ // time, so logging here would repeat once per gateway; the STAGGERED ramp-complete line reports the
+ // total instead.
+ if (clients.size() > 1) {
+ log.info("Subscribed {} gateways to RPC topic {}", clients.size(), topic);
+ }
}
/** Re-issue the RPC-topic subscription for one client after it reconnects (netty-mqtt clears all
diff --git a/src/main/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSender.java b/src/main/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSender.java
index f8713973..4dab3206 100644
--- a/src/main/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSender.java
+++ b/src/main/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSender.java
@@ -27,16 +27,20 @@
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@Slf4j
@@ -68,8 +72,6 @@ static int chunkIndexForTick(long tickNumber, int numChunks) {
return (int) (tickNumber % numChunks);
}
- private static final int MAX_FIRE_THREADS = 16;
-
private final RestClient restClient;
private final String restUrl;
private final List deviceNames;
@@ -80,10 +82,14 @@ static int chunkIndexForTick(long tickNumber, int numChunks) {
private final int intervalSec;
private final int startDelaySec;
private final Mode mode;
+ private final int maxFireThreads;
+ private final int maxInFlight;
private static final ObjectMapper MAPPER = new ObjectMapper();
private ScheduledExecutorService scheduler;
private ExecutorService firePool;
+ private HttpClient httpClient;
+ private Semaphore inFlight;
private ScheduledFuture burstFuture;
private List> chunks;
private String ruleEngineUrl;
@@ -91,10 +97,14 @@ static int chunkIndexForTick(long tickNumber, int numChunks) {
private long spreadTickMs; // SPREAD inter-chunk period, set in start()
private final AtomicLong burstsFired = new AtomicLong();
private final AtomicLong devicesDispatched = new AtomicLong();
+ private final AtomicLong postsOk = new AtomicLong();
+ private final AtomicLong postsFailed = new AtomicLong();
+ private final AtomicLong postsDropped = new AtomicLong();
public RpcBurstSender(RestClient restClient, String restUrl, List deviceNames,
JsonNode commandTemplate, String queue, int timeoutMs, int chunkSize,
- int intervalSec, int startDelaySec, Mode mode) {
+ int intervalSec, int startDelaySec, Mode mode, int maxFireThreads,
+ int maxInFlight) {
this.restClient = restClient;
this.restUrl = restUrl;
this.deviceNames = deviceNames;
@@ -105,6 +115,8 @@ public RpcBurstSender(RestClient restClient, String restUrl, List device
this.intervalSec = intervalSec;
this.startDelaySec = startDelaySec;
this.mode = mode;
+ this.maxFireThreads = maxFireThreads;
+ this.maxInFlight = maxInFlight;
}
public void start() {
@@ -120,8 +132,19 @@ public void start() {
ruleEngineUrl = restUrl + "/api/rule-engine/USER/" + userId + "/" + queue + "/" + timeoutMs;
chunks = chunk(deviceNames, chunkSize);
- int fireThreads = Math.min(Math.max(1, chunks.size()), MAX_FIRE_THREADS);
+ int fireThreads = Math.min(Math.max(1, chunks.size()), Math.max(1, maxFireThreads));
firePool = Executors.newFixedThreadPool(fireThreads, ThingsBoardThreadFactory.forName("rpc-burst-fire"));
+ // Submissions are ASYNCHRONOUS: the scheduler hands a chunk to the HTTP client and returns, so the
+ // cadence is never coupled to how long the rule engine takes to reply. The reply is still consumed —
+ // its status feeds ok/failed accounting — it just no longer holds a thread. `inFlight` bounds
+ // outstanding requests so a slow server cannot grow this pod without limit.
+ inFlight = new Semaphore(Math.max(1, maxInFlight));
+ httpClient = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofSeconds(10))
+ .executor(firePool)
+ .build();
+ log.info("RPC sender: async submissions, callback pool {} threads, max in-flight {} ({} chunks of {})",
+ fireThreads, maxInFlight, chunks.size(), chunkSize);
scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("rpc-burst-sched"));
long intervalMs = intervalSec * 1000L;
@@ -148,64 +171,84 @@ public void start() {
}
}
+ /**
+ * Submit one chunk without waiting for the reply. The reply is still consumed — its status drives the
+ * ok/failed counters and the slow-post warning — but on the HTTP client's callback thread, so neither
+ * the scheduler nor the caller is held for the round trip. Returns immediately.
+ *
+ * If {@code maxInFlight} outstanding requests are already pending the chunk is DROPPED rather than
+ * queued or blocked on: silently blocking here would re-couple the offered rate to server latency, which
+ * is exactly what this sender must not do. Drops are counted and logged so a shortfall is visible.
+ */
+ private void postChunkAsync(List deviceChunk, String label, long slowThresholdMs) {
+ if (!inFlight.tryAcquire()) {
+ postsDropped.incrementAndGet();
+ log.warn("RPC {} chunk DROPPED ({} devices): {} requests already in flight — the server is not keeping up with the offered rate",
+ label, deviceChunk.size(), maxInFlight);
+ return;
+ }
+ long startedAt = System.currentTimeMillis();
+ HttpRequest request;
+ try {
+ request = HttpRequest.newBuilder(URI.create(ruleEngineUrl))
+ .timeout(Duration.ofMillis(timeoutMs))
+ .header("Content-Type", "application/json")
+ .header("X-Authorization", "Bearer " + restClient.getToken())
+ .POST(HttpRequest.BodyPublishers.ofString(
+ buildBody(MAPPER, commandTemplate, deviceChunk).toString()))
+ .build();
+ } catch (Exception e) {
+ inFlight.release();
+ postsFailed.incrementAndGet();
+ log.warn("RPC {} chunk could not be built ({} devices): {}", label, deviceChunk.size(), e.getMessage());
+ return;
+ }
+ httpClient.sendAsync(request, HttpResponse.BodyHandlers.discarding())
+ .whenComplete((response, error) -> {
+ inFlight.release();
+ long elapsed = System.currentTimeMillis() - startedAt;
+ if (error != null) {
+ postsFailed.incrementAndGet();
+ log.warn("RPC {} chunk failed ({} devices) after {}ms: {}", label, deviceChunk.size(), elapsed, error.getMessage());
+ return;
+ }
+ int status = response.statusCode();
+ if (status >= 200 && status < 300) {
+ postsOk.incrementAndGet();
+ recordDispatched(deviceChunk.size()); // count dispatched only on a successful post (D4)
+ } else {
+ postsFailed.incrementAndGet();
+ log.warn("RPC {} chunk rejected ({} devices) after {}ms: HTTP {}", label, deviceChunk.size(), elapsed, status);
+ }
+ if (elapsed > slowThresholdMs) {
+ log.warn("RPC {} chunk reply took {}ms > {}ms — the rule engine is slow to answer (submission rate is unaffected)",
+ label, elapsed, slowThresholdMs);
+ }
+ });
+ }
+
private void fireBurst() {
recordBurstFired();
- long startedAt = System.currentTimeMillis();
- AtomicInteger ok = new AtomicInteger();
- AtomicInteger failed = new AtomicInteger();
- List> futures = new ArrayList<>();
for (List deviceChunk : chunks) {
- futures.add(firePool.submit(() -> {
- try {
- restClient.getRestTemplate().postForEntity(ruleEngineUrl, buildBody(MAPPER, commandTemplate, deviceChunk), String.class);
- ok.incrementAndGet();
- recordDispatched(deviceChunk.size()); // count dispatched only after a successful post (D4)
- } catch (Exception e) {
- failed.incrementAndGet();
- log.warn("RPC burst chunk failed ({} devices): {}", deviceChunk.size(), e.getMessage());
- }
- }));
- }
- for (Future f : futures) {
- try {
- f.get(timeoutMs + 5000L, TimeUnit.MILLISECONDS);
- } catch (Exception e) {
- log.warn("RPC burst chunk did not complete in time", e);
- }
+ postChunkAsync(deviceChunk, "burst", intervalSec * 1000L);
}
- long elapsed = System.currentTimeMillis() - startedAt;
- if (elapsed > intervalSec * 1000L) {
- log.warn("RPC burst took {}ms, exceeding the {}s interval — burst cadence will slip", elapsed, intervalSec);
- }
- log.info("RPC burst: devices={}, chunks ok={}, failed={}, elapsed={}ms",
- deviceNames.size(), ok.get(), failed.get(), elapsed);
+ log.info("RPC burst {}: {} chunks submitted ({} devices); ok={} failed={} dropped={} so far",
+ burstsFired.get(), chunks.size(), deviceNames.size(), postsOk.get(), postsFailed.get(), postsDropped.get());
}
/**
* SPREAD mode: fire ONE chunk per tick, rotating through the chunk list so each chunk is sent once
- * per sweep (= one interval), staggered rather than all at once. Posts are async (firePool) so a slow
- * REST call never delays the next tick; a full sweep is counted as one "burst" for the stats.
+ * per sweep (= one interval), staggered rather than all at once.
*/
private void fireNextSpreadChunk() {
int numChunks = chunks.size();
List deviceChunk = chunks.get(chunkIndexForTick(spreadTick, numChunks));
- firePool.submit(() -> {
- long startedAt = System.currentTimeMillis();
- try {
- restClient.getRestTemplate().postForEntity(ruleEngineUrl, buildBody(MAPPER, commandTemplate, deviceChunk), String.class);
- recordDispatched(deviceChunk.size()); // count dispatched only after a successful post (D4)
- long elapsed = System.currentTimeMillis() - startedAt;
- if (elapsed > spreadTickMs) {
- log.warn("RPC spread chunk post took {}ms > {}ms tick — drip falling behind", elapsed, spreadTickMs);
- }
- } catch (Exception e) {
- log.warn("RPC spread chunk failed ({} devices): {}", deviceChunk.size(), e.getMessage());
- }
- });
+ postChunkAsync(deviceChunk, "spread", spreadTickMs);
spreadTick++;
if (chunkIndexForTick(spreadTick, numChunks) == 0) { // wrapped: a full sweep of all chunks = one interval
recordBurstFired();
- log.info("RPC spread: sweep {} dispatched ({} device-RPCs total so far)", burstsFired.get(), devicesDispatched.get());
+ log.info("RPC spread: sweep {} dispatched ({} device-RPCs total; ok={} failed={} dropped={})",
+ burstsFired.get(), devicesDispatched.get(), postsOk.get(), postsFailed.get(), postsDropped.get());
}
}
@@ -220,18 +263,30 @@ void recordDispatched(int deviceCount) {
}
String dispatchSummary() {
- return String.format("RPC burst sender stopped: %d bursts fired, %d device-RPCs dispatched",
- burstsFired.get(), devicesDispatched.get());
+ return String.format("RPC burst sender stopped: %d bursts fired, %d device-RPCs dispatched (posts ok=%d failed=%d dropped=%d)",
+ burstsFired.get(), devicesDispatched.get(), postsOk.get(), postsFailed.get(), postsDropped.get());
}
public void stop() {
- log.info(dispatchSummary());
+ // Stop scheduling first, then give outstanding replies a moment to land so the summary counts them.
if (burstFuture != null) {
burstFuture.cancel(true);
}
if (scheduler != null) {
scheduler.shutdownNow();
}
+ if (inFlight != null) {
+ int outstanding = maxInFlight - inFlight.availablePermits();
+ if (outstanding > 0) {
+ log.info("RPC sender: waiting up to {}ms for {} in-flight posts to complete", timeoutMs, outstanding);
+ try {
+ inFlight.tryAcquire(maxInFlight, timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ log.info(dispatchSummary());
if (firePool != null) {
firePool.shutdownNow();
}
diff --git a/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java b/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java
index 078494f0..7d53781b 100644
--- a/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java
+++ b/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java
@@ -119,6 +119,13 @@ protected synchronized StatsReporter statsReporter() {
@Value("${gateway.overwriteActivityTime:false}")
protected boolean gatewayOverwriteActivityTime;
+ @Value("${onboard.mode:PHASED}")
+ protected String onboardMode;
+ @Value("${onboard.maxConcurrent:200}")
+ protected int onboardMaxConcurrent;
+ @Value("${onboard.firstJitterSec:60}")
+ protected int onboardFirstJitterSec;
+
@Autowired
@Qualifier("randomTelemetryGenerator")
protected MessageGenerator tsMsgGenerator;
diff --git a/src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java b/src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java
new file mode 100644
index 00000000..ddbf6259
--- /dev/null
+++ b/src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java
@@ -0,0 +1,33 @@
+/**
+ * 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.tools.service.shared.onboarding;
+
+/**
+ * One entity's onboarding step for the staggered mode: bring a single gateway/device fully online
+ * (connect -> optionally announce sub-devices -> subscribe -> start its own telemetry cadence).
+ * Implementations are mode-specific; the engine only paces the calls.
+ */
+public interface EntityLifecycle {
+
+ /** Number of entities to onboard; the engine drives indices [0, entityCount()). */
+ int entityCount();
+
+ /**
+ * Onboard entity {@code idx} synchronously. Must throw on failure — the engine counts it as
+ * failed, releases its slot, and continues (never blocks the ramp).
+ */
+ void onboard(int idx) throws Exception;
+}
diff --git a/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java b/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java
new file mode 100644
index 00000000..b2dbe415
--- /dev/null
+++ b/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java
@@ -0,0 +1,146 @@
+/**
+ * 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.tools.service.shared.onboarding;
+
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.tools.service.gateway.EphemeralSchedule;
+
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Mode-agnostic paced onboarding: schedules each entity's first onboard over a jittered window and
+ * runs at most {@code maxConcurrentOnboards} onboards at once, signalling ramp-complete when every
+ * entity has reached a terminal state (onboarded or failed). Reuses the ephemeral engine's
+ * tryAcquire+reschedule pacing, but drives a persistent (synchronous) onboard instead of a churn cycle.
+ */
+@Slf4j
+public class StaggeredOnboardingEngine {
+
+ public interface RampCompleteCallback {
+ void onRampComplete(int onboarded, int failed);
+ }
+
+ private final EntityLifecycle lifecycle;
+ private final int maxConcurrentOnboards;
+ private final long firstJitterMillis;
+ private final Random rng;
+
+ private final ScheduledExecutorService timer;
+ private final ExecutorService workers;
+ private final Semaphore permits;
+
+ private final AtomicInteger onboarded = new AtomicInteger();
+ private final AtomicInteger failed = new AtomicInteger();
+ private final AtomicInteger terminal = new AtomicInteger();
+ private volatile boolean running;
+ private volatile RampCompleteCallback onComplete;
+ // Periodic onboarding-progress line (replaces per-entity subscribe/announce log spam under STAGGERED);
+ // cancelled at ramp-complete so the final "Ramp complete: ..." line closes it out.
+ private volatile ScheduledFuture progressTask;
+
+ private static final long PROGRESS_LOG_INTERVAL_SEC = 10L;
+
+ public StaggeredOnboardingEngine(EntityLifecycle lifecycle, int maxConcurrentOnboards,
+ int firstJitterSec, int schedulerThreads, long seed) {
+ this.lifecycle = lifecycle;
+ this.maxConcurrentOnboards = Math.max(1, maxConcurrentOnboards);
+ this.firstJitterMillis = Math.max(0, firstJitterSec) * 1000L;
+ this.rng = new Random(EphemeralSchedule.scheduleSeed(seed, 0));
+ this.timer = Executors.newScheduledThreadPool(Math.max(1, schedulerThreads));
+ this.workers = Executors.newFixedThreadPool(this.maxConcurrentOnboards);
+ this.permits = new Semaphore(this.maxConcurrentOnboards);
+ }
+
+ public void start(RampCompleteCallback cb) {
+ this.onComplete = cb;
+ this.running = true;
+ int count = lifecycle.entityCount();
+ log.info("Staggered onboarding starting: {} entities, maxConcurrent={}, firstJitter={}ms",
+ count, maxConcurrentOnboards, firstJitterMillis);
+ if (count <= 0) {
+ fireComplete();
+ return;
+ }
+ for (int i = 0; i < count; i++) {
+ final int idx = i;
+ long offset = EphemeralSchedule.firstOffsetMillis(rng, firstJitterMillis);
+ timer.schedule(() -> onboardOne(idx), offset, TimeUnit.MILLISECONDS);
+ }
+ this.progressTask = timer.scheduleAtFixedRate(this::logProgress,
+ PROGRESS_LOG_INTERVAL_SEC, PROGRESS_LOG_INTERVAL_SEC, TimeUnit.SECONDS);
+ }
+
+ private void logProgress() {
+ log.info("STAGGERED onboarding progress: {} / {} onboarded, {} in-flight, {} failed",
+ String.format(java.util.Locale.US, "%,d", onboarded.get()),
+ String.format(java.util.Locale.US, "%,d", lifecycle.entityCount()),
+ inFlightCount(), failed.get());
+ }
+
+ private void onboardOne(int idx) {
+ if (!running) {
+ return;
+ }
+ if (!permits.tryAcquire()) {
+ // no free slot: reschedule on the timer (never block a timer thread)
+ timer.schedule(() -> onboardOne(idx), 1 + rng.nextInt(50), TimeUnit.MILLISECONDS);
+ return;
+ }
+ workers.submit(() -> {
+ try {
+ lifecycle.onboard(idx);
+ onboarded.incrementAndGet();
+ } catch (Exception e) {
+ failed.incrementAndGet();
+ log.warn("Onboard failed for entity {}: {}", idx, e.toString());
+ } finally {
+ permits.release();
+ if (terminal.incrementAndGet() == lifecycle.entityCount()) {
+ fireComplete();
+ }
+ }
+ });
+ }
+
+ private void fireComplete() {
+ ScheduledFuture pt = this.progressTask;
+ if (pt != null) {
+ pt.cancel(false);
+ }
+ log.info("Ramp complete: {} onboarded, {} failed", onboarded.get(), failed.get());
+ RampCompleteCallback cb = this.onComplete;
+ if (cb != null) {
+ cb.onRampComplete(onboarded.get(), failed.get());
+ }
+ }
+
+ public void stop() {
+ running = false;
+ timer.shutdownNow();
+ workers.shutdownNow();
+ }
+
+ public int onboardedCount() { return onboarded.get(); }
+ public int failedCount() { return failed.get(); }
+ public int inFlightCount() { return maxConcurrentOnboards - permits.availablePermits(); }
+}
diff --git a/src/main/resources/tb-ce-performance-tests.yml b/src/main/resources/tb-ce-performance-tests.yml
index 8118b858..0e5cc5ef 100644
--- a/src/main/resources/tb-ce-performance-tests.yml
+++ b/src/main/resources/tb-ce-performance-tests.yml
@@ -199,6 +199,14 @@ gateway:
chunkSize: "${GATEWAY_RPC_SENDER_CHUNK_SIZE:500}" # devices per rule-engine REST call (TBEL result-size bound)
queue: "${GATEWAY_RPC_SENDER_QUEUE:RpcCalls}" # rule-engine queue name in the URL
timeoutMs: "${GATEWAY_RPC_SENDER_TIMEOUT_MS:10000}" # rule-engine call timeout; MUST equal the rule chain's hardcoded TIMEOUT_MS
+ fireThreads: "${GATEWAY_RPC_SENDER_FIRE_THREADS:48}" # size of the HTTP client's callback pool (reply handling and accounting). Submissions are async, so this does NOT bound the offered rate
+ maxInFlight: "${GATEWAY_RPC_SENDER_MAX_IN_FLIGHT:512}" # cap on outstanding (unanswered) chunk posts, so a slow server cannot grow this pod without limit. Beyond it chunks are DROPPED and counted rather than blocking the cadence — a non-zero dropped count means the server could not absorb the offered rate
+
+# Onboarding strategy for the persistent gateway/device modes.
+onboard:
+ mode: "${ONBOARD_MODE:PHASED}" # PHASED (default, today's warmup) | STAGGERED
+ maxConcurrent: "${ONBOARD_MAX_CONCURRENT:200}" # STAGGERED: max entities onboarding at once
+ firstJitterSec: "${ONBOARD_FIRST_JITTER_SEC:60}" # STAGGERED: each entity's first onboard spread over [0,this)s
customer:
startIdx: "${CUSTOMER_START_IDX:0}"
diff --git a/src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java b/src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java
new file mode 100644
index 00000000..18c2178e
--- /dev/null
+++ b/src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java
@@ -0,0 +1,121 @@
+/**
+ * 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.tools.service.device;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.thingsboard.mqtt.MqttClient;
+import org.thingsboard.tools.service.shared.RestClientService;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+// Same broker-free idiom as MqttGatewayAPITestTest: this test class IS the SUT (extends
+// MqttDeviceAPITest) so it can reach the package-private/protected inherited state
+// (deviceStartIdx/deviceEndIdx/mqttClients/deviceClients/seed/...) without a Spring context or a real
+// MQTT broker. init()/@PostConstruct is never invoked, so every field a test relies on is set explicitly.
+class MqttDeviceAPITestTest extends MqttDeviceAPITest {
+
+ ScheduledExecutorService scheduler;
+
+ @BeforeEach
+ void setUp() {
+ scheduler = mock(ScheduledExecutorService.class);
+ RestClientService rcs = mock(RestClientService.class);
+ when(rcs.getScheduler()).thenReturn(scheduler);
+ restClientService = rcs;
+ }
+
+ // --- scheduleDeviceTelemetry(): MPS-derived period + jitter, and the no-timer guard ---
+
+ @Test
+ void telemetryPeriodIsDerivedFromEntityCountAndMessagesPerSecond() {
+ deviceStartIdx = 0;
+ deviceEndIdx = 10; // entityCount = 10
+ testMessagesPerSecond = 5;
+ seed = 0;
+
+ // Raw type (not ScheduledFuture): a wildcard-typed mock hits a generic-capture mismatch on
+ // thenReturn (javac can't unify two independent "capture of ?" instantiations).
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ ScheduledFuture fakeFuture = mock(ScheduledFuture.class);
+ when(scheduler.scheduleAtFixedRate(any(), anyLong(), anyLong(), eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(fakeFuture);
+
+ MqttClient client = mock(MqttClient.class);
+ scheduleDeviceTelemetry(3, client, "DW00000003");
+
+ ArgumentCaptor delayCaptor = ArgumentCaptor.forClass(Long.class);
+ ArgumentCaptor periodCaptor = ArgumentCaptor.forClass(Long.class);
+ verify(scheduler).scheduleAtFixedRate(any(), delayCaptor.capture(), periodCaptor.capture(), eq(TimeUnit.MILLISECONDS));
+
+ long expectedPeriodMs = (10 * 1000L) / 5; // entityCount * 1000 / MPS = 2000ms
+ assertThat(periodCaptor.getValue()).isEqualTo(expectedPeriodMs);
+ assertThat(delayCaptor.getValue()).isBetween(0L, expectedPeriodMs - 1);
+ assertThat(deviceTelemetryTimers).containsExactly(fakeFuture);
+ }
+
+ @Test
+ void noTelemetryTimerScheduledWhenMessagesPerSecondIsZeroOrLess() {
+ deviceStartIdx = 0;
+ deviceEndIdx = 10;
+ testMessagesPerSecond = 0; // no-publish mode: mirrors AbstractAPITest.runApiTests' no-publish branch
+
+ MqttClient client = mock(MqttClient.class);
+ scheduleDeviceTelemetry(0, client, "DW00000000");
+
+ verify(scheduler, never()).scheduleAtFixedRate(any(), anyLong(), anyLong(), any());
+ assertThat(deviceTelemetryTimers).isEmpty();
+ }
+
+ // --- checkStaggeredSupported(): fail fast on the unsupported configs instead of silently diverging ---
+
+ @Test
+ void checkStaggeredSupportedThrowsWhenAlarmsEnabled() {
+ alarmsPerSecond = 1;
+ assertThatThrownBy(this::checkStaggeredSupported).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void checkStaggeredSupportedPassesWhenAlarmsDisabled() {
+ alarmsPerSecond = 0;
+ checkStaggeredSupported(); // must not throw
+ }
+
+ // --- prepareStaggeredModel(): same device-name resolution as PHASED's connectDevices() body ---
+
+ @Test
+ void prepareStaggeredModelResolvesNamesFromIndexRangeWhenNoDevicesLoaded() {
+ deviceStartIdx = 100;
+ deviceEndIdx = 103;
+
+ prepareStaggeredModel();
+
+ assertThat(staggeredDeviceNames).hasSize(3);
+ }
+}
diff --git a/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java b/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java
new file mode 100644
index 00000000..83fc9b5a
--- /dev/null
+++ b/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java
@@ -0,0 +1,226 @@
+/**
+ * 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.tools.service.gateway;
+
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import io.netty.util.concurrent.Promise;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.thingsboard.mqtt.MqttClient;
+import org.thingsboard.tools.service.mqtt.DeviceClient;
+import org.thingsboard.tools.service.shared.RestClientService;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+// Same broker-free idiom as MqttGatewayBatchAPITestTest: this test class IS the SUT (extends
+// MqttGatewayAPITest) so it can reach the package-private/protected inherited state
+// (gatewayStartIdx/gatewayEndIdx/deviceStartIdx/deviceEndIdx/mqttClients/deviceClients/seed/...) without
+// a Spring context or a real MQTT broker. init()/@PostConstruct is never invoked, so every field a test
+// relies on is set explicitly.
+class MqttGatewayAPITestTest extends MqttGatewayAPITest {
+
+ ScheduledExecutorService scheduler;
+
+ @BeforeEach
+ void setUp() {
+ scheduler = mock(ScheduledExecutorService.class);
+ RestClientService rcs = mock(RestClientService.class);
+ when(rcs.getScheduler()).thenReturn(scheduler);
+ restClientService = rcs;
+ }
+
+ // --- prepareStaggeredModel() vs mapDevicesToGatewayClientConnections(): same device assignment ---
+
+ @Test
+ void prepareStaggeredModelAssignsDevicesIdenticallyToPhasedMapping() {
+ gatewayStartIdx = 0;
+ gatewayEndIdx = 3;
+ deviceStartIdx = 0;
+ deviceEndIdx = 7; // not a multiple of 3: exercises the uneven-remainder case too
+
+ // PHASED reference: 3 already-connected gateway clients, grouped by mqttClients position.
+ MqttClient gw0 = mock(MqttClient.class);
+ MqttClient gw1 = mock(MqttClient.class);
+ MqttClient gw2 = mock(MqttClient.class);
+ mqttClients.add(gw0);
+ mqttClients.add(gw1);
+ mqttClients.add(gw2);
+ clientNames.put(gw0, "GW00000000");
+ clientNames.put(gw1, "GW00000001");
+ clientNames.put(gw2, "GW00000002");
+ mapDevicesToGatewayClientConnections();
+
+ Map> phasedGroups = new HashMap<>();
+ for (DeviceClient dc : deviceClients) {
+ phasedGroups.computeIfAbsent(dc.getMqttClient(), k -> new ArrayList<>()).add(dc.getDeviceName());
+ }
+
+ // STAGGERED model: index-based, built without any connected client.
+ prepareStaggeredModel();
+
+ List byPosition = List.of(gw0, gw1, gw2);
+ for (int gwIdx = 0; gwIdx < 3; gwIdx++) {
+ List staggered = staggeredGatewayDeviceNames.getOrDefault(gwIdx, List.of());
+ List phased = phasedGroups.getOrDefault(byPosition.get(gwIdx), List.of());
+ assertThat(staggered).containsExactlyElementsOf(phased);
+ }
+ assertThat(staggeredGatewayNames).containsExactly("GW00000000", "GW00000001", "GW00000002");
+ }
+
+ // --- scheduleGatewayTelemetry(): MPS-derived period + jitter, and the no-timer guard ---
+
+ @Test
+ void telemetryPeriodIsDerivedFromEntityCountAndMessagesPerSecond() {
+ gatewayStartIdx = 0;
+ gatewayEndIdx = 10; // entityCount = 10
+ testMessagesPerSecond = 5;
+ seed = 0;
+
+ // Raw type (not ScheduledFuture): a wildcard-typed mock hits a generic-capture mismatch on
+ // thenReturn (javac can't unify two independent "capture of ?" instantiations).
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ ScheduledFuture fakeFuture = mock(ScheduledFuture.class);
+ when(scheduler.scheduleAtFixedRate(any(), anyLong(), anyLong(), eq(TimeUnit.MILLISECONDS)))
+ .thenReturn(fakeFuture);
+
+ MqttClient client = mock(MqttClient.class);
+ scheduleGatewayTelemetry(3, client, "GW00000003", List.of("DW00000000"));
+
+ ArgumentCaptor delayCaptor = ArgumentCaptor.forClass(Long.class);
+ ArgumentCaptor periodCaptor = ArgumentCaptor.forClass(Long.class);
+ verify(scheduler).scheduleAtFixedRate(any(), delayCaptor.capture(), periodCaptor.capture(), eq(TimeUnit.MILLISECONDS));
+
+ long expectedPeriodMs = (10 * 1000L) / 5; // entityCount * 1000 / MPS = 2000ms
+ assertThat(periodCaptor.getValue()).isEqualTo(expectedPeriodMs);
+ assertThat(delayCaptor.getValue()).isBetween(0L, expectedPeriodMs - 1);
+ assertThat(gatewayTelemetryTimers).containsExactly(fakeFuture);
+ }
+
+ @Test
+ void noTelemetryTimerScheduledWhenMessagesPerSecondIsZeroOrLess() {
+ gatewayStartIdx = 0;
+ gatewayEndIdx = 10;
+ testMessagesPerSecond = 0; // no-publish mode: mirrors AbstractAPITest.runApiTests' no-publish branch
+
+ MqttClient client = mock(MqttClient.class);
+ scheduleGatewayTelemetry(0, client, "GW00000000", List.of("DW00000000"));
+
+ verify(scheduler, never()).scheduleAtFixedRate(any(), anyLong(), anyLong(), any());
+ assertThat(gatewayTelemetryTimers).isEmpty();
+ }
+
+ @Test
+ void noTelemetryTimerScheduledWhenGatewayHasNoSubDevices() {
+ gatewayStartIdx = 0;
+ gatewayEndIdx = 10;
+ testMessagesPerSecond = 5;
+
+ MqttClient client = mock(MqttClient.class);
+ scheduleGatewayTelemetry(0, client, "GW00000000", List.of());
+
+ verify(scheduler, never()).scheduleAtFixedRate(any(), anyLong(), anyLong(), any());
+ assertThat(gatewayTelemetryTimers).isEmpty();
+ }
+
+ // --- checkStaggeredSupported(): fail fast on the unsupported configs instead of silently diverging ---
+
+ @Test
+ void checkStaggeredSupportedThrowsWhenGatewayBatchDisabled() {
+ gatewayBatchEnabled = false;
+ alarmsPerSecond = 0;
+ assertThatThrownBy(this::checkStaggeredSupported).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void checkStaggeredSupportedThrowsWhenAlarmsEnabled() {
+ gatewayBatchEnabled = true;
+ alarmsPerSecond = 1;
+ assertThatThrownBy(this::checkStaggeredSupported).isInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void checkStaggeredSupportedPassesForTheSupportedCombination() {
+ gatewayBatchEnabled = true;
+ alarmsPerSecond = 0;
+ checkStaggeredSupported(); // must not throw
+ }
+
+ // --- onboardConnectedGateway(): commit-on-success-only (no partial-onboarding leak) ---
+
+ @Test
+ void midOnboardFailureLeavesNoEntryInSharedCollections() {
+ gatewayStartIdx = 100;
+ gatewayEndIdx = 101; // single gateway
+
+ MqttClient client = mock(MqttClient.class);
+ // Simulate a mid-onboard failure: the connect step already "succeeded" (this test starts past
+ // it, with an already-"connected" mock client — see onboardConnectedGateway's javadoc), but the
+ // announce step's publish never confirms.
+ Promise failedAnnounce = ImmediateEventExecutor.INSTANCE.newPromise();
+ failedAnnounce.setFailure(new RuntimeException("simulated announce failure"));
+ when(client.publish(any(), any(), any())).thenReturn(failedAnnounce);
+
+ assertThatThrownBy(() ->
+ onboardConnectedGateway(100, client, "GW00000100", List.of("DW00000000")))
+ .isInstanceOf(Exception.class);
+
+ // The defect this guards: a gateway that fails partway must leave NO trace in any of the shared
+ // collections startRpcBurstSender()/the telemetry scheduler/reconnect recovery read from wholesale
+ // — otherwise a partially-onboarded gateway's un-announced devices would contaminate the RPC
+ // target list.
+ assertThat(mqttClients).isEmpty();
+ assertThat(deviceClients).isEmpty();
+ assertThat(gatewayDeviceNames).isEmpty();
+ assertThat(gatewayTelemetryTimers).isEmpty();
+ verify(client).disconnect();
+ }
+
+ @Test
+ void fullyOnboardedGatewayCommitsAllThreeCollections() throws Exception {
+ gatewayStartIdx = 200;
+ gatewayEndIdx = 201;
+ testMessagesPerSecond = 0; // keep this test focused on the commit, not the telemetry timer
+
+ MqttClient client = mock(MqttClient.class);
+ Promise ackedAnnounce = ImmediateEventExecutor.INSTANCE.newPromise();
+ ackedAnnounce.setSuccess(null);
+ when(client.publish(any(), any(), any())).thenReturn(ackedAnnounce);
+
+ onboardConnectedGateway(200, client, "GW00000200", List.of("DW00000000", "DW00000001"));
+
+ assertThat(mqttClients).containsExactly(client);
+ assertThat(deviceClients).hasSize(2);
+ assertThat(gatewayDeviceNames.get(client)).containsExactly("DW00000000", "DW00000001");
+ verify(client, never()).disconnect();
+ }
+}
diff --git a/src/test/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSenderTest.java b/src/test/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSenderTest.java
index 2b1ee76a..9e0c6c04 100644
--- a/src/test/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSenderTest.java
+++ b/src/test/java/org/thingsboard/tools/service/gateway/rpc/RpcBurstSenderTest.java
@@ -101,7 +101,7 @@ void loadCommandTemplateFallsBackToBuiltInDefault() {
void dispatchSummaryReportsCumulativeBurstsAndDevices() {
RpcBurstSender sender = new RpcBurstSender(
null, null, List.of("d1", "d2"), mapper.createObjectNode(),
- "RpcCalls", 10000, 500, 60, 0, RpcBurstSender.Mode.BURST);
+ "RpcCalls", 10000, 500, 60, 0, RpcBurstSender.Mode.BURST, 48, 512);
sender.recordBurstFired();
sender.recordDispatched(500);
sender.recordBurstFired();
diff --git a/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java b/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java
new file mode 100644
index 00000000..a628372a
--- /dev/null
+++ b/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java
@@ -0,0 +1,142 @@
+/**
+ * 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.tools.service.shared.onboarding;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class StaggeredOnboardingEngineTest {
+
+ /** Stub lifecycle: records each onboard, tracks peak concurrency, can fail chosen indices. */
+ static final class StubLifecycle implements EntityLifecycle {
+ final int count;
+ final int failEvery; // 0 = never fail
+ final AtomicInteger inFlight = new AtomicInteger();
+ final AtomicInteger peakInFlight = new AtomicInteger();
+ final AtomicInteger onboardCalls = new AtomicInteger();
+ final CountDownLatch allTerminal;
+
+ StubLifecycle(int count, int failEvery) {
+ this.count = count;
+ this.failEvery = failEvery;
+ this.allTerminal = new CountDownLatch(count);
+ }
+
+ @Override public int entityCount() { return count; }
+
+ @Override public void onboard(int idx) throws Exception {
+ int now = inFlight.incrementAndGet();
+ peakInFlight.accumulateAndGet(now, Math::max);
+ onboardCalls.incrementAndGet();
+ try {
+ Thread.sleep(5); // simulate work so concurrency is observable
+ if (failEvery > 0 && idx % failEvery == 0) {
+ throw new RuntimeException("stub onboard failure for " + idx);
+ }
+ } finally {
+ inFlight.decrementAndGet();
+ }
+ }
+ }
+
+ @Test
+ void onboardsEveryEntityExactlyOnceAndNeverExceedsConcurrencyCap() throws Exception {
+ StubLifecycle stub = new StubLifecycle(200, 0);
+ AtomicInteger rampOnboarded = new AtomicInteger(-1);
+ AtomicInteger rampFailed = new AtomicInteger(-1);
+ CountDownLatch complete = new CountDownLatch(1);
+
+ StaggeredOnboardingEngine engine =
+ new StaggeredOnboardingEngine(stub, 10, 0, 2, 42L);
+ engine.start((onboarded, failed) -> {
+ rampOnboarded.set(onboarded);
+ rampFailed.set(failed);
+ complete.countDown();
+ });
+
+ assertThat(complete.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(stub.onboardCalls.get()).isEqualTo(200);
+ assertThat(stub.peakInFlight.get()).isLessThanOrEqualTo(10);
+ assertThat(rampOnboarded.get()).isEqualTo(200);
+ assertThat(rampFailed.get()).isEqualTo(0);
+ engine.stop();
+ }
+
+ @Test
+ void countsFailuresAsTerminalSoRampStillCompletes() throws Exception {
+ StubLifecycle stub = new StubLifecycle(50, 10); // idx 0,10,20,30,40 fail -> 5 failures
+ CountDownLatch complete = new CountDownLatch(1);
+ AtomicInteger rampOnboarded = new AtomicInteger();
+ AtomicInteger rampFailed = new AtomicInteger();
+
+ StaggeredOnboardingEngine engine =
+ new StaggeredOnboardingEngine(stub, 8, 0, 2, 1L);
+ engine.start((onboarded, failed) -> {
+ rampOnboarded.set(onboarded);
+ rampFailed.set(failed);
+ complete.countDown();
+ });
+
+ assertThat(complete.await(10, TimeUnit.SECONDS)).isTrue();
+ assertThat(rampOnboarded.get()).isEqualTo(45);
+ assertThat(rampFailed.get()).isEqualTo(5);
+ assertThat(engine.onboardedCount()).isEqualTo(45);
+ assertThat(engine.failedCount()).isEqualTo(5);
+ engine.stop();
+ }
+
+ @Test
+ void firesRampCompleteImmediatelyForZeroEntities() throws Exception {
+ StubLifecycle stub = new StubLifecycle(0, 0);
+ CountDownLatch complete = new CountDownLatch(1);
+ StaggeredOnboardingEngine engine =
+ new StaggeredOnboardingEngine(stub, 4, 0, 1, 0L);
+ engine.start((onboarded, failed) -> complete.countDown());
+ assertThat(complete.await(2, TimeUnit.SECONDS)).isTrue();
+ engine.stop();
+ }
+
+ /**
+ * Guards the ramp-complete-fires-exactly-once contract with a counter, not a {@code
+ * CountDownLatch(1)}: a latch's {@code countDown()} is a silent no-op once it's already at zero, so a
+ * test that only awaits the latch would pass even if the callback fired twice. An {@link
+ * AtomicInteger}, checked after giving a hypothetical duplicate a grace window to land, actually
+ * catches a double-fire.
+ */
+ @Test
+ void rampCompleteFiresExactlyOnce() throws Exception {
+ StubLifecycle stub = new StubLifecycle(50, 0);
+ AtomicInteger completeCalls = new AtomicInteger();
+ CountDownLatch firstComplete = new CountDownLatch(1);
+
+ StaggeredOnboardingEngine engine =
+ new StaggeredOnboardingEngine(stub, 8, 0, 2, 7L);
+ engine.start((onboarded, failed) -> {
+ completeCalls.incrementAndGet();
+ firstComplete.countDown();
+ });
+
+ assertThat(firstComplete.await(10, TimeUnit.SECONDS)).isTrue();
+ Thread.sleep(200); // grace window: let any hypothetical duplicate fire land before asserting
+ assertThat(completeCalls.get()).isEqualTo(1);
+ engine.stop();
+ }
+}