Refactor handling of version conflicts and sequence numbers

Author: jasontedor

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

@@ -78,6 +78,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -86,6 +87,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
+import java.util.function.Supplier;
 
 public class InternalEngine extends Engine {
 
@@ -422,7 +424,7 @@ private SearcherManager createSearcherManager() throws EngineException {
 
     @Override
     public GetResult get(Get get, Function<String, Searcher> searcherFactory) throws EngineException {
-        try (ReleasableLock lock = readLock.acquire()) {
+        try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
             if (get.realtime()) {
                 VersionValue versionValue = versionMap.getUnderLock(get.uid());
@@ -444,11 +446,28 @@ public GetResult get(Get get, Function<String, Searcher> searcherFactory) throws
         }
     }
 
-    private boolean checkVersionConflict(
-            final Operation op,
-            final long currentVersion,
-            final long expectedVersion,
-            final boolean deleted) {
+    /**
+     * Checks for version conflicts. If a version conflict exists, the optional return value represents the operation result. Otherwise, if
+     * no conflicts are found, the optional return value is not present.
+     *
+     * @param <T>             the result type
+     * @param op              the operation
+     * @param currentVersion  the current version
+     * @param expectedVersion the expected version
+     * @param deleted         {@code true} if the current version is not found or represents a delete
+     * @param onSuccess       if there is a version conflict that can be ignored, the result of the operation
+     * @param onFailure       if there is a version conflict that can not be ignored, the result of the operation
+     * @return if there is a version conflict, the optional value is present and represents the operation result, otherwise the return value
+     * is not present
+     */
+    private <T extends Result> Optional<T> checkVersionConflict(
+        final Operation op,
+        final long currentVersion,
+        final long expectedVersion,
+        final boolean deleted,
+        final Supplier<T> onSuccess,
+        final Function<VersionConflictEngineException, T> onFailure) {
+        final T result;
         if (op.versionType() == VersionType.FORCE) {
             if (engineConfig.getIndexSettings().getIndexVersionCreated().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
                 // If index was created in 5.0 or later, 'force' is not allowed at all
@@ -462,14 +481,22 @@ private boolean checkVersionConflict(
         if (op.versionType().isVersionConflictForWrites(currentVersion, expectedVersion, deleted)) {
             if (op.origin().isRecovery()) {
                 // version conflict, but okay
-                return true;
+                result = onSuccess.get();
             } else {
                 // fatal version conflict
-                throw new VersionConflictEngineException(shardId, op.type(), op.id(),
+                final VersionConflictEngineException e =
+                    new VersionConflictEngineException(
+                        shardId,
+                        op.type(),
+                        op.id(),
                         op.versionType().explainConflictForWrites(currentVersion, expectedVersion, deleted));
+                result = onFailure.apply(e);
             }
+
+            return Optional.of(result);
+        } else {
+            return Optional.empty();
         }
-        return false;
     }
 
     private long checkDeletedAndGCed(VersionValue versionValue) {
@@ -485,7 +512,7 @@ private long checkDeletedAndGCed(VersionValue versionValue) {
     @Override
     public IndexResult index(Index index) {
         IndexResult result;
-        try (ReleasableLock lock = readLock.acquire()) {
+        try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
             if (index.origin().isRecovery()) {
                 // Don't throttle recovery operations
@@ -584,6 +611,7 @@ private IndexResult innerIndex(Index index) throws IOException {
         final Translog.Location location;
         final long updatedVersion;
         IndexResult indexResult = null;
+        long seqNo = index.seqNo();
         try (Releasable ignored = acquireLock(index.uid())) {
             lastWriteNanos = index.startTime();
             /* if we have an autoGeneratedID that comes into the engine we can potentially optimize
@@ -648,35 +676,32 @@ private IndexResult innerIndex(Index index) throws IOException {
                 }
             }
             final long expectedVersion = index.version();
-            final boolean conflict = checkVersionConflict(index, currentVersion, expectedVersion, deleted);
-
-            final long seqNo;
-            if (index.origin() == Operation.Origin.PRIMARY) {
-                if (!conflict) {
+            final Optional<IndexResult> checkVersionConflictResult =
+                checkVersionConflict(
+                    index,
+                    currentVersion,
+                    expectedVersion,
+                    deleted,
+                    () -> new IndexResult(currentVersion, index.seqNo(), false),
+                    e -> new IndexResult(e, currentVersion, index.seqNo()));
+
+            if (checkVersionConflictResult.isPresent()) {
+                indexResult = checkVersionConflictResult.get();
+            } else {
+                // no version conflict
+                if (index.origin() == Operation.Origin.PRIMARY) {
                     seqNo = seqNoService.generateSeqNo();
-                } else {
-                    seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
                 }
-            } else {
-                seqNo = index.seqNo();
-            }
 
-            if (conflict) {
-                // skip index operation because of version conflict on recovery
-                indexResult = new IndexResult(expectedVersion, seqNo, false);
-            } else {
+                /**
+                 * Update the document's sequence number and primary term; the sequence number here is derived here from either the sequence
+                 * number service if this is on the primary, or the existing document's sequence number if this is on the replica. The
+                 * primary term here has already been set, see IndexShard#prepareIndex where the Engine$Index operation is created.
+                 */
+                index.parsedDoc().updateSeqID(seqNo, index.primaryTerm());
                 updatedVersion = index.versionType().updateVersion(currentVersion, expectedVersion);
                 index.parsedDoc().version().setLongValue(updatedVersion);
 
-                // Update the document's sequence number and primary term, the
-                // sequence number here is derived here from either the sequence
-                // number service if this is on the primary, or the existing
-                // document's sequence number if this is on the replica. The
-                // primary term here has already been set, see
-                // IndexShard.prepareIndex where the Engine.Index operation is
-                // created
-                index.parsedDoc().updateSeqID(seqNo, index.primaryTerm());
-
                 if (currentVersion == Versions.NOT_FOUND && forceUpdateDocument == false) {
                     // document does not exists, we can optimize for create, but double check if assertions are running
                     assert assertDocDoesNotExist(index, canOptimizeAddDocument == false);
@@ -686,17 +711,17 @@ private IndexResult innerIndex(Index index) throws IOException {
                 }
                 indexResult = new IndexResult(updatedVersion, seqNo, deleted);
                 location = index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY
-                        ? translog.add(new Translog.Index(index, indexResult))
-                        : null;
+                    ? translog.add(new Translog.Index(index, indexResult))
+                    : null;
                 versionMap.putUnderLock(index.uid().bytes(), new VersionValue(updatedVersion));
                 indexResult.setTranslogLocation(location);
             }
             indexResult.setTook(System.nanoTime() - index.startTime());
             indexResult.freeze();
             return indexResult;
         } finally {
-            if (indexResult != null && indexResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
-                seqNoService.markSeqNoAsCompleted(indexResult.getSeqNo());
+            if (seqNo != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
+                seqNoService.markSeqNoAsCompleted(seqNo);
             }
         }
 
@@ -741,7 +766,7 @@ private static void update(final Term uid, final List<ParseContext.Document> doc
     @Override
     public DeleteResult delete(Delete delete) {
         DeleteResult result;
-        try (ReleasableLock lock = readLock.acquire()) {
+        try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
             // NOTE: we don't throttle this when merges fall behind because delete-by-id does not create new segments:
             result = innerDelete(delete);
@@ -766,6 +791,7 @@ private DeleteResult innerDelete(Delete delete) throws IOException {
         final long updatedVersion;
         final boolean found;
         DeleteResult deleteResult = null;
+        long seqNo = delete.seqNo();
         try (Releasable ignored = acquireLock(delete.uid())) {
             lastWriteNanos = delete.startTime();
             final long currentVersion;
@@ -782,38 +808,37 @@ private DeleteResult innerDelete(Delete delete) throws IOException {
 
             final long expectedVersion = delete.version();
 
-            final boolean conflict = checkVersionConflict(delete, currentVersion, expectedVersion, deleted);
-
-            final long seqNo;
-            if (delete.origin() == Operation.Origin.PRIMARY) {
-                if (!conflict) {
+            final Optional<DeleteResult> result =
+                checkVersionConflict(
+                    delete,
+                    currentVersion,
+                    expectedVersion,
+                    deleted,
+                    () -> new DeleteResult(expectedVersion, delete.seqNo(), true),
+                    e -> new DeleteResult(e, expectedVersion, delete.seqNo()));
+
+            if (result.isPresent()) {
+                deleteResult = result.get();
+            } else {
+                if (delete.origin() == Operation.Origin.PRIMARY) {
                     seqNo = seqNoService.generateSeqNo();
-                } else {
-                    seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
                 }
-            } else {
-                seqNo = delete.seqNo();
-            }
 
-            if (conflict) {
-                // skip executing delete because of version conflict on recovery
-                deleteResult = new DeleteResult(expectedVersion, seqNo, true);
-            } else {
                 updatedVersion = delete.versionType().updateVersion(currentVersion, expectedVersion);
                 found = deleteIfFound(delete.uid(), currentVersion, deleted, versionValue);
                 deleteResult = new DeleteResult(updatedVersion, seqNo, found);
                 location = delete.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY
-                        ? translog.add(new Translog.Delete(delete, deleteResult))
-                        : null;
+                    ? translog.add(new Translog.Delete(delete, deleteResult))
+                    : null;
                 versionMap.putUnderLock(delete.uid().bytes(),
-                        new DeleteVersionValue(updatedVersion, engineConfig.getThreadPool().estimatedTimeInMillis()));
+                    new DeleteVersionValue(updatedVersion, engineConfig.getThreadPool().estimatedTimeInMillis()));
                 deleteResult.setTranslogLocation(location);
             }
             deleteResult.setTook(System.nanoTime() - delete.startTime());
             deleteResult.freeze();
             return deleteResult;
         } finally {
-            if (deleteResult != null && deleteResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
+            if (seqNo != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
                 seqNoService.markSeqNoAsCompleted(deleteResult.getSeqNo());
             }
         }

core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -152,11 +152,14 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.LongSupplier;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyMap;
 import static org.elasticsearch.index.engine.Engine.Operation.Origin.PRIMARY;
 import static org.elasticsearch.index.engine.Engine.Operation.Origin.REPLICA;
+import static org.elasticsearch.index.engine.Engine.Operation.Origin.PEER_RECOVERY;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -2975,18 +2978,44 @@ public long generateSeqNo() {
         }
     }
 
-    public void testOutOfOrderSequenceNumbersWithVersionConflict() {
+    public void testOutOfOrderSequenceNumbersWithVersionConflict() throws IOException {
         final List<Engine.Operation> operations = new ArrayList<>();
+
         final int numberOfOperations = randomIntBetween(16, 32);
         final Term uid = newUid("1");
         final Document document = testDocumentWithTextField();
+        final AtomicLong sequenceNumber = new AtomicLong();
+        final Engine.Operation.Origin origin = randomFrom(PEER_RECOVERY, PRIMARY, PEER_RECOVERY);
+        final LongSupplier sequenceNumberSupplier =
+            origin == PRIMARY ? () -> SequenceNumbersService.UNASSIGNED_SEQ_NO : sequenceNumber::getAndIncrement;
         document.add(new Field(SourceFieldMapper.NAME, BytesReference.toBytes(B_1), SourceFieldMapper.Defaults.FIELD_TYPE));
         final ParsedDocument doc = testParsedDocument("1", "1", "test", null, document, B_1, null);
         for (int i = 0; i < numberOfOperations; i++) {
             if (randomBoolean()) {
-                operations.add(indexOperation(uid, doc, i, i));
+                final Engine.Index index = new Engine.Index(
+                    uid,
+                    doc,
+                    sequenceNumberSupplier.getAsLong(),
+                    1,
+                    i,
+                    VersionType.EXTERNAL,
+                    origin,
+                    System.nanoTime(),
+                    IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP,
+                    false);
+                operations.add(index);
             } else {
-                operations.add(deleteOperation("test", "1", uid, i, i));
+                final Engine.Delete delete = new Engine.Delete(
+                    "test",
+                    "1",
+                    uid,
+                    sequenceNumberSupplier.getAsLong(),
+                    1,
+                    i,
+                    VersionType.EXTERNAL,
+                    origin,
+                    System.nanoTime());
+                operations.add(delete);
             }
         }
 
@@ -3001,38 +3030,31 @@ public void testOutOfOrderSequenceNumbersWithVersionConflict() {
             }
         }
 
-        assertThat(engine.seqNoService().getLocalCheckpoint(), equalTo((long) (numberOfOperations - 1)));
+        final long expectedLocalCheckpoint;
+        if (origin == PRIMARY) {
+            // we can only advance as far as the number of operations that did not conflict
+            int count = 0;
+
+            // each time the version increments as we walk the list, that counts as a successful operation
+            long version = -1;
+            for (int i = 0; i < numberOfOperations; i++) {
+                if (operations.get(i).version() >= version) {
+                    count++;
+                    version = operations.get(i).version();
+                }
+            }
+
+            // sequence numbers start at zero, so the expected local checkpoint is the number of successful operations minus one
+            expectedLocalCheckpoint = count - 1;
+        } else {
+            expectedLocalCheckpoint = numberOfOperations - 1;
+        }
+
+        assertThat(engine.seqNoService().getLocalCheckpoint(), equalTo(expectedLocalCheckpoint));
         final Engine.GetResult result = engine.get(new Engine.Get(true, uid));
         assertThat(result.exists(), equalTo(exists));
     }
 
-    private Engine.Index indexOperation(final Term uid, final ParsedDocument doc, final int seqNo, final int version) {
-        return new Engine.Index(
-            uid,
-            doc,
-            seqNo,
-            1,
-            version,
-            VersionType.EXTERNAL,
-            Engine.Operation.Origin.PEER_RECOVERY,
-            System.nanoTime(),
-            IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP,
-            false);
-    }
-
-    private Engine.Delete deleteOperation(final String type, final String id, final Term uid, final int seqNo, final int version) {
-        return new Engine.Delete(
-            type,
-            id,
-            uid,
-            seqNo,
-            1,
-            version,
-            VersionType.EXTERNAL,
-            Engine.Operation.Origin.PEER_RECOVERY,
-            System.nanoTime());
-    }
-
     /**
      * Return a tuple representing the sequence ID for the given {@code Get}
      * operation. The first value in the tuple is the sequence number, the