Fix FollowingEngineTests#testOptimizeMultipleVersions

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

@@ -832,7 +832,11 @@ protected long generateSeqNoForOperationOnPrimary(final Operation operation) {
         return doGenerateSeqNoForOperation(operation);
     }
 
-    protected void advanceMaxSeqNoOfUpdatesOrDeletesOnPrimary(long seqNo) {
+    protected void advanceMaxSeqNoOfUpdateOnPrimary(long seqNo) {
+        advanceMaxSeqNoOfUpdatesOrDeletes(seqNo);
+    }
+
+    protected void advanceMaxSeqNoOfDeleteOnPrimary(long seqNo) {
         advanceMaxSeqNoOfUpdatesOrDeletes(seqNo);
     }
 
@@ -900,7 +904,7 @@ public IndexResult index(Index index) throws IOException {
 
                         final boolean toAppend = plan.indexIntoLucene && plan.useLuceneUpdateDocument == false;
                         if (toAppend == false) {
-                            advanceMaxSeqNoOfUpdatesOrDeletesOnPrimary(index.seqNo());
+                            advanceMaxSeqNoOfUpdateOnPrimary(index.seqNo());
                         }
                     } else {
                         markSeqNoAsSeen(index.seqNo());
@@ -1276,7 +1280,7 @@ public DeleteResult delete(Delete delete) throws IOException {
                         delete.primaryTerm(), delete.version(), delete.versionType(), delete.origin(), delete.startTime(),
                         delete.getIfSeqNo(), delete.getIfPrimaryTerm());
 
-                    advanceMaxSeqNoOfUpdatesOrDeletesOnPrimary(delete.seqNo());
+                    advanceMaxSeqNoOfDeleteOnPrimary(delete.seqNo());
                 } else {
                     markSeqNoAsSeen(delete.seqNo());
                 }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java

@@ -35,7 +35,7 @@
 /**
  * An engine implementation for following shards.
  */
-public final class FollowingEngine extends InternalEngine {
+public class FollowingEngine extends InternalEngine {
 
 
     /**
@@ -118,14 +118,27 @@ protected long generateSeqNoForOperationOnPrimary(final Operation operation) {
     }
 
     @Override
-    protected void advanceMaxSeqNoOfUpdatesOrDeletesOnPrimary(long seqNo) {
+    protected void advanceMaxSeqNoOfDeleteOnPrimary(long seqNo) {
         if (Assertions.ENABLED) {
             final long localCheckpoint = getProcessedLocalCheckpoint();
             final long maxSeqNoOfUpdates = getMaxSeqNoOfUpdatesOrDeletes();
             assert localCheckpoint < maxSeqNoOfUpdates || maxSeqNoOfUpdates >= seqNo :
                 "maxSeqNoOfUpdates is not advanced local_checkpoint=" + localCheckpoint + " msu=" + maxSeqNoOfUpdates + " seq_no=" + seqNo;
         }
-        super.advanceMaxSeqNoOfUpdatesOrDeletesOnPrimary(seqNo); // extra safe in production code
+
+        super.advanceMaxSeqNoOfDeleteOnPrimary(seqNo);
+    }
+
+    @Override
+    protected void advanceMaxSeqNoOfUpdateOnPrimary(long seqNo) {
+        // In some scenarios it is possible to advance maxSeqNoOfUpdatesOrDeletes over the leader
+        // maxSeqNoOfUpdatesOrDeletes, since in this engine (effectively it is a replica) we don't check if the previous version
+        // was a delete and it's possible to consider it as an update, advancing the max sequence number over the leader
+        // maxSeqNoOfUpdatesOrDeletes.
+        // The goal of this marker it's just an optimization and it won't affect the correctness or durability of the indexed documents.
+
+        // See FollowingEngineTests#testConcurrentUpdateOperationsWithDeletesCanAdvanceMaxSeqNoOfUpdates or #72527 for more details.
+        super.advanceMaxSeqNoOfUpdateOnPrimary(seqNo);
     }
 
     @Override

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineTests.java

@@ -18,9 +18,9 @@
 import org.elasticsearch.common.CheckedBiFunction;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.index.IndexSettings;
@@ -35,6 +35,7 @@
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.seqno.RetentionLeases;
 import org.elasticsearch.index.seqno.SequenceNumbers;
+import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.index.translog.Translog;
@@ -55,6 +56,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
@@ -291,12 +293,21 @@ private FollowingEngine createEngine(Store store, EngineConfig config) throws IO
 
     private Engine.Index indexForFollowing(String id, long seqNo, Engine.Operation.Origin origin) {
         final long version = randomBoolean() ? 1 : randomNonNegativeLong();
+        return indexForFollowing(id, seqNo, origin, version);
+    }
+
+    private Engine.Index indexForFollowing(String id, long seqNo, Engine.Operation.Origin origin, long version) {
         final ParsedDocument parsedDocument = EngineTestCase.createParsedDoc(id, null);
         return new Engine.Index(EngineTestCase.newUid(parsedDocument), parsedDocument, seqNo, primaryTerm.get(), version,
             VersionType.EXTERNAL, origin, System.currentTimeMillis(), IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP, randomBoolean(),
             SequenceNumbers.UNASSIGNED_SEQ_NO, 0);
     }
 
+    private Engine.Delete deleteForFollowing(String id, long seqNo, Engine.Operation.Origin origin, long version) {
+        return IndexShard.prepareDelete(id, seqNo, primaryTerm.get(), version, VersionType.EXTERNAL,
+            origin, SequenceNumbers.UNASSIGNED_SEQ_NO, SequenceNumbers.UNASSIGNED_PRIMARY_TERM);
+    }
+
     private Engine.Index indexForPrimary(String id) {
         final ParsedDocument parsedDoc = EngineTestCase.createParsedDoc(id, null);
         return new Engine.Index(EngineTestCase.newUid(parsedDoc), primaryTerm.get(), parsedDoc);
@@ -450,6 +461,85 @@ public void testOptimizeSingleDocConcurrently() throws Exception {
         });
     }
 
+    public void testConcurrentIndexOperationsWithDeletesCanAdvanceMaxSeqNoOfUpdates() throws Exception {
+        // See #72527 for more details
+        Settings followerSettings = Settings.builder()
+            .put("index.number_of_shards", 1)
+            .put("index.number_of_replicas", 0)
+            .put("index.version.created", Version.CURRENT)
+            .put("index.xpack.ccr.following_index", true)
+            .build();
+
+        IndexMetadata followerIndexMetadata = IndexMetadata.builder(index.getName()).settings(followerSettings).build();
+        IndexSettings followerIndexSettings = new IndexSettings(followerIndexMetadata, Settings.EMPTY);
+        try (Store followerStore = createStore(shardId, followerIndexSettings, newDirectory())) {
+            EngineConfig followerConfig =
+                engineConfig(shardId, followerIndexSettings, threadPool, followerStore, logger, xContentRegistry());
+            followerStore.createEmpty();
+            String translogUuid = Translog.createEmptyTranslog(followerConfig.getTranslogConfig().getTranslogPath(),
+                SequenceNumbers.NO_OPS_PERFORMED,
+                shardId,
+                1L
+            );
+            followerStore.associateIndexWithNewTranslog(translogUuid);
+            CountDownLatch concurrentDeleteOpLatch = new CountDownLatch(1);
+            final long indexNewDocWithSameIdSeqNo = 4;
+            FollowingEngine followingEngine = new FollowingEngine(followerConfig) {
+                @Override
+                protected void advanceMaxSeqNoOfUpdateOnPrimary(long seqNo) {
+                    if (seqNo == indexNewDocWithSameIdSeqNo) {
+                        // wait until the concurrent delete finishes meaning that processedLocalCheckpoint == maxSeqNoOfUpdatesOrDeletes
+                        try {
+                            concurrentDeleteOpLatch.await();
+                            assertThat(getProcessedLocalCheckpoint(), equalTo(getMaxSeqNoOfUpdatesOrDeletes()));
+                        } catch (Exception exception) {
+                            throw new RuntimeException(exception);
+                        }
+                    }
+                    super.advanceMaxSeqNoOfUpdateOnPrimary(seqNo);
+                }
+            };
+            TranslogHandler translogHandler = new TranslogHandler(xContentRegistry(), followerConfig.getIndexSettings());
+            followingEngine.recoverFromTranslog(translogHandler, Long.MAX_VALUE);
+            try (followingEngine) {
+                final long leaderMaxSeqNoOfUpdatesOnPrimary = 3;
+                followingEngine.advanceMaxSeqNoOfUpdatesOrDeletes(leaderMaxSeqNoOfUpdatesOnPrimary);
+
+                followingEngine.index(indexForFollowing("1", 0, Engine.Operation.Origin.PRIMARY, 1));
+                followingEngine.delete(deleteForFollowing("1", 1, Engine.Operation.Origin.PRIMARY, 2));
+                followingEngine.index(indexForFollowing("2", 2, Engine.Operation.Origin.PRIMARY, 1));
+                assertThat(followingEngine.getProcessedLocalCheckpoint(), equalTo(2L));
+
+                CyclicBarrier barrier = new CyclicBarrier(3);
+                Thread thread1 = new Thread(() -> {
+                    try {
+                        barrier.await();
+                        followingEngine.delete(deleteForFollowing("2", 3, Engine.Operation.Origin.PRIMARY, 2));
+                        concurrentDeleteOpLatch.countDown();
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+                Thread thread2 = new Thread(() -> {
+                    try {
+                        barrier.await();
+                        followingEngine.index(indexForFollowing("1", indexNewDocWithSameIdSeqNo, Engine.Operation.Origin.PRIMARY, 3));
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+
+                thread1.start();
+                thread2.start();
+                barrier.await();
+                thread1.join();
+                thread2.join();
+
+                assertThat(followingEngine.getMaxSeqNoOfUpdatesOrDeletes(), greaterThanOrEqualTo(leaderMaxSeqNoOfUpdatesOnPrimary));
+            }
+        }
+    }
+
     private void runFollowTest(CheckedBiConsumer<InternalEngine, FollowingEngine, Exception> task) throws Exception {
         final CheckedBiConsumer<InternalEngine, FollowingEngine, Exception> wrappedTask = (leader, follower) -> {
             Thread[] threads = new Thread[between(1, 8)];