fix: clear mqtt pending commands after startup

water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/MqttPendingCommandCleanupListener.java

@@ -0,0 +1,25 @@
+package org.dromara.mqtt;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class MqttPendingCommandCleanupListener {
+
+    private final MqttCommandAckService mqttCommandAckService;
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void onApplicationReady(ApplicationReadyEvent event) {
+        try {
+            int clearedCount = mqttCommandAckService.clearPendingCommands();
+            log.info("[MQTT] 应用启动清理待确认命令完成，清理数量: {}", clearedCount);
+        } catch (RuntimeException e) {
+            log.error("[MQTT] 应用启动清理待确认命令失败", e);
+        }
+    }
+}

water-common/water-common-mqtt/src/test/java/org/dromara/mqtt/MqttPendingCommandCleanupListenerTest.java

@@ -0,0 +1,35 @@
+package org.dromara.mqtt;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Tag;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@Tag("dev")
+class MqttPendingCommandCleanupListenerTest {
+
+    @Test
+    void onApplicationReadyClearsPendingCommands() {
+        MqttCommandAckService service = mock(MqttCommandAckService.class);
+        when(service.clearPendingCommands()).thenReturn(3);
+        MqttPendingCommandCleanupListener listener = new MqttPendingCommandCleanupListener(service);
+
+        listener.onApplicationReady(mock(ApplicationReadyEvent.class));
+
+        verify(service).clearPendingCommands();
+    }
+
+    @Test
+    void onApplicationReadyDoesNotThrowWhenCleanupFails() {
+        MqttCommandAckService service = mock(MqttCommandAckService.class);
+        when(service.clearPendingCommands()).thenThrow(new IllegalStateException("redis unavailable"));
+        MqttPendingCommandCleanupListener listener = new MqttPendingCommandCleanupListener(service);
+
+        assertThatCode(() -> listener.onApplicationReady(mock(ApplicationReadyEvent.class)))
+            .doesNotThrowAnyException();
+    }
+}