feedback

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

@@ -168,8 +168,6 @@ public RecoveryResponse recoverToTarget() throws IOException {
                 // but we must have everything above the local checkpoint in the commit
                 requiredSeqNoRangeStart =
                     Long.parseLong(phase1Snapshot.getIndexCommit().getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)) + 1;
-                assert requiredSeqNoRangeStart >= 0 :
-                    "base commit contains an illegal local checkpoint " + (requiredSeqNoRangeStart - 1);
                 try {
                     phase1(phase1Snapshot.getIndexCommit(), translog::totalOperations);
                 } catch (final Exception e) {
@@ -182,6 +180,9 @@ public RecoveryResponse recoverToTarget() throws IOException {
                     }
                 }
             }
+            assert startingSeqNo >= 0 : "startingSeqNo must be non negative. got: " + startingSeqNo;
+            assert requiredSeqNoRangeStart >= startingSeqNo : "requiredSeqNoRangeStart [" + requiredSeqNoRangeStart + "] is lower than ["
+                + startingSeqNo + "]";
 
             runUnderPrimaryPermit(() -> shard.initiateTracking(request.targetAllocationId()));
 
@@ -190,7 +191,15 @@ public RecoveryResponse recoverToTarget() throws IOException {
             } catch (final Exception e) {
                 throw new RecoveryEngineException(shard.shardId(), 1, "prepare target for translog failed", e);
             }
-            final long endingSeqNo = determineEndingSeqNo();
+
+            final long endingSeqNo = shard.seqNoStats().getMaxSeqNo();
+            /*
+             * We need to wait for all operations up to the current max to complete, otherwise we can not guarantee that all
+             * operations in the required range will be available for replaying from the translog of the source.
+             */
+            cancellableThreads.execute(() -> shard.waitForOpsToComplete(endingSeqNo));
+
+            logger.trace("all operations up to [{}] completed, which will be used as an ending sequence number", endingSeqNo);
 
             logger.trace("snapshot translog for recovery; current size is [{}]", translog.estimateTotalOperationsFromMinSeq(startingSeqNo));
             final long targetLocalCheckpoint;
@@ -227,18 +236,6 @@ private void runUnderPrimaryPermit(CancellableThreads.Interruptable runnable) {
         });
     }
 
