Fix abnormal transmission of session inbox version.

Author: ZHANGSHUNLIN

bifromq-inbox/bifromq-inbox-server/src/main/java/org/apache/bifromq/inbox/server/scheduler/BatchDetachCall.java

@@ -21,6 +21,12 @@
 
 import static java.util.Collections.emptySet;
 
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.bifromq.basekv.client.IMutationPipeline;
 import org.apache.bifromq.basekv.client.exception.BadVersionException;
 import org.apache.bifromq.basekv.client.exception.TryLaterException;
@@ -37,12 +43,6 @@
 import org.apache.bifromq.inbox.storage.proto.InboxServiceRWCoProcInput;
 import org.apache.bifromq.inbox.storage.proto.InboxVersion;
 import org.apache.bifromq.inbox.storage.proto.Replica;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 class BatchDetachCall extends BatchMutationCall<DetachRequest, DetachReply> {
@@ -69,10 +69,12 @@ protected RWCoProcInput makeBatch(
             BatchDetachRequest.Params.Builder paramsBuilder = BatchDetachRequest.Params.newBuilder()
                 .setTenantId(request.getClient().getTenantId())
                 .setInboxId(request.getInboxId())
-                .setVersion(request.getVersion())
                 .setExpirySeconds(request.getExpirySeconds())
                 .setDiscardLWT(request.getDiscardLWT())
                 .setNow(request.getNow());
+            if (request.hasVersion()) {
+                paramsBuilder.setVersion(request.getVersion());
+            }
             reqBuilder.addParams(paramsBuilder.build());
         }
 

bifromq-mqtt/bifromq-mqtt-server/src/main/java/org/apache/bifromq/mqtt/handler/MQTTConnectHandler.java

@@ -387,6 +387,7 @@ private CompletableFuture<ExpireResult> expireInbox(long reqId,
         if (requestClientId.isEmpty()) {
             return CompletableFuture.completedFuture(ExpireResult.NOT_FOUND);
         }
+        // detach and expire the latest version immediately
         return inboxClient.detach(DetachRequest.newBuilder()
                 .setReqId(reqId)
                 .setInboxId(userSessionId)

bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/integration/v5/MQTTConnectTest.java

@@ -24,6 +24,8 @@
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
 
 import org.apache.bifromq.mqtt.integration.MQTTTest;
 import org.apache.bifromq.mqtt.integration.v5.client.MqttTestClient;
@@ -134,4 +136,39 @@ public void forceTransient() {
         IMqttToken token = client.connect(connOpts);
         assertEquals(token.getResponseProperties().getSessionExpiryInterval(), 0L);
     }
+
+    /**
+     *  Test whether it can reconnect when cleanStart=true but the session expiration interval is not 0, when reconnecting.
+     */
+    @Test(groups = "integration")
+    public void reconnectTest() {
+        when(authProvider.auth(any(MQTT5AuthData.class)))
+            .thenReturn(CompletableFuture.completedFuture(MQTT5AuthResult.newBuilder()
+                .setSuccess(Success.newBuilder()
+                    .setTenantId("tenant")
+                    .setUserId("testUser")
+                    .build()).build()));
+        when(authProvider.checkPermission(any(), any()))
+            .thenReturn(CompletableFuture.completedFuture(
+                CheckResult.newBuilder().setGranted(Granted.newBuilder().build()).build()));
+        when(settingProvider.provide(eq(Setting.ForceTransient), eq("tenant"))).thenReturn(false);
+
+        MqttConnectionOptions connOpts = new MqttConnectionOptions();
+        connOpts.setCleanStart(true);
+        connOpts.setSessionExpiryInterval(1800L);
+        connOpts.setUserName("tenant/testUser");
+
+        MqttTestClient client1 = new MqttTestClient(BROKER_URI, "client_id");
+        IMqttToken token1 = client1.connect(connOpts);
+        assertTrue(token1.isComplete());
+        assertNull(token1.getException());
+        client1.disconnect();
+
+        // reconnect
+        MqttTestClient client2 = new MqttTestClient(BROKER_URI, "client_id");
+        IMqttToken token2 = client2.connect(connOpts);
+        assertTrue(token2.isComplete());
+        assertNull(token2.getException());
+    }
+
 }