Memory optimize at log manager.

jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

@@ -432,6 +432,7 @@ private long runApplyTask(final ApplyTask task, long maxCommittedIndex, final bo
                 }
             } finally {
                 this.nodeMetrics.recordLatency(task.type.metricName(), Utils.monotonicMs() - startMs);
+                task.reset();
             }
         }
         try {

jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java

@@ -1360,7 +1360,9 @@ private void executeApplyingTasks(final List<LogEntryAndClosure> tasks) {
                 final List<LogEntryAndClosure> savedTasks = new ArrayList<>(tasks);
                 Utils.runInThread(() -> {
                     for (int i = 0; i < size; i++) {
-                        savedTasks.get(i).done.run(st);
+                        final LogEntryAndClosure task = savedTasks.get(i);
+                        task.done.run(st);
+                        task.reset();
                     }
                 });
                 return;
@@ -1375,18 +1377,21 @@ private void executeApplyingTasks(final List<LogEntryAndClosure> tasks) {
                         final Status st = new Status(RaftError.EPERM, "expected_term=%d doesn't match current_term=%d",
                             task.expectedTerm, this.currTerm);
                         Utils.runClosureInThread(task.done, st);
+                        task.reset();
                     }
                     continue;
                 }
                 if (!this.ballotBox.appendPendingTask(this.conf.getConf(),
                     this.conf.isStable() ? null : this.conf.getOldConf(), task.done)) {
                     Utils.runClosureInThread(task.done, new Status(RaftError.EINTERNAL, "Fail to append task."));
+                    task.reset();
                     continue;
                 }
                 // set task entry info before adding to list.
                 task.entry.getId().setTerm(this.currTerm);
                 task.entry.setType(EnumOutter.EntryType.ENTRY_TYPE_DATA);
                 entries.add(task.entry);
+                task.reset();
             }
             this.logManager.appendEntries(entries, new LeaderStableClosure(entries));
             // update conf.first

jraft-core/src/main/java/com/alipay/sofa/jraft/storage/impl/LogManagerImpl.java

@@ -506,16 +506,20 @@ public void onEvent(final StableClosureEvent event, final long sequence, final b
                 this.lastId = this.ab.flush();
                 setDiskId(this.lastId);
                 LogManagerImpl.this.shutDownLatch.countDown();
+                event.reset();
                 return;
             }
             final StableClosure done = event.done;
+            final EventType eventType = event.type;
+
+            event.reset();
 
             if (done.getEntries() != null && !done.getEntries().isEmpty()) {
                 this.ab.append(done);
             } else {
                 this.lastId = this.ab.flush();
                 boolean ret = true;
-                switch (event.type) {
+                switch (eventType) {
                     case LAST_LOG_ID:
                         ((LastLogIdClosure) done).setLastLogId(this.lastId.copy());
                         break;

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/util/concurrent/disruptor/MessageEvent.java

@@ -28,6 +28,10 @@ public T getMessage() {
         return message;
     }
 
+    public void reset() {
+        this.message = null;
+    }
+
     public void setMessage(T message) {
         this.message = message;
     }

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/util/concurrent/disruptor/TaskHandler.java

@@ -39,11 +39,13 @@ public class TaskHandler implements EventHandler<MessageEvent<Runnable>>, WorkHa
     public void onEvent(final MessageEvent<Runnable> event, final long sequence, final boolean endOfBatch)
                                                                                                           throws Exception {
         event.getMessage().run();
+        event.reset();
     }
 
     @Override
     public void onEvent(final MessageEvent<Runnable> event) throws Exception {
         event.getMessage().run();
+        event.reset();
     }
 
     @Override