RATIS-1968. Remove unsed reset

ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java

@@ -63,6 +63,7 @@ static class PendingOrderedRequest extends PendingClientRequest
     private final long seqNum;
     private final AtomicReference<Function<SlidingWindowEntry, RaftClientRequest>> requestConstructor;
     private volatile boolean isFirst = false;
+    private volatile long firstSeqNum = 0;
 
     PendingOrderedRequest(long callId, long seqNum,
         Function<SlidingWindowEntry, RaftClientRequest> requestConstructor) {
@@ -83,6 +84,10 @@ public void setFirstRequest() {
       isFirst = true;
     }
 
+    public long getCallId() {
+      return callId;
+    }
+
     @Override
     public long getSeqNum() {
       return seqNum;
@@ -133,7 +138,7 @@ private OrderedAsync(RaftClientImpl client, RaftProperties properties) {
   }
 
   private void resetSlidingWindow(RaftClientRequest request) {
-    getSlidingWindow(request).resetFirstSeqNum();
+    getSlidingWindow(request).resetFirstSeqNum(request.getCallId());
   }
 
   private SlidingWindow.Client<PendingOrderedRequest, RaftClientReply> getSlidingWindow(RaftClientRequest request) {

ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java

@@ -73,6 +73,10 @@ public long getSeqNum() {
       return seqNum;
     }
 
+    public long getCallId() {
+      return -1;
+    }
+
     @Override
     public void setReply(DataStreamReply dataStreamReply) {
       replyFuture.complete(dataStreamReply);

ratis-common/src/main/java/org/apache/ratis/util/SlidingWindow.java

@@ -17,6 +17,7 @@
  */
 package org.apache.ratis.util;
 
+import org.apache.ratis.protocol.RaftClientRequest;
 import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,8 +26,10 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.SortedMap;
 import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -56,6 +59,7 @@ interface Request<REPLY> {
 
   interface ClientSideRequest<REPLY> extends Request<REPLY> {
     void setFirstRequest();
+    long getCallId();
   }
 
   interface ServerSideRequest<REPLY> extends Request<REPLY> {
@@ -228,13 +232,14 @@ class Client<REQUEST extends ClientSideRequest<REPLY>, REPLY> {
     private final RequestMap<REQUEST, REPLY> requests;
     /** Delayed requests. */
     private final DelayedRequests delayedRequests = new DelayedRequests();
+    private final ConcurrentHashMap<Long, Long> map = new ConcurrentHashMap<>();
 
     /** The seqNum for the next new request. */
     private long nextSeqNum = 1;
     /** The seqNum of the first request. */
-    private long firstSeqNum = -1;
+    private volatile long firstSeqNum = -1;
     /** Is the first request replied? */
-    private boolean firstReplied;
+    private volatile boolean firstReplied;
     /** The exception, if there is any. */
     private Throwable exception;
 
@@ -300,6 +305,7 @@ private boolean sendOrDelayRequest(REQUEST request, Consumer<REQUEST> sendMethod
 
       if (firstReplied) {
         // already received the reply for the first request, submit any request.
+        map.put(request.getCallId(), getFirstSeqNum());
         sendMethod.accept(request);
         return true;
       }
@@ -309,6 +315,7 @@ private boolean sendOrDelayRequest(REQUEST request, Consumer<REQUEST> sendMethod
         LOG.debug("{}: detect firstSubmitted {} in {}", requests.getName(), request, this);
         firstSeqNum = seqNum;
         request.setFirstRequest();
+        map.put(request.getCallId(), getFirstSeqNum());
         sendMethod.accept(request);
         return true;
       }
@@ -333,7 +340,9 @@ public synchronized void retry(REQUEST request, Consumer<REQUEST> sendMethod) {
     private void removeRepliedFromHead() {
       for (final Iterator<REQUEST> i = requests.iterator(); i.hasNext(); i.remove()) {
         final REQUEST r = i.next();
-        if (!r.hasReply()) {
+        if (r.hasReply()) {
+          map.remove(r.getCallId());
+        } else {
           return;
         }
       }
@@ -360,24 +369,16 @@ private void trySendDelayed(Consumer<REQUEST> sendMethod) {
         // after first received, all other requests can be submitted (out-of-order)
         delayedRequests.getAllAndClear().forEach(
             seqNum -> sendMethod.accept(requests.getNonRepliedRequest(seqNum, "trySendDelayed")));
-      } else {
-        // Otherwise, submit the first only if it is a delayed request
-        final Iterator<REQUEST> i = requests.iterator();
-        if (i.hasNext()) {
-          final REQUEST r = i.next();
-          final Long delayed = delayedRequests.remove(r.getSeqNum());
-          if (delayed != null) {
-            sendOrDelayRequest(r, sendMethod);
-          }
-        }
       }
     }
 
     /** Reset the {@link #firstSeqNum} The stream has an error. */
-    public synchronized void resetFirstSeqNum() {
-      firstSeqNum = -1;
-      firstReplied = false;
-      LOG.debug("After resetFirstSeqNum: {}", this);
+    public synchronized void resetFirstSeqNum(long callId) {
+      if (callId == -1 || getFirstSeqNum() == map.get(callId)) {
+        firstSeqNum = -1;
+        firstReplied = false;
+        LOG.debug("After resetFirstSeqNum: {}", this);
+      }
     }
 
     /** Fail all requests starting from the given seqNum. */
@@ -409,6 +410,10 @@ private void alreadyClosed(REQUEST request, Throwable e) {
     public synchronized boolean isFirst(long seqNum) {
       return seqNum == (firstSeqNum != -1 ? firstSeqNum : requests.firstSeqNum());
     }
+
+    public long getFirstSeqNum() {
+      return firstSeqNum != -1 ? firstSeqNum : requests.firstSeqNum();
+    }
   }
 
   /**