Fix InternalEngineTests#assertOpsOnPrimary.

Backport of elastic/elasticsearch#37746

Author: kovrus

es/es-server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -147,6 +147,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 import java.util.function.ToLongBiFunction;
@@ -1529,32 +1530,101 @@ public void testVersionOnPrimaryWithConcurrentRefresh() throws Exception {
             }
         });
         refreshThread.start();
-        latch.await();
-        assertOpsOnPrimary(ops, Versions.NOT_FOUND, true, engine);
-        running.set(false);
-        refreshThread.join();
+        try {
+            latch.await();
+            assertOpsOnPrimary(ops, Versions.NOT_FOUND, true, engine);
+        } finally {
+            running.set(false);
+            refreshThread.join();
+        }
     }
 
-    private int assertOpsOnPrimary(List<Engine.Operation> ops, long currentOpVersion, boolean docDeleted, InternalEngine engine)
+    private int assertOpsOnPrimary(List<Engine.Operation> ops,
+                                   long currentOpVersion,
+                                   boolean docDeleted,
+                                   InternalEngine engine)
         throws IOException {
         String lastFieldValue = null;
         int opsPerformed = 0;
         long lastOpVersion = currentOpVersion;
         long lastOpSeqNo = UNASSIGNED_SEQ_NO;
-        long lastOpTerm = 0;
-        final AtomicLong currentTerm = new AtomicLong(1);
-        BiFunction<Long, Engine.Index, Engine.Index> indexWithVersion = (version, index) -> new Engine.Index(index.uid(), index.parsedDoc(),
-                                                                                                             UNASSIGNED_SEQ_NO, currentTerm.get(), version, index.versionType(), index.origin(), index.startTime(),
-                                                                                                             index.getAutoGeneratedIdTimestamp(), index.isRetry(), UNASSIGNED_SEQ_NO, 0);
-        BiFunction<Long, Engine.Delete, Engine.Delete> delWithVersion = (version, delete) -> new Engine.Delete(delete.type(), delete.id(),
-                                                                                                               delete.uid(), UNASSIGNED_SEQ_NO, currentTerm.get(), version, delete.versionType(), delete.origin(), delete.startTime(),
-                                                                                                               UNASSIGNED_SEQ_NO, 0);
-        TriFunction<Long, Long, Engine.Index, Engine.Index> indexWithSeq = (seqNo, term, index) -> new Engine.Index(index.uid(),
-                                                                                                                    index.parsedDoc(), UNASSIGNED_SEQ_NO, currentTerm.get(), index.version(), index.versionType(), index.origin(),
-                                                                                                                    index.startTime(), index.getAutoGeneratedIdTimestamp(), index.isRetry(), seqNo, term);
-        TriFunction<Long, Long, Engine.Delete, Engine.Delete> delWithSeq = (seqNo, term, delete) -> new Engine.Delete(delete.type(),
-                                                                                                                      delete.id(), delete.uid(), UNASSIGNED_SEQ_NO, currentTerm.get(), delete.version(), delete.versionType(), delete.origin(),
-                                                                                                                      delete.startTime(), seqNo, term);
+        long lastOpTerm = UNASSIGNED_PRIMARY_TERM;
+        PrimaryTermSupplier currentTerm = (PrimaryTermSupplier) engine.engineConfig.getPrimaryTermSupplier();
+        BiFunction<Long, Engine.Index, Engine.Index> indexWithVersion = (version, index) -> new Engine.Index(
+            index.uid(),
+            index.parsedDoc(),
+            UNASSIGNED_SEQ_NO,
+            currentTerm.get(),
+            version,
+            index.versionType(),
+            index.origin(),
+            index.startTime(),
+            index.getAutoGeneratedIdTimestamp(),
+            index.isRetry(),
+            UNASSIGNED_SEQ_NO,
+            0);
+        BiFunction<Long, Engine.Delete, Engine.Delete> delWithVersion = (version, delete) -> new Engine.Delete(
+            delete.type(),
+            delete.id(),
+            delete.uid(),
+            UNASSIGNED_SEQ_NO,
+            currentTerm.get(),
+            version,
+            delete.versionType(),
+            delete.origin(),
+            delete.startTime(),
+            UNASSIGNED_SEQ_NO,
+            0);
+        TriFunction<Long, Long, Engine.Index, Engine.Index> indexWithSeq = (seqNo, term, index) -> new Engine.Index(
+            index.uid(),
+            index.parsedDoc(),
+            UNASSIGNED_SEQ_NO,
+            currentTerm.get(),
+            index.version(),
+            index.versionType(),
+            index.origin(),
+            index.startTime(),
+            index.getAutoGeneratedIdTimestamp(),
+            index.isRetry(),
+            seqNo,
+            term);
+        TriFunction<Long, Long, Engine.Delete, Engine.Delete> delWithSeq = (seqNo, term, delete) -> new Engine.Delete(
+            delete.type(),
+            delete.id(),
+            delete.uid(),
+            UNASSIGNED_SEQ_NO,
+            currentTerm.get(),
+            delete.version(),
+            delete.versionType(),
+            delete.origin(),
+            delete.startTime(),
+            seqNo,
+            term);
+        Function<Engine.Index, Engine.Index> indexWithCurrentTerm = index -> new Engine.Index(
+            index.uid(),
+            index.parsedDoc(),
+            UNASSIGNED_SEQ_NO,
+            currentTerm.get(),
+            index.version(),
+            index.versionType(),
+            index.origin(),
+            index.startTime(),
+            index.getAutoGeneratedIdTimestamp(),
+            index.isRetry(),
+            index.getIfSeqNoMatch(),
+            index.getIfPrimaryTermMatch());
+        Function<Engine.Delete, Engine.Delete> deleteWithCurrentTerm = delete -> new Engine.Delete(
+            delete.type(),
+            delete.id(),
+            delete.uid(),
+            UNASSIGNED_SEQ_NO,
+            currentTerm.get(),
+            delete.version(),
+            delete.versionType(),
+            delete.origin(),
+            delete.startTime(),
+            delete.getIfSeqNoMatch(),
+            delete.getIfPrimaryTermMatch());
         for (Engine.Operation op : ops) {
             final boolean versionConflict = rarely();
             final boolean versionedOp = versionConflict || randomBoolean();
@@ -1566,7 +1636,8 @@ private int assertOpsOnPrimary(List<Engine.Operation> ops, long currentOpVersion
                 lastOpSeqNo;
             final long conflictingTerm = conflictingSeqNo == lastOpSeqNo || randomBoolean() ? lastOpTerm + 1 : lastOpTerm;
             if (rarely()) {
-                currentTerm.incrementAndGet();
+                currentTerm.set(currentTerm.get() + 1L);
+                engine.rollTranslogGeneration();
             }
             final long correctVersion = docDeleted && randomBoolean() ? Versions.MATCH_DELETED : lastOpVersion;
             logger.info("performing [{}]{}{}",
@@ -1597,7 +1668,7 @@ private int assertOpsOnPrimary(List<Engine.Operation> ops, long currentOpVersion
                             result = engine.index(indexWithVersion.apply(correctVersion, index));
                         }
                     } else {
-                        result = engine.index(index);
+                        result = engine.index(indexWithCurrentTerm.apply(index));
                     }
                     assertThat(result.isCreated(), equalTo(docDeleted));
                     assertThat(result.getVersion(), equalTo(Math.max(lastOpVersion + 1, 1)));
@@ -1631,16 +1702,16 @@ private int assertOpsOnPrimary(List<Engine.Operation> ops, long currentOpVersion
                     } else if (versionedOp) {
                         result = engine.delete(delWithVersion.apply(correctVersion, delete));
                     } else {
-                        result = engine.delete(delete);
+                        result = engine.delete(deleteWithCurrentTerm.apply(delete));
                     }
                     assertThat(result.isFound(), equalTo(docDeleted == false));
                     assertThat(result.getVersion(), equalTo(Math.max(lastOpVersion + 1, 1)));
                     assertThat(result.getResultType(), equalTo(Engine.Result.Type.SUCCESS));
                     assertThat(result.getFailure(), nullValue());
                     docDeleted = true;
                     lastOpVersion = result.getVersion();
-                    lastOpSeqNo = UNASSIGNED_SEQ_NO;
-                    lastOpTerm = 0;
+                    lastOpSeqNo = result.getSeqNo();
+                    lastOpTerm = result.getTerm();
                     opsPerformed++;
                 }
             }
@@ -1668,6 +1739,8 @@ private int assertOpsOnPrimary(List<Engine.Operation> ops, long currentOpVersion
                 engine.clearDeletedTombstones();
                 if (docDeleted) {
                     lastOpVersion = Versions.NOT_FOUND;
+                    lastOpSeqNo = UNASSIGNED_SEQ_NO;
+                    lastOpTerm = UNASSIGNED_PRIMARY_TERM;
                 }
             }
         }

