RATIS-2227. LogEntryProto leak in SegmentedRaftLog (#1199)

Author: jianghuazhu

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

@@ -271,7 +271,7 @@ static class TraceInfo {
       }
 
       static <T> int findFirstUnequalFromTail(T[] current, T[] previous) {
-        int c = current.length - 1;
+        int c = current.length == 0 ? 0 : current.length - 1;
         for(int p = previous.length - 1; p >= 0; p--, c--) {
           if (!previous[p].equals(current[c])) {
             return c;

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderBase.java

@@ -261,7 +261,7 @@ protected ReferenceCountedObject<AppendEntriesRequestProto> nextAppendEntriesReq
     final long halfMs = heartbeatWaitTimeMs/2;
     final Map<Long, ReferenceCountedObject<EntryWithData>> offered = new HashMap<>();
     for (long next = followerNext; leaderNext > next && getHeartbeatWaitTimeMs() - halfMs > 0; next++) {
-      ReferenceCountedObject<EntryWithData> entryWithData = null;
+      final ReferenceCountedObject<EntryWithData> entryWithData;
       try {
         entryWithData = getRaftLog().retainEntryWithData(next);
         if (!buffer.offer(entryWithData.get())) {
@@ -270,9 +270,11 @@ protected ReferenceCountedObject<AppendEntriesRequestProto> nextAppendEntriesReq
         }
         offered.put(next, entryWithData);
       } catch (Exception e){
-        if (entryWithData != null) {
-          entryWithData.release();
+        for (ReferenceCountedObject<EntryWithData> ref : offered.values()) {
+          ref.release();
         }
+        offered.clear();
+        throw e;
       }
     }
     if (buffer.isEmpty()) {