Refactoring the internal keep alive mechanism for persistent session, completely eliminated the periodic touch operation

Author: popduke

bifromq-inbox/bifromq-inbox-client/src/main/java/com/baidu/bifromq/inbox/client/IInboxClient.java

@@ -34,8 +34,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.SendLWTRequest;
 import com.baidu.bifromq.inbox.rpc.proto.SubReply;
 import com.baidu.bifromq.inbox.rpc.proto.SubRequest;
-import com.baidu.bifromq.inbox.rpc.proto.TouchReply;
-import com.baidu.bifromq.inbox.rpc.proto.TouchRequest;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubReply;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubRequest;
 import com.baidu.bifromq.inbox.storage.proto.Fetched;
@@ -62,8 +60,6 @@ default int id() {
 
     CompletableFuture<DetachReply> detach(DetachRequest request);
 
-    CompletableFuture<TouchReply> touch(TouchRequest request);
-
     CompletableFuture<SubReply> sub(SubRequest request);
 
     CompletableFuture<UnsubReply> unsub(UnsubRequest request);

bifromq-inbox/bifromq-inbox-client/src/main/java/com/baidu/bifromq/inbox/client/InboxClient.java

@@ -37,8 +37,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.SendLWTRequest;
 import com.baidu.bifromq.inbox.rpc.proto.SubReply;
 import com.baidu.bifromq.inbox.rpc.proto.SubRequest;
-import com.baidu.bifromq.inbox.rpc.proto.TouchReply;
-import com.baidu.bifromq.inbox.rpc.proto.TouchRequest;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubReply;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubRequest;
 import com.baidu.bifromq.plugin.subbroker.CheckReply;
@@ -155,17 +153,6 @@ public CompletableFuture<DetachReply> detach(DetachRequest request) {
             });
     }
 