es/es-testing/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java

@@ -135,11 +135,8 @@ public abstract class EngineTestCase extends ESTestCase {
     protected Path primaryTranslogDir;
     protected Path replicaTranslogDir;
     // A default primary term is used by engine instances created in this test.
-    protected AtomicLong primaryTerm = new AtomicLong();
+    protected final PrimaryTermSupplier primaryTerm = new PrimaryTermSupplier(0L);
 
-    protected static void assertVisibleCount(Engine engine, int numDocs) throws IOException {
-        assertVisibleCount(engine, numDocs, true);
-    }
 
     protected static void assertVisibleCount(Engine engine, int numDocs, boolean refresh) throws IOException {
         if (refresh) {
@@ -585,7 +582,7 @@ public void onFailedEngine(String reason, @Nullable Exception e) {
             intRefreshListenerList,
             new NoneCircuitBreakerService(),
             globalCheckpointSupplier,
-            primaryTerm::get,
+            primaryTerm,
             tombstoneDocSupplier());
     }
 
@@ -981,4 +978,25 @@ public static Translog getTranslog(Engine engine) {
         InternalEngine internalEngine = (InternalEngine) engine;
         return internalEngine.getTranslog();
     }
+
+    public static final class PrimaryTermSupplier implements LongSupplier {
+        private final AtomicLong term;
+
+        PrimaryTermSupplier(long initialTerm) {
+            this.term = new AtomicLong(initialTerm);
+        }
+
+        public long get() {
+            return term.get();
+        }
+
+        public void set(long newTerm) {
+            this.term.set(newTerm);
+        }
+
+        @Override
+        public long getAsLong() {
+            return get();
+        }
+    }
 }