translog: Add a method to retrieve translog operations between two specific sequence numbers

and let users of the existing Translog.newSnapshotFromMinSeqNo(...) indirectly may use of that too.

Author: martijnvg

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

@@ -2424,6 +2424,14 @@ public void addRefreshListener(Translog.Location location, Consumer<Boolean> lis
         }
     }
 
+    /**
+     * Returns a snapshot with translog operations having a sequence number equal or greater than <code>minSeqNo</code> and
+     * equal or lesser than <code>maxSeqNo</code>.
+     */
+    public Translog.Snapshot getTranslogOperationsBetweenMinAndMaxSeqNo(long minSeqNo, long maxSeqNo) throws IOException {
+        return getEngine().getTranslog().newSnapshotBetweenMinAndMaxSeqNo(minSeqNo, maxSeqNo);
+    }
+
     private static class RefreshMetricUpdater implements ReferenceManager.RefreshListener {
 
         private final MeanMetric refreshMetric;

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();
         }
     }
 
@@ -433,16 +435,6 @@ private long sizeInBytesByMinGen(long minGeneration) {
         }
     }
 
-    /**
-     * Returns the size in bytes of the translog files with ops above the given seqNo
-     */
-    private long sizeOfGensAboveSeqNoInBytes(long minSeqNo) {
-        try (ReleasableLock ignored = readLock.acquire()) {
-            ensureOpen();
-            return readersAboveMinSeqNo(minSeqNo).mapToLong(BaseTranslogReader::sizeInBytes).sum();
-        }
-    }
-
     /**
      * Creates a new translog for the specified generation.
      *
@@ -598,10 +590,19 @@ public Snapshot newSnapshotFromGen(long minGeneration) throws IOException {
     }
 
     public Snapshot newSnapshotFromMinSeqNo(long minSeqNo) throws IOException {
+        return newSnapshotBetweenMinAndMaxSeqNo(minSeqNo, Long.MAX_VALUE);
+    }
+
+    /**
+     * Returns a snapshot with operations having a sequence number equal or greater than <code>minSeqNo</code> and
+     * equal or lesser than <code>maxSeqNo</code>.
+     */
+    public Snapshot newSnapshotBetweenMinAndMaxSeqNo(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 +628,16 @@ private Snapshot newMultiSnapshot(TranslogSnapshot[] snapshots) throws IOExcepti
         }
     }
 
-    private Stream<? extends BaseTranslogReader> readersAboveMinSeqNo(long minSeqNo) {
+    private Stream<? extends BaseTranslogReader> readersBetweenMinAndMaxSeqNo(long minSeqNo, long maxSeqNo) {
         assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread() :
-        "callers of readersAboveMinSeqNo must hold a lock: readLock ["
+        "callers of readersBetweenMinAndMaxSeqNo must hold a lock: readLock ["
             + readLock.isHeldByCurrentThread() + "], writeLock [" + readLock.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/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -2497,6 +2497,35 @@ public void testMinSeqNoBasedAPI() throws IOException {
         }
     }
 
+    public void testGetOperationsBetweenMinAndMaxSeqNoAPI() throws IOException {
+        final int operations = randomIntBetween(2, 8096);
+        long seqNo = 0;
+        for (int i = 0; i < operations; i++) {
+            translog.add(new Translog.NoOp(seqNo++, 0, "test'"));
+            if (rarely()) {
+                translog.rollGeneration();
+            }
+        }
+        translog.rollGeneration();
+
+        int iters = randomIntBetween(8, 32);
+        for (int iter = 0; iter < iters; iter++) {
+            int min = randomIntBetween(0, operations - 1);
+            int max = randomIntBetween(min, operations);
+            try (Translog.Snapshot snapshot = translog.newSnapshotBetweenMinAndMaxSeqNo(min, max)) {
+                Translog.Operation operation;
+                do {
+                    operation = snapshot.next();
+                } while (operation.seqNo() < (min - 1));
+
+                for (long expectedSeqNo = min; expectedSeqNo < max; expectedSeqNo++) {
+                    operation = snapshot.next();
+                    assertThat(operation.seqNo(), equalTo(expectedSeqNo));
+                }
+            }
+        }
+    }
+
     public void testSimpleCommit() throws IOException {
         final int operations = randomIntBetween(1, 4096);
         long seqNo = 0;