-    @Override
-    public CompletableFuture<TouchReply> touch(TouchRequest request) {
-        return rpcClient.invoke(request.getTenantId(), null, request, InboxServiceGrpc.getTouchMethod())
-            .exceptionally(e -> {
-                log.debug("Touch inbox failed", e);
-                return TouchReply.newBuilder()
-                    .setReqId(request.getReqId())
-                    .setCode(TouchReply.Code.ERROR).build();
-            });
-    }
-
     @Override
     public CompletableFuture<SubReply> sub(SubRequest request) {
         return rpcClient.invoke(request.getTenantId(), null, request, InboxServiceGrpc.getSubMethod())

bifromq-inbox/bifromq-inbox-client/src/test/java/com.baidu.bifromq.inbox.client/InboxClientTest.java

@@ -38,8 +38,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.GetRequest;
 import com.baidu.bifromq.inbox.rpc.proto.SubReply;
 import com.baidu.bifromq.inbox.rpc.proto.SubRequest;
-import com.baidu.bifromq.inbox.rpc.proto.TouchReply;
-import com.baidu.bifromq.inbox.rpc.proto.TouchRequest;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubReply;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubRequest;
 import com.baidu.bifromq.plugin.subbroker.CheckReply;
@@ -180,17 +178,6 @@ public void unsubRPCException() {
         assertEquals(reply.getCode(), UnsubReply.Code.ERROR);
     }
 
-    @Test
-    public void touchRPCException() {
-        TouchRequest touchRequest = TouchRequest.newBuilder().setTenantId("TenantId").build();
-
-        when(rpcClient.invoke(anyString(), isNull(), any(), any()))
-            .thenReturn(CompletableFuture.failedFuture(new RuntimeException("Mocked")));
-        TouchReply reply = inboxClient.touch(touchRequest).join();
-        verify(rpcClient).invoke(eq(touchRequest.getTenantId()), isNull(), eq(touchRequest), any());
-        assertEquals(reply.getCode(), TouchReply.Code.ERROR);
-    }
-
     @Test
     public void expireRPCException() {
         ExpireRequest expireRequest = ExpireRequest.newBuilder().setTenantId("TenantId").build();

bifromq-inbox/bifromq-inbox-coproc-proto/src/main/proto/inboxservice/InboxStoreCoProc.proto

@@ -68,7 +68,6 @@ message BatchCreateRequest{
     uint64 now = 9;
   }
   repeated Params params = 1;
-  Replica leader = 2; // the leader replica
 }
 
 message BatchCreateReply{
@@ -108,7 +107,6 @@ message BatchDetachRequest{
     uint32 expirySeconds = 5;
     bool discardLWT = 6;
     uint64 now = 7;
-    optional Replica sender = 8; // the replica on which the detach task is running to trigger the detach request
   }
   repeated Params params = 1;
   Replica leader = 2; // the leader replica
@@ -123,26 +121,6 @@ message BatchDetachReply{
   repeated Code code = 1;
 }
 
-message BatchTouchRequest{
-  message Params{
-    string tenantId = 1;
-    string inboxId = 2;
-    uint64 incarnation = 3;
-    uint64 version = 4;
-    uint64 now = 5;
-  }
-  repeated Params params = 1;
-}
-
-message BatchTouchReply{
-  enum Code{
-    OK = 0;
-    NO_INBOX = 1;
-    CONFLICT = 2;
-  }
-  repeated Code code = 1;
-}
-
 message BatchDeleteRequest{
   message Params{
     string tenantId = 1;
@@ -177,7 +155,6 @@ message BatchSubRequest{
     uint64 now = 7;
   }
   repeated Params params = 1;
-  Replica leader = 2; // the leader replica
 }
 
 message BatchSubReply{
@@ -201,7 +178,6 @@ message BatchUnsubRequest{
     uint64 now = 6;
   }
   repeated Params params = 1;
-  Replica leader = 2; // the leader replica
 }
 
 message BatchUnsubReply{
@@ -278,7 +254,6 @@ message BatchCommitRequest{
     uint64 now = 8;
   }
   repeated Params params = 1;
-  Replica leader = 2; // the leader replica
 }
 
 message BatchCommitReply{
@@ -374,10 +349,9 @@ message InboxServiceROCoProcInput{
     BatchGetRequest batchGet = 2;
     BatchFetchRequest batchFetch = 3;
     BatchCheckSubRequest batchCheckSub = 4;
-    BatchTouchRequest batchTouch = 5;
-    BatchSendLWTRequest batchSendLWT = 6;
-    ExpireTenantRequest expireTenant = 7;
-    GCRequest gc = 8;
+    BatchSendLWTRequest batchSendLWT = 5;
+    ExpireTenantRequest expireTenant = 6;
+    GCRequest gc = 7;
   }
 }
 
@@ -387,9 +361,8 @@ message InboxServiceROCoProcOutput{
     BatchGetReply batchGet = 2;
     BatchFetchReply batchFetch = 3;
     BatchCheckSubReply batchCheckSub = 4;
-    BatchTouchReply batchTouch = 5;
-    BatchSendLWTReply batchSendLWT = 6;
-    ExpireTenantReply expireTenant = 7;
-    GCReply gc = 8;
+    BatchSendLWTReply batchSendLWT = 5;
+    ExpireTenantReply expireTenant = 6;
+    GCReply gc = 7;
   }
 }

bifromq-inbox/bifromq-inbox-rpc-definition/src/main/java/com/baidu/bifromq/inbox/RPCBluePrint.java

@@ -26,7 +26,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.InboxServiceGrpc;
 import com.baidu.bifromq.inbox.rpc.proto.SendLWTRequest;
 import com.baidu.bifromq.inbox.rpc.proto.SubRequest;
-import com.baidu.bifromq.inbox.rpc.proto.TouchRequest;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubRequest;
 import com.baidu.bifromq.plugin.subbroker.CheckRequest;
 
@@ -51,8 +50,6 @@ public class RPCBluePrint {
             .keyHashFunc(request -> getDelivererKey(request.getTenantId(), request.getInboxId())).build())
         .methodSemantic(InboxServiceGrpc.getCreateMethod(), BluePrint.WCHUnaryMethod.<CreateRequest>builder()
             .keyHashFunc(request -> getDelivererKey(request.getClient().getTenantId(), request.getInboxId())).build())
-        .methodSemantic(InboxServiceGrpc.getTouchMethod(), BluePrint.WCHUnaryMethod.<TouchRequest>builder()
-            .keyHashFunc(request -> getDelivererKey(request.getTenantId(), request.getInboxId())).build())
         .methodSemantic(InboxServiceGrpc.getSubMethod(), BluePrint.WCHUnaryMethod.<SubRequest>builder()
             .keyHashFunc(request -> getDelivererKey(request.getTenantId(), request.getInboxId())).build())
         .methodSemantic(InboxServiceGrpc.getUnsubMethod(), BluePrint.WCHUnaryMethod.<UnsubRequest>builder()

bifromq-inbox/bifromq-inbox-rpc-definition/src/main/proto/inboxservice/InboxService.proto

@@ -18,7 +18,6 @@ service InboxService {
   rpc attach (AttachRequest) returns (AttachReply);
   rpc detach (DetachRequest) returns (DetachReply);
   rpc create (CreateRequest) returns (CreateReply);
-  rpc touch (TouchRequest) returns (TouchReply);
   rpc sub(SubRequest) returns (SubReply);
   rpc unsub(UnsubRequest) returns (UnsubReply);
   rpc receive (stream SendRequest) returns (stream SendReply);
@@ -85,7 +84,6 @@ message DetachRequest {
   bool discardLWT = 6;
   commontype.ClientInfo client = 7;
   uint64 now = 8;
-  optional Replica sender = 9; // if the request is triggered by detach task on the replica
 }
 
 message DetachReply {
@@ -125,27 +123,6 @@ message CreateReply {
   Code code = 2;
 }
 
-message TouchRequest {
-  uint64 reqId = 1;
-  string tenantId = 2;
-  string inboxId = 3;
-  uint64 incarnation = 4;
-  uint64 version = 5;
-  uint64 now = 6;
-}
-
-message TouchReply {
-  enum Code {
-    OK = 0;
-    NO_INBOX = 1;
-    CONFLICT = 2;
-    TRY_LATER = 3;
-    ERROR = 4;
-  }
-  uint64 reqId = 1;
-  Code code = 2;
-}
-
 message SubRequest {
   uint64 reqId = 1;
   string tenantId = 2;

bifromq-inbox/bifromq-inbox-server/src/main/java/com/baidu/bifromq/inbox/server/InboxServer.java

@@ -26,7 +26,6 @@
 import com.baidu.bifromq.inbox.server.scheduler.InboxInsertScheduler;
 import com.baidu.bifromq.inbox.server.scheduler.InboxSendLWTScheduler;
 import com.baidu.bifromq.inbox.server.scheduler.InboxSubScheduler;
-import com.baidu.bifromq.inbox.server.scheduler.InboxTouchScheduler;
 import com.baidu.bifromq.inbox.server.scheduler.InboxUnSubScheduler;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.micrometer.core.instrument.Metrics;
@@ -59,7 +58,6 @@ class InboxServer implements IInboxServer {
             .deleteScheduler(new InboxDeleteScheduler(builder.inboxStoreClient))
             .subScheduler(new InboxSubScheduler(builder.inboxStoreClient))
             .unsubScheduler(new InboxUnSubScheduler(builder.inboxStoreClient))
-            .touchScheduler(new InboxTouchScheduler(builder.inboxStoreClient))
             .tenantGCRunner(new TenantGCRunner(builder.inboxStoreClient))
             .build();
         if (builder.workerThreads == 0) {

bifromq-inbox/bifromq-inbox-server/src/main/java/com/baidu/bifromq/inbox/server/InboxService.java

@@ -47,8 +47,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.SendRequest;
 import com.baidu.bifromq.inbox.rpc.proto.SubReply;
 import com.baidu.bifromq.inbox.rpc.proto.SubRequest;
-import com.baidu.bifromq.inbox.rpc.proto.TouchReply;
-import com.baidu.bifromq.inbox.rpc.proto.TouchRequest;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubReply;
 import com.baidu.bifromq.inbox.rpc.proto.UnsubRequest;
 import com.baidu.bifromq.inbox.server.scheduler.IInboxAttachScheduler;
@@ -62,7 +60,6 @@
 import com.baidu.bifromq.inbox.server.scheduler.IInboxInsertScheduler;
 import com.baidu.bifromq.inbox.server.scheduler.IInboxSendLWTScheduler;
 import com.baidu.bifromq.inbox.server.scheduler.IInboxSubScheduler;
-import com.baidu.bifromq.inbox.server.scheduler.IInboxTouchScheduler;
 import com.baidu.bifromq.inbox.server.scheduler.IInboxUnsubScheduler;
 import com.baidu.bifromq.inbox.storage.proto.TopicFilterOption;
 import com.baidu.bifromq.plugin.subbroker.CheckReply;
@@ -87,7 +84,6 @@ class InboxService extends InboxServiceGrpc.InboxServiceImplBase {
     private final IInboxCheckSubScheduler checkSubScheduler;
     private final IInboxInsertScheduler insertScheduler;
     private final IInboxCommitScheduler commitScheduler;
-    private final IInboxTouchScheduler touchScheduler;
     private final IInboxCreateScheduler createScheduler;
     private final IInboxAttachScheduler attachScheduler;
     private final IInboxDetachScheduler detachScheduler;
@@ -111,7 +107,6 @@ class InboxService extends InboxServiceGrpc.InboxServiceImplBase {
                  IInboxDeleteScheduler deleteScheduler,
                  IInboxSubScheduler subScheduler,
                  IInboxUnsubScheduler unsubScheduler,
-                 IInboxTouchScheduler touchScheduler,
                  ITenantGCRunner tenantGCRunner) {
         this.inboxClient = inboxClient;
         this.distClient = distClient;
@@ -127,7 +122,6 @@ class InboxService extends InboxServiceGrpc.InboxServiceImplBase {
         this.deleteScheduler = deleteScheduler;
         this.subScheduler = subScheduler;
         this.unsubScheduler = unsubScheduler;
-        this.touchScheduler = touchScheduler;
         this.tenantGCRunner = tenantGCRunner;
     }
 
@@ -225,26 +219,6 @@ public void detach(DetachRequest request, StreamObserver<DetachReply> responseOb
             }), responseObserver);
     }
 
-    @Override
-    public void touch(TouchRequest request, StreamObserver<TouchReply> responseObserver) {
-        log.trace("Handling touch {}", request);
-        response(tenantId -> touchScheduler.schedule(request)
-            .exceptionally(e -> {
-                if (e instanceof BatcherUnavailableException || e.getCause() instanceof BatcherUnavailableException) {
-                    return TouchReply.newBuilder()
-                        .setReqId(request.getReqId())
-                        .setCode(TouchReply.Code.TRY_LATER)
-                        .build();
-                } else {
-                    log.debug("Failed to touch inbox", e);
-                    return TouchReply.newBuilder()
-                        .setReqId(request.getReqId())
-                        .setCode(TouchReply.Code.ERROR)
-                        .build();
-                }
-            }), responseObserver);
-    }
-
     @Override
     public void sub(SubRequest request, StreamObserver<SubReply> responseObserver) {
         log.trace("Handling sub {}", request);
@@ -612,7 +586,6 @@ public void stop() {
             detachScheduler.close();
             deleteScheduler.close();
             createScheduler.close();
-            touchScheduler.close();
             subScheduler.close();
             unsubScheduler.close();
 

bifromq-inbox/bifromq-inbox-server/src/main/java/com/baidu/bifromq/inbox/server/scheduler/BatchCommitCall.java

@@ -28,7 +28,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.CommitRequest;
 import com.baidu.bifromq.inbox.storage.proto.BatchCommitRequest;
 import com.baidu.bifromq.inbox.storage.proto.InboxServiceRWCoProcInput;
-import com.baidu.bifromq.inbox.storage.proto.Replica;
 import java.util.HashSet;
 import java.util.Queue;
 import java.util.Set;
@@ -46,11 +45,7 @@ protected MutationCallTaskBatch<CommitRequest, CommitReply> newBatch(long ver) {
     @Override
     protected RWCoProcInput makeBatch(
         Iterable<ICallTask<CommitRequest, CommitReply, MutationCallBatcherKey>> callTasks) {
-        BatchCommitRequest.Builder reqBuilder = BatchCommitRequest.newBuilder()
-            .setLeader(Replica.newBuilder()
-                .setRangeId(batcherKey.id)
-                .setStoreId(batcherKey.leaderStoreId)
-                .build());
+        BatchCommitRequest.Builder reqBuilder = BatchCommitRequest.newBuilder();
         callTasks.forEach(call -> {
             CommitRequest req = call.call();
             BatchCommitRequest.Params.Builder paramsBuilder = BatchCommitRequest.Params.newBuilder()

bifromq-inbox/bifromq-inbox-server/src/main/java/com/baidu/bifromq/inbox/server/scheduler/BatchCreateCall.java

@@ -28,7 +28,6 @@
 import com.baidu.bifromq.inbox.rpc.proto.CreateRequest;
 import com.baidu.bifromq.inbox.storage.proto.BatchCreateRequest;
 import com.baidu.bifromq.inbox.storage.proto.InboxServiceRWCoProcInput;
-import com.baidu.bifromq.inbox.storage.proto.Replica;
 import com.baidu.bifromq.type.ClientInfo;
 import java.util.HashSet;
 import java.util.Queue;
@@ -48,11 +47,7 @@ protected MutationCallTaskBatch<CreateRequest, CreateReply> newBatch(long ver) {
     @Override
     protected RWCoProcInput makeBatch(
         Iterable<ICallTask<CreateRequest, CreateReply, MutationCallBatcherKey>> callTasks) {
-        BatchCreateRequest.Builder reqBuilder = BatchCreateRequest.newBuilder()
-            .setLeader(Replica.newBuilder()
-                .setRangeId(batcherKey.id)
-                .setStoreId(batcherKey.leaderStoreId)
-                .build());
+        BatchCreateRequest.Builder reqBuilder = BatchCreateRequest.newBuilder();
         callTasks.forEach(call -> {
             CreateRequest request = call.call();
             ClientInfo client = request.getClient();