RATIS-2234. Remove lock race between heartbeat and append log channels (#1205)

Author: SzyWilliam

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

@@ -17,6 +17,7 @@
  */
 package org.apache.ratis.util;
 
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 
@@ -45,6 +46,13 @@ public static AutoCloseableLock acquire(final Lock lock, Runnable preUnlock) {
     return new AutoCloseableLock(lock, preUnlock);
   }
 
+  public static AutoCloseableLock tryAcquire(final Lock lock, Runnable preUnlock, TimeDuration timeout)
+      throws InterruptedException {
+    Objects.requireNonNull(timeout, "timeout == null");
+    final boolean locked = lock.tryLock(timeout.getDuration(), timeout.getUnit());
+    return locked? new AutoCloseableLock(lock, preUnlock): null;
+  }
+
   private final Lock underlying;
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final Runnable preUnlock;

ratis-server/src/main/java/org/apache/ratis/server/raftlog/RaftLogBase.java

@@ -119,7 +119,7 @@ public boolean isOpened() {
 
   @Override
   public boolean updateCommitIndex(long majorityIndex, long currentTerm, boolean isLeader) {
-    try(AutoCloseableLock writeLock = writeLock()) {
+    try(AutoCloseableLock writeLock = tryWriteLock(TimeDuration.ONE_SECOND)) {
       final long oldCommittedIndex = getLastCommittedIndex();
       final long newCommitIndex = Math.min(majorityIndex, getFlushIndex());
       if (oldCommittedIndex < newCommitIndex) {
@@ -133,6 +133,9 @@ public boolean updateCommitIndex(long majorityIndex, long currentTerm, boolean i
           return commitIndex.updateIncreasingly(newCommitIndex, traceIndexChange);
         }
       }
+    } catch (InterruptedException e) {
+      LOG.warn("{}: Interrupted to updateCommitIndex: majorityIndex={}, currentTerm={}, isLeader={}",
+          getName(), majorityIndex, currentTerm, isLeader, e);
     }
     return false;
   }
@@ -375,6 +378,10 @@ public AutoCloseableLock writeLock() {
     return AutoCloseableLock.acquire(lock.writeLock());
   }
 
+  public AutoCloseableLock tryWriteLock(TimeDuration timeout) throws InterruptedException {
+    return AutoCloseableLock.tryAcquire(lock.writeLock(), null, timeout);
+  }
+
   public boolean hasWriteLock() {
     return this.lock.isWriteLockedByCurrentThread();
   }