Inactive shard flush should wait for ongoing one

server/src/internalClusterTest/java/org/elasticsearch/action/admin/indices/diskusage/IndexDiskUsageAnalyzerIT.java

@@ -66,12 +66,12 @@ public static class EngineTestPlugin extends Plugin implements EnginePlugin {
         public Optional<EngineFactory> getEngineFactory(IndexSettings indexSettings) {
             return Optional.of(config -> new InternalEngine(config) {
                 @Override
-                public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
+                public boolean flush(boolean force, boolean waitIfOngoing) throws EngineException {
                     final ShardId shardId = config.getShardId();
                     if (failOnFlushShards.contains(shardId)) {
                         throw new EngineException(shardId, "simulated IO");
                     }
-                    super.flush(force, waitIfOngoing);
+                    return super.flush(force, waitIfOngoing);
                 }
             });
         }

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

@@ -1015,8 +1015,9 @@ public boolean refreshNeeded() {
      * @param force         if <code>true</code> a lucene commit is executed even if no changes need to be committed.
      * @param waitIfOngoing if <code>true</code> this call will block until all currently running flushes have finished.
      *                      Otherwise this call will return without blocking.
+     * @return <code>true</code> if the flush happened, else <code>false</code> (e.g., if it did not wait for an ongoing request)
      */
-    public abstract void flush(boolean force, boolean waitIfOngoing) throws EngineException;
+    public abstract boolean flush(boolean force, boolean waitIfOngoing) throws EngineException;
 
     /**
      * Flushes the state of the engine including the transaction log, clearing memory and persisting

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

@@ -1895,7 +1895,7 @@ public boolean shouldPeriodicallyFlush() {
     }
 
     @Override
-    public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
+    public boolean flush(boolean force, boolean waitIfOngoing) throws EngineException {
         ensureOpen();
         if (force && waitIfOngoing == false) {
             assert false : "wait_if_ongoing must be true for a force flush: force=" + force + " wait_if_ongoing=" + waitIfOngoing;
@@ -1908,7 +1908,7 @@ public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
             if (flushLock.tryLock() == false) {
                 // if we can't get the lock right away we block if needed otherwise barf
                 if (waitIfOngoing == false) {
-                    return;
+                    return false;
                 }
                 logger.trace("waiting for in-flight flush to finish");
                 flushLock.lock();
@@ -1967,6 +1967,7 @@ public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
         if (engineConfig.isEnableGcDeletes()) {
             pruneDeletedTombstones();
         }
+        return true;
     }
 
     private void refreshLastCommittedSegmentInfos() {

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

@@ -437,8 +437,8 @@ public boolean shouldPeriodicallyFlush() {
     }
 
     @Override
-    public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
-        // noop
+    public boolean flush(boolean force, boolean waitIfOngoing) throws EngineException {
+        return true; // noop
     }
 
     @Override

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

@@ -1343,8 +1343,9 @@ public BulkStats bulkStats() {
      * Executes the given flush request against the engine.
      *
      * @param request the flush request
+     * @return <code>true</code> if the flush happened, else <code>false</code> (e.g., if it did not wait for an ongoing request)
      */
-    public void flush(FlushRequest request) {
+    public boolean flush(FlushRequest request) {
         final boolean waitIfOngoing = request.waitIfOngoing();
         final boolean force = request.force();
         logger.trace("flush with {}", request);
@@ -1355,8 +1356,11 @@ public void flush(FlushRequest request) {
          */
         verifyNotClosed();
         final long time = System.nanoTime();
-        getEngine().flush(force, waitIfOngoing);
-        flushMetric.inc(System.nanoTime() - time);
+        boolean flushHappened = getEngine().flush(force, waitIfOngoing);
+        if (flushHappened) {
+            flushMetric.inc(System.nanoTime() - time);
+        }
+        return flushHappened;
     }
 
     /**
@@ -2188,8 +2192,12 @@ public void onFailure(Exception e) {
 
                     @Override
                     protected void doRun() {
-                        flush(new FlushRequest().waitIfOngoing(false).force(false));
-                        periodicFlushMetric.inc();
+                        if (flush(new FlushRequest().waitIfOngoing(false).force(false))) {
+                            periodicFlushMetric.inc();
+                        } else {
+                            // In case the flush did not happen, revert active flag so that a next flushOnIdle request can happen (#87888)
+                            active.set(true);
+                        }
                     }
                 });
             }

server/src/test/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseActionTests.java

@@ -62,7 +62,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -170,7 +170,7 @@ private void executeOnPrimaryOrReplica(boolean phase1) throws Throwable {
 
     public void testShardIsFlushed() throws Throwable {
         final ArgumentCaptor<FlushRequest> flushRequest = ArgumentCaptor.forClass(FlushRequest.class);
-        doNothing().when(indexShard).flush(flushRequest.capture());
+        doReturn(true).when(indexShard).flush(flushRequest.capture());
         executeOnPrimaryOrReplica();
         verify(indexShard, times(1)).flush(any(FlushRequest.class));
         assertThat(flushRequest.getValue().force(), is(true));