Catch JournalReader.next() violations in Mode.COMMITS

JournalReader.next() is specified as an Iterator and should be throwing
NSEs. This is not the case when we ask for only commits, as the
corresponding check is done only in hasNext() and thus we would give out
uncommitted entries.

Fix this by adding an override to check for this condition.

JIRA: CONTROLLER-2106
Change-Id: I1f391cc19b9674646786c93656b83519e0f1f47b
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>

diff --git a/atomix-storage/src/main/java/io/atomix/storage/journal/CommitsSegmentJournalReader.java b/atomix-storage/src/main/java/io/atomix/storage/journal/CommitsSegmentJournalReader.java
index 8dfb14fa86380579c1c37d3b3aded624d8e36c2e..374eed901b1d40b6119f433a6c33029613caa9b9 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/storage/journal/CommitsSegmentJournalReader.java
+++ b/atomix-storage/src/main/java/io/atomix/storage/journal/CommitsSegmentJournalReader.java
@@ -15,6 +15,8 @@
  */
 package io.atomix.storage.journal;
 
+import java.util.NoSuchElementException;
+
 /**
  * A {@link JournalReader} traversing only committed entries.
  */
@@ -25,6 +27,18 @@ final class CommitsSegmentJournalReader<E> extends SegmentedJournalReader<E> {
 
     @Override
     public boolean hasNext() {
-        return getNextIndex() <= journal.getCommitIndex() && super.hasNext();
+        return isNextCommited() && super.hasNext();
+    }
+
+    @Override
+    public Indexed<E> next() {
+        if (isNextCommited()) {
+            return super.next();
+        }
+        throw new NoSuchElementException();
+    }
+
+    private boolean isNextCommited() {
+        return getNextIndex() <= journal.getCommitIndex();
     }
 }

diff --git a/atomix-storage/src/main/java/io/atomix/storage/journal/JournalReader.java b/atomix-storage/src/main/java/io/atomix/storage/journal/JournalReader.java
index ed2047441cc58403419496b0ef583e9592093fb2..730ab205b852d80fb24af3878219485733dd82d0 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/storage/journal/JournalReader.java
+++ b/atomix-storage/src/main/java/io/atomix/storage/journal/JournalReader.java
@@ -16,6 +16,7 @@
 package io.atomix.storage.journal;
 
 import java.util.Iterator;
+import java.util.NoSuchElementException;
 
 /**
  * Log reader.
@@ -80,6 +81,7 @@ public interface JournalReader<E> extends Iterator<Indexed<E>>, AutoCloseable {
    * Returns the next entry in the reader.
    *
    * @return The next entry in the reader.
+   * @throws NoSuchElementException there is no next entry
    */
   @Override
   Indexed<E> next();

diff --git a/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournalReader.java b/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournalReader.java
index 24a91589a6b72fe47ef281df733c01e98fb6c187..b77b9def430ceb5a8d6ff37a98c3b8fa3940cd82 100644 (file)
--- a/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournalReader.java
+++ b/atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedJournalReader.java
@@ -123,7 +123,7 @@ sealed class SegmentedJournalReader<E> implements JournalReader<E> permits Commi
   }
 
   @Override
-  public final Indexed<E> next() {
+  public Indexed<E> next() {
     if (currentReader.hasNext()) {
       previousEntry = currentReader.getCurrentEntry();
       return currentReader.next();

diff --git a/atomix-storage/src/test/java/io/atomix/storage/journal/AbstractJournalTest.java b/atomix-storage/src/test/java/io/atomix/storage/journal/AbstractJournalTest.java
index f2ef0b131265e2e6503a8e6c875aad3afe9ee2e2..54f6e6d5d80765e04d4a383134ce223138c75898 100644 (file)
--- a/atomix-storage/src/test/java/io/atomix/storage/journal/AbstractJournalTest.java
+++ b/atomix-storage/src/test/java/io/atomix/storage/journal/AbstractJournalTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
@@ -31,6 +32,7 @@ import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.NoSuchElementException;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -323,6 +325,28 @@ public abstract class AbstractJournalTest {
         }
     }
 
+    // Same as testWriteReadCommittedEntries(), but does not use hasNext() but checks whether an exception is thrown
+    @Test
+    public void testWriteReadCommittedEntriesException() throws Exception {
+        try (Journal<TestEntry> journal = createJournal()) {
+            JournalWriter<TestEntry> writer = journal.writer();
+            JournalReader<TestEntry> reader = journal.openReader(1, JournalReader.Mode.COMMITS);
+
+            for (int i = 1; i <= entriesPerSegment * 5; i++) {
+                writer.append(ENTRY);
+                assertThrows(NoSuchElementException.class, reader::next);
+                writer.commit(i);
+                Indexed<TestEntry> entry = reader.next();
+                assertEquals(i, entry.index());
+                assertEquals(32, entry.entry().bytes().length);
+                reader.reset(i);
+                entry = reader.next();
+                assertEquals(i, entry.index());
+                assertEquals(32, entry.entry().bytes().length);
+            }
+        }
+    }
+
     @Test
     public void testReadAfterCompact() throws Exception {
         try (SegmentedJournal<TestEntry> journal = createJournal()) {