Guard against duplicate log indexes

We saw an issue where a duplicate log index was added to the journal.
The duplicates were contiguous. It is unclear at this point how it
happened but we should guard against it so I added a check to ensure the
new index > the last index.

Change-Id: I5acc0fa5b4fe7f4352fc7935e7262834894878f3
Signed-off-by: Tom Pantelis <tpanteli@brocade.com>

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
index c245206f641f3a4ff31da8608076f1c3d68cb4f6..0d125ed6e2cc992619151212f990e83c7e8abafe 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
@@ -12,12 +12,17 @@ import com.google.common.base.Preconditions;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Abstract class handling the mapping of
  * logical LogEntry Index and the physical list index.
  */
 public abstract class AbstractReplicatedLogImpl implements ReplicatedLog {
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractReplicatedLogImpl.class);
+
+    private final String logContext;
 
     // We define this as ArrayList so we can use ensureCapacity.
     private ArrayList<ReplicatedLogEntry> journal;
@@ -32,18 +37,19 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog {
     private int dataSize = 0;
 
     public AbstractReplicatedLogImpl(long snapshotIndex,
-        long snapshotTerm, List<ReplicatedLogEntry> unAppliedEntries) {
+        long snapshotTerm, List<ReplicatedLogEntry> unAppliedEntries, String logContext) {
         this.snapshotIndex = snapshotIndex;
         this.snapshotTerm = snapshotTerm;
-        this.journal = new ArrayList<>(unAppliedEntries);
+        this.logContext = logContext;
 
-        for(ReplicatedLogEntry entry: journal) {
-            dataSize += entry.size();
+        this.journal = new ArrayList<>(unAppliedEntries.size());
+        for(ReplicatedLogEntry entry: unAppliedEntries) {
+            append(entry);
         }
     }
 
     public AbstractReplicatedLogImpl() {
-        this(-1L, -1L, Collections.<ReplicatedLogEntry>emptyList());
+        this(-1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), "");
     }
 
     protected int adjustedIndex(long logEntryIndex) {
@@ -112,9 +118,16 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog {
     }
 
     @Override
-    public void append(ReplicatedLogEntry replicatedLogEntry) {
-        journal.add(replicatedLogEntry);
-        dataSize += replicatedLogEntry.size();
+    public boolean append(ReplicatedLogEntry replicatedLogEntry) {
+        if(replicatedLogEntry.getIndex() > lastIndex()) {
+            journal.add(replicatedLogEntry);
+            dataSize += replicatedLogEntry.size();
+            return true;
+        } else {
+            LOG.warn("{}: Cannot append new entry - new index {} is not greater than the last index {}",
+                    logContext, replicatedLogEntry.getIndex(), lastIndex(), new Exception("stack trace"));
+            return false;
+        }
     }
 
     @Override

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLog.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLog.java
index ffa4cb117231f3f603e2cd8a51db3853fd9b2fe8..7ff6e86227451fda3d52ead7bb142e6235d0c52a 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLog.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLog.java
@@ -65,8 +65,10 @@ public interface ReplicatedLog {
      * Appends an entry to the log.
      *
      * @param replicatedLogEntry the entry to append
+     * @return true if the entry was successfully appended, false otherwise. An entry can fail to append if
+     *         the index is already included in the log.
      */
-    void append(ReplicatedLogEntry replicatedLogEntry);
+    boolean append(ReplicatedLogEntry replicatedLogEntry);
 
     /**
      * Optimization method to increase the capacity of the journal log prior to appending entries.

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImpl.java
index 7e1cb69afff8124d8cd794b3d68aec0f019d1e1b..9eaf6c35052c912c15f18b3d63853912640415d2 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImpl.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImpl.java
@@ -30,7 +30,7 @@ class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
 
     private ReplicatedLogImpl(final long snapshotIndex, final long snapshotTerm, final List<ReplicatedLogEntry> unAppliedEntries,
             final RaftActorContext context) {
-        super(snapshotIndex, snapshotTerm, unAppliedEntries);
+        super(snapshotIndex, snapshotTerm, unAppliedEntries, context.getId());
         this.context = Preconditions.checkNotNull(context);
     }
 
@@ -98,7 +98,9 @@ class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
         context.getLogger().debug("{}: Append log entry and persist {} ", context.getId(), replicatedLogEntry);
 
         // FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
-        append(replicatedLogEntry);
+        if(!append(replicatedLogEntry)) {
+            return;
+        }
 
         // When persisting events with persist it is guaranteed that the
         // persistent actor will not receive further commands between the

diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java
index a3e4396f37a95f698c8bc660fc558dbd443fd38a..8a8d313ed8935f262a02fbba992eb34935721523 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImplTest.java
@@ -332,7 +332,5 @@ public class AbstractReplicatedLogImplTest {
         @Override
         public void captureSnapshotIfReady(ReplicatedLogEntry replicatedLogEntry) {
         }
-
-
     }
 }

diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImplTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImplTest.java
index c6ee950d142afe1b0ff963aa9fef7b0fec6a2e0c..d2fe7dd4fe9c436c3954d3c63682653cf69f4271 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImplTest.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicatedLogImplTest.java
@@ -77,24 +77,48 @@ public class ReplicatedLogImplTest {
     public void testAppendAndPersistExpectingNoCapture() throws Exception {
         ReplicatedLog log = ReplicatedLogImpl.newInstance(context);
 
-        MockReplicatedLogEntry logEntry = new MockReplicatedLogEntry(1, 1, new MockPayload("1"));
+        MockReplicatedLogEntry logEntry1 = new MockReplicatedLogEntry(1, 1, new MockPayload("1"));
 
-        log.appendAndPersist(logEntry);
+        log.appendAndPersist(logEntry1);
 
-        verifyPersist(logEntry);
+        verifyPersist(logEntry1);
 
         assertEquals("size", 1, log.size());
 
         reset(mockPersistence);
 
+        MockReplicatedLogEntry logEntry2 = new MockReplicatedLogEntry(1, 2, new MockPayload("2"));
+        Procedure<ReplicatedLogEntry> mockCallback = Mockito.mock(Procedure.class);
+        log.appendAndPersist(logEntry2, mockCallback);
+
+        verifyPersist(logEntry2);
+
+        verify(mockCallback).apply(same(logEntry2));
+
+        assertEquals("size", 2, log.size());
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void testAppendAndPersisWithDuplicateEntry() throws Exception {
+        ReplicatedLog log = ReplicatedLogImpl.newInstance(context);
+
         Procedure<ReplicatedLogEntry> mockCallback = Mockito.mock(Procedure.class);
+        MockReplicatedLogEntry logEntry = new MockReplicatedLogEntry(1, 1, new MockPayload("1"));
+
         log.appendAndPersist(logEntry, mockCallback);
 
         verifyPersist(logEntry);
 
-        verify(mockCallback).apply(same(logEntry));
+        assertEquals("size", 1, log.size());
 
-        assertEquals("size", 2, log.size());
+        reset(mockPersistence, mockCallback);
+
+        log.appendAndPersist(logEntry, mockCallback);
+
+        verifyNoMoreInteractions(mockPersistence, mockCallback);
+
+        assertEquals("size", 1, log.size());
     }
 
     @Test