Clarify flush result when skipped due to collision

The unclear boolean flag before was easy to misinterpret. This
PR clarifies the boolean flag better, to understand that it
details to whether a flush request was skipped due to collisions
or not (rather than the flush request being performed or not).

Author: kingherc

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

@@ -36,6 +36,7 @@
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.SubscribableListener;
@@ -2490,10 +2491,21 @@ public RefreshResult(boolean refreshed) {
         }
     }
 
-    public record FlushResult(boolean flushPerformed, long generation) {
+    /**
+     * The result of a {@link FlushRequest}.
+     *
+     * @param skippedDueToCollision signifies whether the flush request was skipped due to a collision detected. Specifically it is
+     *                              <code>true</code> if <code>waitIfOngoing==false</code> and an ongoing request is detected, which means
+     *                              the flush request was skipped, else it is <code>false</code>, which means there was no collision and
+     *                              the flush request was processed (even if in the end it did not actually flush anything).
+     * @param generation            the generation of the index commit of the flush. may be {@link FlushResult#UNKNOWN_GENERATION} if
+     *                              unknown, e.g., if the flush was skipped or not performed ultimately.
+     */
+    public record FlushResult(boolean skippedDueToCollision, long generation) {
 
         public static final long UNKNOWN_GENERATION = -1L;
-        public static final FlushResult NO_FLUSH = new FlushResult(false, UNKNOWN_GENERATION);
+        public static final FlushResult FLUSH_REQUEST_SKIPPED_DUE_TO_COLLISION = new FlushResult(true, UNKNOWN_GENERATION);
+        public static final FlushResult FLUSH_REQUEST_PROCESSED_AND_NOT_PERFORMED = new FlushResult(false, UNKNOWN_GENERATION);
     }
 
     /**

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

@@ -2260,7 +2260,7 @@ protected void flushHoldingLock(boolean force, boolean waitIfOngoing, ActionList
                 // if we can't get the lock right away we block if needed otherwise barf
                 if (waitIfOngoing == false) {
                     logger.trace("detected an in-flight flush, not blocking to wait for it's completion");
-                    listener.onResponse(FlushResult.NO_FLUSH);
+                    listener.onResponse(FlushResult.FLUSH_REQUEST_SKIPPED_DUE_TO_COLLISION);
                     return;
                 }
                 logger.trace("waiting for in-flight flush to finish");
@@ -2347,7 +2347,7 @@ protected void flushHoldingLock(boolean force, boolean waitIfOngoing, ActionList
             pruneDeletedTombstones();
         }
 
-        waitForCommitDurability(generation, listener.map(v -> new FlushResult(true, generation)));
+        waitForCommitDurability(generation, listener.map(v -> new FlushResult(false, generation)));
     }
 
     protected final boolean isFlushLockIsHeldByCurrentThread() {

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

@@ -452,7 +452,7 @@ public boolean shouldPeriodicallyFlush() {
 
     @Override
     protected void flushHoldingLock(boolean force, boolean waitIfOngoing, ActionListener<FlushResult> listener) throws EngineException {
-        listener.onResponse(new FlushResult(true, lastCommittedSegmentInfos.getGeneration()));
+        listener.onResponse(new FlushResult(false, lastCommittedSegmentInfos.getGeneration()));
     }
 
     @Override

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

@@ -1503,11 +1503,10 @@ public BulkStats bulkStats() {
      * Executes the given flush request against the engine.
      *
      * @param request the flush request
-     * @return <code>false</code> if <code>waitIfOngoing==false</code> and an ongoing request is detected, else <code>true</code>.
-     *         If <code>false</code> is returned, no flush happened.
+     * @return the flush result
      */
-    public boolean flush(FlushRequest request) {
-        PlainActionFuture<Boolean> future = new PlainActionFuture<>();
+    public Engine.FlushResult flush(FlushRequest request) {
+        PlainActionFuture<Engine.FlushResult> future = new PlainActionFuture<>();
         flush(request, future);
         return future.actionGet();
     }
@@ -1516,12 +1515,9 @@ public boolean flush(FlushRequest request) {
      * Executes the given flush request against the engine.
      *
      * @param request the flush request
-     * @param listener to notify after full durability has been achieved.
-     *                 <code>false</code> if <code>waitIfOngoing==false</code>
-     *                 and an ongoing request is detected, else <code>true</code>.
-     *                 If <code>false</code> is returned, no flush happened.
+     * @param listener to notify after full durability has been achieved
      */
-    public void flush(FlushRequest request, ActionListener<Boolean> listener) {
+    public void flush(FlushRequest request, ActionListener<Engine.FlushResult> listener) {
         final boolean waitIfOngoing = request.waitIfOngoing();
         final boolean force = request.force();
         logger.trace("flush with {}", request);
@@ -1533,11 +1529,7 @@ public void flush(FlushRequest request, ActionListener<Boolean> listener) {
              */
             verifyNotClosed();
             final long startTime = System.nanoTime();
-            getEngine().flush(
-                force,
-                waitIfOngoing,
-                ActionListener.runBefore(l.map(Engine.FlushResult::flushPerformed), () -> flushMetric.inc(System.nanoTime() - startTime))
-            );
+            getEngine().flush(force, waitIfOngoing, ActionListener.runBefore(l, () -> flushMetric.inc(System.nanoTime() - startTime)));
         });
     }
 
@@ -2541,8 +2533,8 @@ public void flushOnIdle(long inactiveTimeNS) {
                 threadPool.executor(ThreadPool.Names.FLUSH)
                     .execute(() -> flush(new FlushRequest().waitIfOngoing(false).force(false), new ActionListener<>() {
                         @Override
-                        public void onResponse(Boolean flushed) {
-                            if (flushed == false) {
+                        public void onResponse(Engine.FlushResult flushResult) {
+                            if (flushResult.skippedDueToCollision()) {
                                 // In case an ongoing flush was detected, revert active flag so that a next flushOnIdle request
                                 // will retry (#87888)
                                 active.set(true);
@@ -4151,7 +4143,7 @@ public void afterWriteOperation() {
                     threadPool.executor(ThreadPool.Names.FLUSH).execute(() -> {
                         flush(new FlushRequest(), new ActionListener<>() {
                             @Override
-                            public void onResponse(Boolean flushed) {
+                            public void onResponse(Engine.FlushResult flushResult) {
                                 periodicFlushMetric.inc();
                             }
 

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

@@ -36,6 +36,7 @@
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ReplicationGroup;
 import org.elasticsearch.index.shard.ShardId;
@@ -190,7 +191,7 @@ private void executeOnPrimaryOrReplica(IndexShard indexShard, boolean phase1) th
 
     public void testShardIsFlushed() throws Throwable {
         final ArgumentCaptor<FlushRequest> flushRequest = ArgumentCaptor.forClass(FlushRequest.class);
-        doReturn(true).when(indexShard).flush(flushRequest.capture());
+        doReturn(Engine.FlushResult.FLUSH_REQUEST_PROCESSED_AND_NOT_PERFORMED).when(indexShard).flush(flushRequest.capture());
         executeOnPrimaryOrReplica();
         verify(indexShard, times(1)).flush(any(FlushRequest.class));
         assertThat(flushRequest.getValue().force(), is(true));

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -4342,8 +4342,8 @@ protected void commitIndexWriter(final IndexWriter writer, final Translog transl
             shard.flushOnIdle(0);
             mockLog.awaitAllExpectationsMatched();
 
-            // A direct call to flush (with waitIfOngoing=false) should not wait and return false immediately
-            assertFalse(shard.flush(new FlushRequest().waitIfOngoing(false).force(false)));
+            // A direct call to flush (with waitIfOngoing=false) should not wait and return immediately
+            assertTrue(shard.flush(new FlushRequest().waitIfOngoing(false).force(false)).skippedDueToCollision());
 
             // Allow first flushOnIdle to complete
             readyToCompleteFlushLatch.countDown();
@@ -4362,8 +4362,8 @@ protected void commitIndexWriter(final IndexWriter writer, final Translog transl
             // The second flushOnIdle (that did not happen) should have turned the active flag to true
             assertTrue(shard.isActive());
 
-            // After all the previous flushes are done, issue a final flush (for any remaining documents) that should return true
-            assertTrue(shard.flush(new FlushRequest()));
+            // After all the previous flushes are done, issue a final flush (for any remaining documents) that should be processed
+            assertFalse(shard.flush(new FlushRequest()).skippedDueToCollision());
 
             closeShards(shard);
         }
@@ -4463,7 +4463,7 @@ protected void flushHoldingLock(boolean force, boolean waitIfOngoing, ActionList
         assertThat(pendingListeners.size(), is(numberOfFlushes));
         assertThat(shard.flushStats().getPeriodic(), is(equalTo(0L)));
 
-        pendingListeners.forEach(l -> l.onResponse(new Engine.FlushResult(true, 1)));
+        pendingListeners.forEach(l -> l.onResponse(new Engine.FlushResult(false, 1)));
         assertThat(shard.flushStats().getPeriodic(), is(equalTo((long) numberOfFlushes)));
 
         closeShards(shard);

server/src/test/java/org/elasticsearch/index/shard/ShardGetServiceTests.java

@@ -344,7 +344,7 @@ public void testGetFromTranslog() throws IOException {
         PlainActionFuture<Engine.FlushResult> flushFuture = new PlainActionFuture<>();
         engine.flush(true, true, flushFuture);
         var flushResult = flushFuture.actionGet();
-        assertTrue(flushResult.flushPerformed());
+        assertFalse(flushResult.skippedDueToCollision());
         assertThat(flushResult.generation(), equalTo(lastUnsafeGeneration + 1));
         assertThat(engine.getLastUnsafeSegmentGenerationForGets(), equalTo(lastUnsafeGeneration));
         // No longer in translog