-    private long determineEndingSeqNo() {
-        final long endingSeqNo = shard.seqNoStats().getMaxSeqNo();
-        /*
-         * We need to wait for all operations up to the current max to complete, otherwise we can not guarantee that all
-         * operations in the required range will be available for replaying from the translog of the source.
-         */
-        cancellableThreads.execute(() -> shard.waitForOpsToComplete(endingSeqNo));
-
-        logger.trace("all operations up to [{}] completed, which will be used as an ending sequence number", endingSeqNo);
-        return endingSeqNo + 1;
-    }
-
     /**
      * Determines if the source translog is ready for a sequence-number-based peer recovery. The main condition here is that the source
      * translog contains all operations above the local checkpoint on the target. We already know the that translog contains or will contain
@@ -525,16 +522,16 @@ static class SendSnapshotResult {
      * Operations are bulked into a single request depending on an operation count limit or size-in-bytes limit.
      *
      * @param startingSeqNo the sequence number for which only operations with a sequence number greater than this will be sent
-     * @param requiredSeqNoRangeStart the lower sequence number of the required range (ending with endingSeqNo - 1)
-     * @param endingSeqNo   the upper bound of the sequence number range to be sent (exclusive)
+     * @param requiredSeqNoRangeStart the lower sequence number of the required range
+     * @param endingSeqNo   the upper bound of the sequence number range to be sent (inclusive)
      * @param snapshot      the translog snapshot to replay operations from  @return the local checkpoint on the target and the total
      *                      number of operations sent
      * @throws IOException if an I/O exception occurred reading the translog snapshot
      */
     protected SendSnapshotResult sendSnapshot(final long startingSeqNo, long requiredSeqNoRangeStart, long endingSeqNo,
                                               final Translog.Snapshot snapshot) throws IOException {
-        assert requiredSeqNoRangeStart <= endingSeqNo :
-            "requiredSeqNoRangeStart " + requiredSeqNoRangeStart + " is larger than endingSeqNo" + endingSeqNo;
+        assert requiredSeqNoRangeStart <= endingSeqNo + 1:
+            "requiredSeqNoRangeStart " + requiredSeqNoRangeStart + " is larger than endingSeqNo " + endingSeqNo;
         assert startingSeqNo <= requiredSeqNoRangeStart :
             "startingSeqNo " + startingSeqNo + " is larger than requiredSeqNoRangeStart " + requiredSeqNoRangeStart;
         int ops = 0;
@@ -562,7 +559,7 @@ protected SendSnapshotResult sendSnapshot(final long startingSeqNo, long require
             cancellableThreads.checkForCancel();
 
             final long seqNo = operation.seqNo();
-            if (seqNo < startingSeqNo || seqNo >= endingSeqNo) {
+            if (seqNo < startingSeqNo || seqNo > endingSeqNo) {
                 skippedOps++;
                 continue;
             }
@@ -587,7 +584,7 @@ protected SendSnapshotResult sendSnapshot(final long startingSeqNo, long require
             cancellableThreads.executeIO(sendBatch);
         }
 
-        if (requiredOpsTracker.getCheckpoint() < endingSeqNo - 1) {
+        if (requiredOpsTracker.getCheckpoint() < endingSeqNo) {
             throw new IllegalStateException("translog replay failed to covered required sequence numbers" +
                 " (required range [" + requiredSeqNoRangeStart + ":" + endingSeqNo + "). first missing op is ["
                 + (requiredOpsTracker.getCheckpoint() + 1) + "]");

core/src/test/java/org/elasticsearch/indices/recovery/RecoverySourceHandlerTests.java

@@ -188,7 +188,7 @@ public void testSendSnapshotSendsOps() throws IOException {
         operations.add(null);
         final long startingSeqNo = randomIntBetween(0, numberOfDocsWithValidSequenceNumbers - 1);
         final long requiredStartingSeqNo = randomIntBetween((int) startingSeqNo, numberOfDocsWithValidSequenceNumbers - 1);
-        final long endingSeqNo = randomIntBetween((int) requiredStartingSeqNo, numberOfDocsWithValidSequenceNumbers);
+        final long endingSeqNo = randomIntBetween((int) requiredStartingSeqNo, numberOfDocsWithValidSequenceNumbers - 1);
         // todo add proper tests
         RecoverySourceHandler.SendSnapshotResult result = handler.sendSnapshot(startingSeqNo, requiredStartingSeqNo,
             endingSeqNo, new Translog.Snapshot() {
@@ -209,7 +209,7 @@ public Translog.Operation next() throws IOException {
                     return operations.get(counter++);
                 }
             });
-        final int expectedOps = (int) (endingSeqNo - startingSeqNo);
+        final int expectedOps = (int) (endingSeqNo - startingSeqNo + 1);
         assertThat(result.totalOperations, equalTo(expectedOps));
         final ArgumentCaptor<List> shippedOpsCaptor = ArgumentCaptor.forClass(List.class);
         if (expectedOps > 0) {
@@ -227,7 +227,7 @@ public Translog.Operation next() throws IOException {
         if (endingSeqNo >= requiredStartingSeqNo + 1) {
             // check that missing ops blows up
             List<Translog.Operation> requiredOps = operations.subList(0, operations.size() - 1).stream() // remove last null marker
-                .filter(o -> o.seqNo() >= requiredStartingSeqNo && o.seqNo() < endingSeqNo).collect(Collectors.toList());
+                .filter(o -> o.seqNo() >= requiredStartingSeqNo && o.seqNo() <= endingSeqNo).collect(Collectors.toList());
             List<Translog.Operation> opsToSkip = randomSubsetOf(randomIntBetween(1, requiredOps.size()), requiredOps);
             expectThrows(IllegalStateException.class, () ->
                 handler.sendSnapshot(startingSeqNo, requiredStartingSeqNo,