Ensure only unapplied committed entries are applied on leader startup.

atomix · Jun 1, 2015 · dceb49e · dceb49e
1 parent 803fa49
commit dceb49e
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 10 deletions.
diff --git a/raft/src/main/java/net/kuujo/copycat/raft/state/LeaderState.java b/raft/src/main/java/net/kuujo/copycat/raft/state/LeaderState.java
@@ -56,7 +56,6 @@ public synchronized CompletableFuture<AbstractState> open() {
     context.getContext().execute(() -> {
       commitEntries().whenComplete((result, error) -> {
         if (error == null) {
-          applyEntries();
           startHeartbeatTimer();
         }
       });
@@ -90,6 +89,7 @@ private CompletableFuture<Void> commitEntries() {
       context.checkThread();
       if (isOpen()) {
         if (error == null) {
+          applyEntries(resultIndex);
           future.complete(null);
         } else {
           transition(RaftState.FOLLOWER);
@@ -102,25 +102,24 @@ private CompletableFuture<Void> commitEntries() {
   /**
    * Applies all unapplied entries to the log.
    */
-  private void applyEntries() {
+  private void applyEntries(long index) {
     if (!context.getLog().isEmpty()) {
       int count = 0;
-      long lastIndex = context.getLog().lastIndex();
-      for (long commitIndex = Math.max(context.getCommitIndex(), context.getLog().firstIndex()); commitIndex <= lastIndex; commitIndex++) {
-        try (Entry entry = context.getLog().getEntry(commitIndex)) {
+      for (long lastApplied = Math.max(context.getLastApplied(), context.getLog().firstIndex()); lastApplied <= index; lastApplied++) {
+        try (Entry entry = context.getLog().getEntry(lastApplied)) {
           if (entry != null) {
             try {
               context.getStateMachine().apply(entry);
             } catch (ApplicationException e) {
-
+              LOGGER.info("{} - an application error occurred: {}", context.getCluster().member().id(), e);
             } finally {
-              context.setLastApplied(commitIndex);
+              context.setLastApplied(lastApplied);
             }
           }
         }
         count++;
       }
-      LOGGER.debug("{} - Applied {} entries to log", context.getCluster().member().id(), count);
+      LOGGER.debug("{} - applied {} entries to log", context.getCluster().member().id(), count);
     }
   }
 

diff --git a/raft/src/main/java/net/kuujo/copycat/raft/state/RaftContext.java b/raft/src/main/java/net/kuujo/copycat/raft/state/RaftContext.java
@@ -439,9 +439,9 @@ public long getCommitIndex() {
    */
   RaftContext setGlobalIndex(long globalIndex) {
     if (globalIndex < 0)
-      throw new IllegalArgumentException("recycle index must be positive");
+      throw new IllegalArgumentException("global index must be positive");
     if (globalIndex < this.globalIndex)
-      throw new IllegalArgumentException("cannot decrease recycle index");
+      throw new IllegalArgumentException("cannot decrease global index");
     this.globalIndex = globalIndex;
     compactor.setCompactIndex(globalIndex);
     return this;