From a1b12f424158590b84853ae4b290faf65d7bf6bd Mon Sep 17 00:00:00 2001 From: yuhaiming <40624989@qq.com> Date: Fri, 17 Jul 2026 17:11:52 +0800 Subject: [PATCH] fix: gate mqtt retries until startup cleanup --- .../config/MqttCommandAckConfigUnitTest.java | 17 ++-- .../dromara/mqtt/MqttCommandAckService.java | 26 +++++- .../dromara/mqtt/MqttCommandRetryTask.java | 2 +- .../mqtt/MqttCommandAckServiceTest.java | 85 +++++++++++++++++-- .../mqtt/MqttCommandRetryTaskTest.java | 36 ++++++++ 5 files changed, 147 insertions(+), 19 deletions(-) create mode 100644 water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandRetryTaskTest.java diff --git a/water-admin/src/test/java/org/dromara/web/config/MqttCommandAckConfigUnitTest.java b/water-admin/src/test/java/org/dromara/web/config/MqttCommandAckConfigUnitTest.java index a5e1b15..33a54bd 100644 --- a/water-admin/src/test/java/org/dromara/web/config/MqttCommandAckConfigUnitTest.java +++ b/water-admin/src/test/java/org/dromara/web/config/MqttCommandAckConfigUnitTest.java @@ -14,8 +14,8 @@ import static org.assertj.core.api.Assertions.assertThat; class MqttCommandAckConfigUnitTest { @Test - void commandAckMaxRetryCountIsThree() throws Exception { - Integer maxRetryCount = null; + void commandAckRetrySettingsMatchRuntimeRequirements() throws Exception { + Map commandAckConfig = null; Yaml yaml = new Yaml(); try (InputStream inputStream = new ClassPathResource("application.yml").getInputStream()) { for (Object document : yaml.loadAll(inputStream)) { @@ -27,17 +27,16 @@ class MqttCommandAckConfigUnitTest { continue; } Object commandAck = mqttConfig.get("command-ack"); - if (!(commandAck instanceof Map commandAckConfig)) { - continue; - } - Object value = commandAckConfig.get("max-retry-count"); - if (value instanceof Number number) { - maxRetryCount = number.intValue(); + if (commandAck instanceof Map config) { + commandAckConfig = config; } } } - assertThat(maxRetryCount).isEqualTo(3); + assertThat(commandAckConfig).isNotNull(); + assertThat(commandAckConfig.get("retry-interval-ms")).isEqualTo(10000); + assertThat(commandAckConfig.get("scan-interval-ms")).isEqualTo(5000); + assertThat(commandAckConfig.get("max-retry-count")).isEqualTo(3); } @Test diff --git a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java index c3cb3c2..644b76d 100644 --- a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java +++ b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandAckService.java @@ -37,6 +37,7 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler { private MqttClientManager mqttClientManager; private final MqttProperties mqttProperties; private final AppDeviceMapper appDeviceMapper; + private volatile boolean startupCleanupReady; private static final String DEVICE_STATUS_LOCK_PREFIX = "lock:mqtt:device:status:"; static boolean ackMatchesPendingCommand(String topicDeviceNo, DeviceCommand pendingCommand) { @@ -133,14 +134,33 @@ public class MqttCommandAckService implements IDeviceCommandAckHandler { } public int clearPendingCommands() { - Set commandIds = pendingIds().readAll(); + RSet pendingIds = pendingIds(); + Set commandIds = pendingIds.readAll(); + pendingIds.removeAll(commandIds); + startupCleanupReady = true; + + RuntimeException cleanupFailure = null; for (String commandId : commandIds) { - RedisUtils.deleteObject(pendingKey(commandId)); + try { + RedisUtils.deleteObject(pendingKey(commandId)); + } catch (RuntimeException e) { + if (cleanupFailure == null) { + cleanupFailure = new RuntimeException("Failed to delete MQTT pending command object: " + commandId, e); + } else { + cleanupFailure.addSuppressed(e); + } + } + } + if (cleanupFailure != null) { + throw cleanupFailure; } - pendingIds().clear(); return commandIds.size(); } + boolean isStartupCleanupReady() { + return startupCleanupReady; + } + public void removePending(String commandId) { if (StringUtils.isBlank(commandId)) { return; diff --git a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandRetryTask.java b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandRetryTask.java index e2a26ac..1c5bd86 100644 --- a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandRetryTask.java +++ b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttCommandRetryTask.java @@ -16,7 +16,7 @@ public class MqttCommandRetryTask { @Scheduled(fixedDelayString = "${mqtt.command-ack.scan-interval-ms:5000}") public void retryExpiredCommands() { - if (mqttProperties.getCommandAck().isEnabled()) { + if (ackService.isStartupCleanupReady() && mqttProperties.getCommandAck().isEnabled()) { // log.info("[设备命令重新下发] 定时器调用成功 "); ackService.retryExpiredCommands(); } diff --git a/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java index 561bd51..8326610 100644 --- a/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java +++ b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandAckServiceTest.java @@ -23,13 +23,16 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.test.util.ReflectionTestUtils; import java.time.Duration; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; 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.eq; import static org.mockito.Mockito.*; @@ -99,35 +102,48 @@ class MqttCommandAckServiceTest { } @Test - void clearPendingCommandsDeletesPendingObjectsAndClearsIds() { + void clearPendingCommandsRemovesSnapshotBeforeDeletingPendingObjects() { MqttProperties properties = new MqttProperties(); AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class); RedissonClient redissonClient = mock(RedissonClient.class); RSet pendingIds = mock(RSet.class); + Set snapshot = new LinkedHashSet<>(List.of("cmd-1", "cmd-2")); + AtomicBoolean snapshotRemoved = new AtomicBoolean(); when(redissonClient.getSet("mqtt:command:pending:ids")).thenReturn(pendingIds); - when(pendingIds.readAll()).thenReturn(Set.of("cmd-1", "cmd-2")); + when(pendingIds.readAll()).thenReturn(snapshot); + doAnswer(invocation -> { + snapshotRemoved.set(true); + return true; + }).when(pendingIds).removeAll(snapshot); MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); try (MockedStatic redis = mockStatic(RedisUtils.class)) { redis.when(RedisUtils::getClient).thenReturn(redissonClient); + redis.when(() -> RedisUtils.deleteObject(any(String.class))).thenAnswer(invocation -> { + assertThat(snapshotRemoved).isTrue(); + return true; + }); int cleared = service.clearPendingCommands(); assertThat(cleared).isEqualTo(2); + assertThat(service.isStartupCleanupReady()).isTrue(); redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1")); redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-2")); - verify(pendingIds).clear(); + verify(pendingIds).removeAll(snapshot); + verify(pendingIds, never()).clear(); } } @Test - void clearPendingCommandsReturnsZeroForEmptySet() { + void clearPendingCommandsOpensRetryGateForEmptySet() { MqttProperties properties = new MqttProperties(); AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class); RedissonClient redissonClient = mock(RedissonClient.class); RSet pendingIds = mock(RSet.class); + Set snapshot = Set.of(); when(redissonClient.getSet("mqtt:command:pending:ids")).thenReturn(pendingIds); - when(pendingIds.readAll()).thenReturn(Set.of()); + when(pendingIds.readAll()).thenReturn(snapshot); MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); try (MockedStatic redis = mockStatic(RedisUtils.class)) { @@ -136,7 +152,64 @@ class MqttCommandAckServiceTest { int cleared = service.clearPendingCommands(); assertThat(cleared).isZero(); - verify(pendingIds).clear(); + assertThat(service.isStartupCleanupReady()).isTrue(); + verify(pendingIds).removeAll(snapshot); + verify(pendingIds, never()).clear(); + } + } + + @Test + void clearPendingCommandsKeepsRetryGateClosedWhenSnapshotRemovalFails() { + MqttProperties properties = new MqttProperties(); + AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class); + RedissonClient redissonClient = mock(RedissonClient.class); + RSet pendingIds = mock(RSet.class); + Set snapshot = Set.of("cmd-1"); + when(redissonClient.getSet("mqtt:command:pending:ids")).thenReturn(pendingIds); + when(pendingIds.readAll()).thenReturn(snapshot); + doThrow(new IllegalStateException("redis unavailable")).when(pendingIds).removeAll(snapshot); + + MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(RedisUtils::getClient).thenReturn(redissonClient); + + assertThat(service.isStartupCleanupReady()).isFalse(); + assertThatThrownBy(service::clearPendingCommands) + .isInstanceOf(IllegalStateException.class) + .hasMessage("redis unavailable"); + + assertThat(service.isStartupCleanupReady()).isFalse(); + redis.verify(() -> RedisUtils.deleteObject(any(String.class)), never()); + verify(pendingIds, never()).clear(); + } + } + + @Test + void clearPendingCommandsContinuesDeletingObjectsAndReportsFailures() { + MqttProperties properties = new MqttProperties(); + AppDeviceMapper appDeviceMapper = mock(AppDeviceMapper.class); + RedissonClient redissonClient = mock(RedissonClient.class); + RSet pendingIds = mock(RSet.class); + Set snapshot = new LinkedHashSet<>(List.of("cmd-1", "cmd-2")); + when(redissonClient.getSet("mqtt:command:pending:ids")).thenReturn(pendingIds); + when(pendingIds.readAll()).thenReturn(snapshot); + + MqttCommandAckService service = new MqttCommandAckService(properties, appDeviceMapper); + try (MockedStatic redis = mockStatic(RedisUtils.class)) { + redis.when(RedisUtils::getClient).thenReturn(redissonClient); + redis.when(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1")) + .thenThrow(new IllegalStateException("delete failed")); + + assertThatThrownBy(service::clearPendingCommands) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("cmd-1") + .hasCauseInstanceOf(IllegalStateException.class); + + verify(pendingIds).removeAll(snapshot); + verify(pendingIds, never()).clear(); + redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-1")); + redis.verify(() -> RedisUtils.deleteObject("mqtt:command:pending:cmd-2")); + assertThat(service.isStartupCleanupReady()).isTrue(); } } diff --git a/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandRetryTaskTest.java b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandRetryTaskTest.java new file mode 100644 index 0000000..2188c76 --- /dev/null +++ b/water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttCommandRetryTaskTest.java @@ -0,0 +1,36 @@ +package org.dromara.mqtt; + +import org.dromara.mqtt.config.properties.MqttProperties; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@Tag("dev") +class MqttCommandRetryTaskTest { + + @Test + void retryExpiredCommandsSkipsRetryBeforeStartupCleanupIsReady() { + MqttCommandAckService ackService = mock(MqttCommandAckService.class); + when(ackService.isStartupCleanupReady()).thenReturn(false); + MqttCommandRetryTask task = new MqttCommandRetryTask(ackService, new MqttProperties()); + + task.retryExpiredCommands(); + + verify(ackService, never()).retryExpiredCommands(); + } + + @Test + void retryExpiredCommandsRetriesAfterStartupCleanupIsReady() { + MqttCommandAckService ackService = mock(MqttCommandAckService.class); + when(ackService.isStartupCleanupReady()).thenReturn(true); + MqttCommandRetryTask task = new MqttCommandRetryTask(ackService, new MqttProperties()); + + task.retryExpiredCommands(); + + verify(ackService).retryExpiredCommands(); + } +}