Replica recovery could go into an endless flushing loop

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

@@ -817,6 +817,12 @@ public final boolean refreshNeeded() {
     // NOTE: do NOT rename this to something containing flush or refresh!
     public abstract void writeIndexingBuffer() throws EngineException;
 
+    /**
+     * Checks if this engine should be flushed to free translog.
+     * This check is mainly based on the uncommitted translog size and the translog flush threshold setting.
+     */
+    public abstract boolean shouldFlushToFreeTranslog();
+
     /**
      * Flushes the state of the engine including the transaction log, clearing memory.
      *

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

@@ -1462,6 +1462,25 @@ final boolean tryRenewSyncCommit() {
         return renewed;
     }
 
+    @Override
+    public boolean shouldFlushToFreeTranslog() {
+        ensureOpen();
+        final long flushThreshold = config().getIndexSettings().getFlushThresholdSize().getBytes();
+        final long uncommittedSizeOfCurrentCommit = translog.uncommittedSizeInBytes();
+        // If flushThreshold is too small, we may continuously flush even there is no uncommitted operations.
+        if (uncommittedSizeOfCurrentCommit < flushThreshold || translog.uncommittedOperations() == 0) {
+            return false;
+        }
+        /*
+         * We should only flush ony if the shouldFlush condition can become false after flushing.
+         * This condition will change if the `uncommittedSize` of the new commit is smaller than
+         * the `uncommittedSize` of the current commit. This method is to maintain translog only,
+         * thus the IndexWriter#hasUncommittedChanges condition is not considered.
+         */
+        final long uncommittedSizeOfNewCommit = translog.sizeOfGensAboveSeqNoInBytes(localCheckpointTracker.getCheckpoint() + 1);
+        return uncommittedSizeOfNewCommit < uncommittedSizeOfCurrentCommit;
+    }
+
     @Override
     public CommitId flush() throws EngineException {
         return flush(false, false);
@@ -1492,7 +1511,9 @@ public CommitId flush(boolean force, boolean waitIfOngoing) throws EngineExcepti
                 logger.trace("acquired flush lock immediately");
             }
             try {
-                if (indexWriter.hasUncommittedChanges() || force) {
+                // Only flush if (1) Lucene has uncommitted docs, or (2) forced by caller, or (3) the
+                // newly created commit points to a different translog generation (can free translog)
+                if (indexWriter.hasUncommittedChanges() || force || shouldFlushToFreeTranslog()) {
                     ensureCanFlush();
                     try {
                         translog.rollGeneration();

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

@@ -1597,17 +1597,16 @@ public boolean restoreFromRepository(Repository repository) {
     }
 
     /**
-     * Tests whether or not the translog should be flushed. This test is based on the current size of the translog comparted to the
-     * configured flush threshold size.
+     * Tests whether or not the engine should be flushed to free translog.
+     * This test is based on the current size of the translog compared to the configured flush threshold size.
      *
-     * @return {@code true} if the translog should be flushed
+     * @return {@code true} if the engine should be flushed
      */
     boolean shouldFlush() {
         final Engine engine = getEngineOrNull();
         if (engine != null) {
             try {
-                final Translog translog = engine.getTranslog();
-                return translog.shouldFlush();
+                return engine.shouldFlushToFreeTranslog();
             } catch (final AlreadyClosedException e) {
                 // we are already closed, no need to flush or roll
             }

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -436,7 +436,7 @@ 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) {
+    public long sizeOfGensAboveSeqNoInBytes(long minSeqNo) {
         try (ReleasableLock ignored = readLock.acquire()) {
             ensureOpen();
             return readersAboveMinSeqNo(minSeqNo).mapToLong(BaseTranslogReader::sizeInBytes).sum();
@@ -523,17 +523,6 @@ public Location add(final Operation operation) throws IOException {
         }
     }
 
-    /**
-     * Tests whether or not the translog should be flushed. This test is based on the current size
-     * of the translog comparted to the configured flush threshold size.
-     *
-     * @return {@code true} if the translog should be flushed
-     */
-    public boolean shouldFlush() {
-        final long size = this.uncommittedSizeInBytes();
-        return size > this.indexSettings.getFlushThresholdSize().getBytes();
-    }
-
     /**
      * Tests whether or not the translog generation should be rolled to a new generation. This test
      * is based on the size of the current generation compared to the configured generation

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.index.engine;
 
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
+import com.carrotsearch.randomizedtesting.generators.RandomNumbers;
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -46,6 +47,7 @@
 import org.apache.lucene.index.NoMergePolicy;
 import org.apache.lucene.index.NumericDocValues;
 import org.apache.lucene.index.PointValues;
+import org.apache.lucene.index.SegmentInfos;
 import org.apache.lucene.index.Term;
 import org.apache.lucene.index.TieredMergePolicy;
 import org.apache.lucene.search.IndexSearcher;
@@ -163,6 +165,7 @@
 import static org.elasticsearch.index.engine.Engine.Operation.Origin.REPLICA;
 import static org.elasticsearch.index.translog.TranslogDeletionPolicies.createTranslogDeletionPolicy;
 import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.sameInstance;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
@@ -4439,4 +4442,37 @@ public void testCleanUpCommitsWhenGlobalCheckpointAdvanced() throws Exception {
             assertThat(DirectoryReader.listCommits(store.directory()), contains(commits.get(commits.size() - 1)));
         }
     }
+
+    public void testShouldFlushToFreeTranslog() throws Exception {
+        assertThat("Empty engine does not need flushing", engine.shouldFlushToFreeTranslog(), equalTo(false));
+        int numDocs = between(10, 100);
+        for (int id = 0; id < numDocs; id++) {
+            final ParsedDocument doc = testParsedDocument(Integer.toString(id), null, testDocumentWithTextField(), SOURCE, null);
+            engine.index(indexForDoc(doc));
+        }
+        assertThat("Not exceeded translog flush threshold yet", engine.shouldFlushToFreeTranslog(), equalTo(false));
+        long flushThreshold = RandomNumbers.randomLongBetween(random(), 100, engine.getTranslog().uncommittedSizeInBytes());
+        final IndexSettings indexSettings = engine.config().getIndexSettings();
+        final IndexMetaData indexMetaData = IndexMetaData.builder(indexSettings.getIndexMetaData())
+            .settings(Settings.builder().put(indexSettings.getSettings())
+                .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), flushThreshold + "b")).build();
+        indexSettings.updateIndexMetaData(indexMetaData);
+        engine.onSettingsChanged();
+        assertThat(engine.getTranslog().uncommittedOperations(), equalTo(numDocs));
+        assertThat(engine.shouldFlushToFreeTranslog(), equalTo(true));
+        engine.flush();
+        assertThat(engine.getTranslog().uncommittedOperations(), equalTo(0));
+        // Stale operations skipped by Lucene but added to translog - still able to flush
+        for (int id = 0; id < numDocs; id++) {
+            final ParsedDocument doc = testParsedDocument(Integer.toString(id), null, testDocumentWithTextField(), SOURCE, null);
+            final Engine.IndexResult result = engine.index(replicaIndexForDoc(doc, 1L, id, false));
+            assertThat(result.isCreated(), equalTo(false));
+        }
+        SegmentInfos lastCommitInfo = engine.getLastCommittedSegmentInfos();
+        assertThat(engine.getTranslog().uncommittedOperations(), equalTo(numDocs));
+        assertThat(engine.shouldFlushToFreeTranslog(), equalTo(true));
+        engine.flush(false, false);
+        assertThat(engine.getLastCommittedSegmentInfos(), not(sameInstance(lastCommitInfo)));
+        assertThat(engine.getTranslog().uncommittedOperations(), equalTo(0));
+    }
 }

server/src/test/java/org/elasticsearch/indices/recovery/RecoveryTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.indices.recovery;
 
+import com.carrotsearch.randomizedtesting.generators.RandomNumbers;
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.IndexCommit;
 import org.apache.lucene.index.IndexWriter;
@@ -306,4 +307,26 @@ public void testSequenceBasedRecoveryKeepsTranslog() throws Exception {
         }
     }
 
+    public void testShouldFlushAfterPeerRecovery() throws Exception {
+        try (ReplicationGroup shards = createGroup(0)) {
+            shards.startAll();
+            long translogSizeOnPrimary = 0;
+            int numDocs = shards.indexDocs(between(10, 100));
+            translogSizeOnPrimary += shards.getPrimary().getTranslog().uncommittedSizeInBytes();
+            shards.flush();
+
+            final IndexShard replica = shards.addReplica();
+            IndexMetaData.Builder builder = IndexMetaData.builder(replica.indexSettings().getIndexMetaData());
+            long flushThreshold = RandomNumbers.randomLongBetween(random(), 100, translogSizeOnPrimary);
+            builder.settings(Settings.builder().put(replica.indexSettings().getSettings())
+                .put(IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING.getKey(), flushThreshold + "b")
+            );
+            replica.indexSettings().updateIndexMetaData(builder.build());
+            replica.onSettingsChanged();
+            shards.recoverReplica(replica);
+            assertBusy(() -> assertThat(getEngine(replica).shouldFlushToFreeTranslog(), equalTo(false)));
+            assertThat(replica.getTranslog().totalOperations(), equalTo(numDocs));
+            shards.assertAllEqual(numDocs);
+        }
+    }
 }