fix sofastack#754

googlespot · Jan 25, 2022 · 67e3335 · 67e3335
1 parent 4f02797
commit 67e3335
Showing 1 changed file with 18 additions and 6 deletions.
diff --git a/jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java b/jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java
@@ -1936,12 +1936,24 @@ public Message handleAppendEntriesRequest(final AppendEntriesRequest request, fi
 
             updateLastLeaderTimestamp(Utils.monotonicMs());
 
-            if (entriesCount > 0 && this.snapshotExecutor != null && this.snapshotExecutor.isInstallingSnapshot()) {
-                LOG.warn("Node {} received AppendEntriesRequest while installing snapshot.", getNodeId());
-                return RpcFactoryHelper //
-                    .responseFactory() //
-                    .newResponse(AppendEntriesResponse.getDefaultInstance(), RaftError.EBUSY,
-                        "Node %s:%s is installing snapshot.", this.groupId, this.serverId);
+            if (entriesCount > 0) {
+                if (this.snapshotExecutor != null && this.snapshotExecutor.isInstallingSnapshot()) {
+                    LOG.warn("Node {} received AppendEntriesRequest while installing snapshot.", getNodeId());
+                    return RpcFactoryHelper //
+                        .responseFactory() //
+                        .newResponse(AppendEntriesResponse.getDefaultInstance(), RaftError.EBUSY,
+                            "Node %s:%s is installing snapshot.", this.groupId, this.serverId);
+                }
+                // Make sure applyQueue has enough space to hold logs that have not yet been applied
+                long remaining = this.applyQueue.getBufferSize()
+                                 - (this.logManager.getLastLogIndex() - this.fsmCaller.getLastAppliedIndex());
+                if (remaining < entriesCount) {
+                    LOG.warn("Node {} received AppendEntriesRequest but applyQueue not enough", getNodeId());
+                    return RpcFactoryHelper //
+                        .responseFactory() //
+                        .newResponse(AppendEntriesResponse.getDefaultInstance(), RaftError.EBUSY,
+                            "Node %s:%s is busy.", this.groupId, this.serverId);
+                }
             }
 
             final long prevLogIndex = request.getPrevLogIndex();