Provide a lower and upper bound when internally reading translog operations (#26708)

translog: Add a method to retrieve translog operations between two specific sequence numbers
and let users of the existing Translog.getSnapshotBetween(...) indirectly may use of that too.

Author: martijnvg

core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -246,7 +246,7 @@ public void restoreLocalCheckpointFromTranslog() throws IOException {
         try (ReleasableLock ignored = writeLock.acquire()) {
             ensureOpen();
             final long localCheckpoint = seqNoService().getLocalCheckpoint();
-            try (Translog.Snapshot snapshot = getTranslog().newSnapshotFromMinSeqNo(localCheckpoint + 1)) {
+            try (Translog.Snapshot snapshot = getTranslog().newSnapshotFrom(localCheckpoint + 1)) {
                 Translog.Operation operation;
                 while ((operation = snapshot.next()) != null) {
                     if (operation.seqNo() > localCheckpoint) {

core/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java

@@ -82,7 +82,7 @@ public void resync(final IndexShard indexShard, final ActionListener<ResyncTask>
         ActionListener<ResyncTask> resyncListener = null;
         try {
             final long startingSeqNo = indexShard.getGlobalCheckpoint() + 1;
-            Translog.Snapshot snapshot = indexShard.getTranslog().newSnapshotFromMinSeqNo(startingSeqNo);
+            Translog.Snapshot snapshot = indexShard.getTranslog().newSnapshotFrom(startingSeqNo);
             resyncListener = new ActionListener<ResyncTask>() {
                 @Override
                 public void onResponse(final ResyncTask resyncTask) {

core/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -416,7 +416,9 @@ private int totalOperations(long minGeneration) {
     public int estimateTotalOperationsFromMinSeq(long minSeqNo) {
         try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
-            return readersAboveMinSeqNo(minSeqNo).mapToInt(BaseTranslogReader::totalOperations).sum();
+            return readersBetweenMinAndMaxSeqNo(minSeqNo, Long.MAX_VALUE)
+                    .mapToInt(BaseTranslogReader::totalOperations)
+                    .sum();
         }
     }
 
@@ -439,7 +441,7 @@ private long sizeInBytesByMinGen(long minGeneration) {
     private long sizeOfGensAboveSeqNoInBytes(long minSeqNo) {
         try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
-            return readersAboveMinSeqNo(minSeqNo).mapToLong(BaseTranslogReader::sizeInBytes).sum();
+            return readersBetweenMinAndMaxSeqNo(minSeqNo, Long.MAX_VALUE).mapToLong(BaseTranslogReader::sizeInBytes).sum();
         }
     }
 
@@ -597,11 +599,23 @@ public Snapshot newSnapshotFromGen(long minGeneration) throws IOException {
         }
     }
 
-    public Snapshot newSnapshotFromMinSeqNo(long minSeqNo) throws IOException {
+    /**
+     * Returns a snapshot with operations having a sequence number equal to or greater than <code>minSeqNo</code>.
+     */
+    public Snapshot newSnapshotFrom(long minSeqNo) throws IOException {
+        return getSnapshotBetween(minSeqNo, Long.MAX_VALUE);
+    }
+
+    /**
+     * Returns a snapshot with operations having a sequence number equal to or greater than <code>minSeqNo</code> and
+     * equal to or lesser than <code>maxSeqNo</code>.
+     */
+    public Snapshot getSnapshotBetween(long minSeqNo, long maxSeqNo) throws IOException {
         try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
-            TranslogSnapshot[] snapshots = readersAboveMinSeqNo(minSeqNo).map(BaseTranslogReader::newSnapshot)
-                .toArray(TranslogSnapshot[]::new);
+            TranslogSnapshot[] snapshots = readersBetweenMinAndMaxSeqNo(minSeqNo, maxSeqNo)
+                    .map(BaseTranslogReader::newSnapshot)
+                    .toArray(TranslogSnapshot[]::new);
             return newMultiSnapshot(snapshots);
         }
     }
@@ -627,14 +641,14 @@ private Snapshot newMultiSnapshot(TranslogSnapshot[] snapshots) throws IOExcepti
         }
     }
 
-    private Stream<? extends BaseTranslogReader> readersAboveMinSeqNo(long minSeqNo) {
-        assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread() :
-        "callers of readersAboveMinSeqNo must hold a lock: readLock ["
-            + readLock.isHeldByCurrentThread() + "], writeLock [" + readLock.isHeldByCurrentThread() + "]";
+    private Stream<? extends BaseTranslogReader> readersBetweenMinAndMaxSeqNo(long minSeqNo, long maxSeqNo) {
+        assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
+
         return Stream.concat(readers.stream(), Stream.of(current))
             .filter(reader -> {
-                final long maxSeqNo = reader.getCheckpoint().maxSeqNo;
-                return maxSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || maxSeqNo >= minSeqNo;
+                final Checkpoint checkpoint = reader.getCheckpoint();
+                return checkpoint.maxSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO ||
+                        checkpoint.minSeqNo <= maxSeqNo && checkpoint.maxSeqNo >= minSeqNo;
             });
     }
 

core/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -188,7 +188,7 @@ public RecoveryResponse recoverToTarget() throws IOException {
 
             logger.trace("snapshot translog for recovery; current size is [{}]", translog.estimateTotalOperationsFromMinSeq(startingSeqNo));
             final long targetLocalCheckpoint;
-            try(Translog.Snapshot snapshot = translog.newSnapshotFromMinSeqNo(startingSeqNo)) {
+            try(Translog.Snapshot snapshot = translog.newSnapshotFrom(startingSeqNo)) {
                 targetLocalCheckpoint = phase2(startingSeqNo, snapshot);
             } catch (Exception e) {
                 throw new RecoveryEngineException(shard.shardId(), 2, "phase2 failed", e);
@@ -244,7 +244,7 @@ boolean isTranslogReadyForSequenceNumberBasedRecovery() throws IOException {
             logger.trace("all operations up to [{}] completed, checking translog content", endingSeqNo);
 
             final LocalCheckpointTracker tracker = new LocalCheckpointTracker(shard.indexSettings(), startingSeqNo, startingSeqNo - 1);
-            try (Translog.Snapshot snapshot = shard.getTranslog().newSnapshotFromMinSeqNo(startingSeqNo)) {
+            try (Translog.Snapshot snapshot = shard.getTranslog().newSnapshotFrom(startingSeqNo)) {
                 Translog.Operation operation;
                 while ((operation = snapshot.next()) != null) {
                     if (operation.seqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {

core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -76,6 +76,7 @@
 import java.io.Closeable;
 import java.io.EOFException;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.charset.Charset;
@@ -88,6 +89,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -104,7 +106,9 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.LongStream;
+import java.util.stream.Stream;
 
 import static com.carrotsearch.randomizedtesting.RandomizedTest.randomLongBetween;
 import static org.elasticsearch.common.util.BigArrays.NON_RECYCLING_INSTANCE;
@@ -891,7 +895,7 @@ protected void doRun() throws Exception {
                         // these are what we expect the snapshot to return (and potentially some more).
                         Set<Translog.Operation> expectedOps = new HashSet<>(writtenOps.keySet());
                         expectedOps.removeIf(op -> op.seqNo() <= committedLocalCheckpointAtView);
-                        try (Translog.Snapshot snapshot = translog.newSnapshotFromMinSeqNo(committedLocalCheckpointAtView + 1L)) {
+                        try (Translog.Snapshot snapshot = translog.newSnapshotFrom(committedLocalCheckpointAtView + 1L)) {
                             Translog.Operation op;
                             while ((op = snapshot.next()) != null) {
                                 expectedOps.remove(op);
@@ -2484,7 +2488,7 @@ public void testMinSeqNoBasedAPI() throws IOException {
             }
             assertThat(translog.estimateTotalOperationsFromMinSeq(seqNo), equalTo(expectedSnapshotOps));
             int readFromSnapshot = 0;
-            try (Translog.Snapshot snapshot = translog.newSnapshotFromMinSeqNo(seqNo)) {
+            try (Translog.Snapshot snapshot = translog.newSnapshotFrom(seqNo)) {
                 assertThat(snapshot.totalOperations(), equalTo(expectedSnapshotOps));
                 Translog.Operation op;
                 while ((op = snapshot.next()) != null) {
@@ -2500,6 +2504,38 @@ public void testMinSeqNoBasedAPI() throws IOException {
         }
     }
 
+    public void testGetSnapshotBetween() throws IOException {
+        final int numOperations = randomIntBetween(2, 8196);
+        final List<Integer> sequenceNumbers = IntStream.range(0, numOperations).boxed().collect(Collectors.toList());
+        Collections.shuffle(sequenceNumbers, random());
+        for (Integer sequenceNumber : sequenceNumbers) {
+            translog.add(new Translog.NoOp(sequenceNumber, 0, "test"));
+            if (rarely()) {
+                translog.rollGeneration();
+            }
+        }
+        translog.rollGeneration();
+
+        final int iters = randomIntBetween(8, 32);
+        for (int iter = 0; iter < iters; iter++) {
+            int min = randomIntBetween(0, numOperations - 1);
+            int max = randomIntBetween(min, numOperations);
+            try (Translog.Snapshot snapshot = translog.getSnapshotBetween(min, max)) {
+                final List<Translog.Operation> operations = new ArrayList<>();
+                for (Translog.Operation operation = snapshot.next(); operation != null; operation = snapshot.next()) {
+                    if (operation.seqNo() >= min && operation.seqNo() <= max) {
+                        operations.add(operation);
+                    }
+                }
+                operations.sort(Comparator.comparingLong(Translog.Operation::seqNo));
+                Iterator<Translog.Operation> iterator = operations.iterator();
+                for (long expectedSeqNo = min; expectedSeqNo < max; expectedSeqNo++) {
+                    assertThat(iterator.next().seqNo(), equalTo(expectedSeqNo));
+                }
+            }
+        }
+    }
+
     public void testSimpleCommit() throws IOException {
         final int operations = randomIntBetween(1, 4096);
         long seqNo = 0;