Revert changes not related to metrics

Author: kingherc

server/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java

@@ -45,16 +45,16 @@ public FlushRequest(StreamInput in) throws IOException {
     }
 
     /**
-     * Returns {@code true} iff a flush should block (e.g., if the engine is being reset or
-     * if a another flush operation is already running). Otherwise {@code false}
+     * Returns {@code true} iff a flush should block
+     * if a another flush operation is already running. Otherwise {@code false}
      */
     public boolean waitIfOngoing() {
         return this.waitIfOngoing;
     }
 
     /**
-     * if set to {@code true} the flush will block (e.g., if the engine is being reset or
-     * if a another flush operation is already running) until the flush can be performed.
+     * if set to {@code true} the flush will block
+     * if a another flush operation is already running until the flush can be performed.
      * The default is <code>true</code>
      */
     public FlushRequest waitIfOngoing(boolean waitIfOngoing) {

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

@@ -1352,7 +1352,7 @@ public void externalRefresh(String source, ActionListener<Engine.RefreshResult>
      * Returns how many bytes we are currently moving from heap to disk
      */
     public long getWritingBytes() {
-        return withEngineOrNullIfBeingReset(engine -> {
+        return tryWithEngineOrNull(engine -> {
             if (engine == null) {
                 return 0L;
             }
@@ -1372,7 +1372,7 @@ public RefreshStats refreshStats() {
     }
 
     public FlushStats flushStats() {
-        return withEngineOrNullIfBeingReset(
+        return tryWithEngineOrNull(
             engine -> new FlushStats(
                 flushMetric.count(),
                 periodicFlushMetric.count(),
@@ -1404,7 +1404,7 @@ public SeqNoStats seqNoStats() {
     }
 
     public IndexingStats indexingStats() {
-        return withEngineOrNullIfBeingReset(engine -> {
+        return tryWithEngineOrNull(engine -> {
             final boolean throttled;
             final long throttleTimeInMillis;
             if (engine == null) {
@@ -1528,7 +1528,7 @@ public boolean flush(FlushRequest request) {
      * @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 or engine reset is detected, else <code>true</code>.
+     *                 and an ongoing request is detected, else <code>true</code>.
      *                 If <code>false</code> is returned, no flush happened.
      */
     public void flush(FlushRequest request, ActionListener<Boolean> listener) {
@@ -1543,23 +1543,11 @@ public void flush(FlushRequest request, ActionListener<Boolean> listener) {
              */
             verifyNotClosed();
             final long startTime = System.nanoTime();
-            final var finalListener = ActionListener.runBefore(
-                l.map(Engine.FlushResult::flushPerformed),
-                () -> flushMetric.inc(System.nanoTime() - startTime)
+            getEngine().flush(
+                force,
+                waitIfOngoing,
+                ActionListener.runBefore(l.map(Engine.FlushResult::flushPerformed), () -> flushMetric.inc(System.nanoTime() - startTime))
             );
-            if (waitIfOngoing) {
-                getEngine().flush(force, waitIfOngoing, finalListener);
-            } else {
-                withEngineOrNullIfBeingReset(engine -> {
-                    if (engine == null) {
-                        logger.trace("detected engine is closed or being reset, not blocking");
-                        finalListener.onResponse(Engine.FlushResult.NO_FLUSH);
-                    } else {
-                        engine.flush(force, waitIfOngoing, finalListener);
-                    }
-                    return null;
-                });
-            }
         });
     }
 
@@ -1590,12 +1578,8 @@ public long readGlobalCheckpointForRecovery(Map<String, String> commitUserData)
      */
     public void trimTranslog() {
         verifyNotClosed();
-        withEngineOrNullIfBeingReset(engine -> {
-            if (engine != null) {
-                engine.trimUnreferencedTranslogFiles();
-            }
-            return null;
-        });
+        final Engine engine = getEngine();
+        engine.trimUnreferencedTranslogFiles();
     }
 
     /**
@@ -2539,7 +2523,7 @@ private void verifyNotClosed(Exception suppressed) throws IllegalIndexShardState
      * Returns number of heap bytes used by the indexing buffer for this shard, or 0 if the shard is closed
      */
     public long getIndexBufferRAMBytesUsed() {
-        return withEngineOrNullIfBeingReset(engine -> {
+        return tryWithEngineOrNull(engine -> {
             if (engine == null) {
                 return 0L;
             }
@@ -2560,37 +2544,33 @@ public void addShardFailureCallback(Consumer<ShardFailure> onShardFailure) {
      * indexing operation, so we can flush the index.
      */
     public void flushOnIdle(long inactiveTimeNS) {
-        withEngineOrNullIfBeingReset(engineOrNull -> {
-            if (engineOrNull != null && System.nanoTime() - engineOrNull.getLastWriteNanos() >= inactiveTimeNS) {
-                boolean wasActive = active.getAndSet(false);
-                if (wasActive) {
-                    logger.debug("flushing shard on inactive");
-                    threadPool.executor(ThreadPool.Names.FLUSH)
-                        .execute(() -> flush(new FlushRequest().waitIfOngoing(false).force(false), new ActionListener<>() {
-                            @Override
-                            public void onResponse(Boolean flushed) {
-                                if (flushed == false) {
-                                    // In case an ongoing flush was detected, revert active flag so that a next flushOnIdle request
-                                    // will retry (#87888)
-                                    active.set(true);
-                                }
-                                periodicFlushMetric.inc();
+        Engine engineOrNull = getEngineOrNull();
+        if (engineOrNull != null && System.nanoTime() - engineOrNull.getLastWriteNanos() >= inactiveTimeNS) {
+            boolean wasActive = active.getAndSet(false);
+            if (wasActive) {
+                logger.debug("flushing shard on inactive");
+                threadPool.executor(ThreadPool.Names.FLUSH)
+                    .execute(() -> flush(new FlushRequest().waitIfOngoing(false).force(false), new ActionListener<>() {
+                        @Override
+                        public void onResponse(Boolean flushed) {
+                            if (flushed == false) {
+                                // In case an ongoing flush was detected, revert active flag so that a next flushOnIdle request
+                                // will retry (#87888)
+                                active.set(true);
                             }
+                            periodicFlushMetric.inc();
+                        }
 
-                            @Override
-                            public void onFailure(Exception e) {
-                                if (state != IndexShardState.CLOSED) {
-                                    active.set(true);
-                                    logger.warn("failed to flush shard on inactive", e);
-                                }
+                        @Override
+                        public void onFailure(Exception e) {
+                            if (state != IndexShardState.CLOSED) {
+                                active.set(true);
+                                logger.warn("failed to flush shard on inactive", e);
                             }
-                        }));
-                }
-            } else {
-                logger.trace(() -> "flush on idle skipped as it is either closed, engine being reset, or has been recently written to");
+                        }
+                    }));
             }
-            return null;
-        });
+        }
     }
 
     public boolean isActive() {
@@ -2781,21 +2761,19 @@ public IndexEventListener getIndexEventListener() {
      * setting is set to true, throttling will pause indexing completely. Otherwise, indexing will be throttled to one thread.
      */
     public void activateThrottling() {
-        withEngineOrNullIfBeingReset(engine -> {
-            if (engine != null) {
-                engine.activateThrottling();
-            }
-            return null;
-        });
+        try {
+            getEngine().activateThrottling();
+        } catch (AlreadyClosedException ex) {
+            // ignore
+        }
     }
 
     public void deactivateThrottling() {
-        withEngineOrNullIfBeingReset(engine -> {
-            if (engine != null) {
-                engine.deactivateThrottling();
-            }
-            return null;
-        });
+        try {
+            getEngine().deactivateThrottling();
+        } catch (AlreadyClosedException ex) {
+            // ignore
+        }
     }
 
     private void suspendThrottling() {
@@ -2840,16 +2818,12 @@ private void handleRefreshException(Exception e) {
      * Called when our shard is using too much heap and should move buffered indexed/deleted documents to disk.
      */
     public void writeIndexingBuffer() {
-        withEngineOrNullIfBeingReset(engine -> {
-            if (engine != null) {
-                try {
-                    engine.writeIndexingBuffer();
-                } catch (Exception e) {
-                    handleRefreshException(e);
-                }
-            }
-            return null;
-        });
+        try {
+            Engine engine = getEngine();
+            engine.writeIndexingBuffer();
+        } catch (Exception e) {
+            handleRefreshException(e);
+        }
     }
 
     /**
@@ -3436,7 +3410,7 @@ Engine getEngine() {
      * NOTE: returns null if engine is not yet started (e.g. recovery phase 1, copying over index files, is still running), or if engine is
      * closed.
      *
-     * @deprecated use the withEngine* family of functions instead, which consider the situation an engine may be reset / changed.
+     * @deprecated use the try/withEngine* family of functions instead, which consider the situation an engine may be reset / changed.
      */
     @Deprecated
     public Engine getEngineOrNull() {
@@ -3471,7 +3445,7 @@ private Engine getAndSetCurrentEngine(Engine newEngine) {
      * @return              the result of the operation
      * @param <R>           the type of the result
      */
-    public <R> R withEngineOrNullIfBeingReset(Function<Engine, R> operation) {
+    public <R> R tryWithEngineOrNull(Function<Engine, R> operation) {
         return withEngine(operation, true, true);
     }
 

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

@@ -4369,80 +4369,6 @@ protected void commitIndexWriter(final IndexWriter writer, final Translog transl
         }
     }
 
-    @TestLogging(reason = "testing traces of concurrent flush and engine reset", value = "org.elasticsearch.index.shard.IndexShard:TRACE")
-    public void testFlushOnIdleDoesNotWaitWhileEngineIsReset() throws Exception {
-        final var preparedForReset = new AtomicBoolean();
-        final var shard = newStartedShard(true, Settings.EMPTY, config -> {
-            if (preparedForReset.get()) {
-                return new ReadOnlyEngine(config, null, new TranslogStats(), false, Function.identity(), true, true);
-            } else {
-                return new InternalEngine(config) {
-                    @Override
-                    public void prepareForEngineReset() throws IOException {
-                        assertTrue(preparedForReset.compareAndSet(false, true));
-                    }
-                };
-            }
-        });
-        final var engineResetLock = shard.getEngine().getEngineConfig().getEngineResetLock();
-
-        final var release = new CountDownLatch(1);
-        final var reset = new PlainActionFuture<Void>();
-        final var resetEngineThread = new Thread(() -> {
-            try {
-                shard.acquirePrimaryOperationPermit(reset.delegateFailure((l, permit) -> {
-                    try (permit) {
-                        shard.resetEngine(newEngine -> {
-                            assertThat(engineResetLock.isWriteLockedByCurrentThread(), equalTo(true));
-                            assertThat(newEngine, instanceOf(ReadOnlyEngine.class));
-                            safeAwait(release);
-                        });
-                        assertThat(preparedForReset.get(), equalTo(true));
-                        l.onResponse(null);
-                    }
-                }), EsExecutors.DIRECT_EXECUTOR_SERVICE);
-            } catch (Exception e) {
-                reset.onFailure(e);
-            }
-        });
-        resetEngineThread.start();
-
-        assertBusy(() -> assertThat(engineResetLock.isWriteLocked(), equalTo(true)));
-
-        try (var mockLog = MockLog.capture(IndexShard.class)) {
-            mockLog.addExpectation(
-                new MockLog.SeenEventExpectation(
-                    "should see flush on idle returning since it will not wait for the engine reset",
-                    IndexShard.class.getCanonicalName(),
-                    Level.TRACE,
-                    "flush on idle skipped"
-                )
-            );
-            shard.flushOnIdle(0);
-            mockLog.awaitAllExpectationsMatched();
-        }
-
-        try (var mockLog = MockLog.capture(IndexShard.class)) {
-            mockLog.addExpectation(
-                new MockLog.SeenEventExpectation(
-                    "should see flush returning since it will not wait for the engine reset",
-                    IndexShard.class.getCanonicalName(),
-                    Level.TRACE,
-                    "detected engine is closed or being reset, not blocking"
-                )
-            );
-            // A direct call to flush (with waitIfOngoing=false) should not wait and return false immediately
-            assertFalse(shard.flush(new FlushRequest().waitIfOngoing(false).force(false)));
-            mockLog.awaitAllExpectationsMatched();
-        }
-
-        release.countDown();
-        safeGet(reset);
-        assertThat(engineResetLock.isWriteLocked(), equalTo(false));
-        resetEngineThread.join();
-        closeShards(shard);
-    }
-
     public void testFlushOnIdleAfterOp() throws Exception {
         // Holding the write lock makes the index/delete op to halt before being processed by the engine
         final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
@@ -5340,7 +5266,7 @@ public void prepareForEngineReset() throws IOException {
         closeShards(shard);
     }
 
-    public void testWithEngineOrNullIfBeingReset() throws Exception {
+    public void testTryWithEngineOrNull() throws Exception {
         final var preparedForReset = new AtomicBoolean();
         final var shard = newStartedShard(true, Settings.EMPTY, config -> {
             if (preparedForReset.get()) {
@@ -5379,7 +5305,7 @@ public void prepareForEngineReset() throws IOException {
 
         assertBusy(() -> assertThat(engineResetLock.isWriteLocked(), equalTo(true)));
 
-        shard.withEngineOrNullIfBeingReset(engine -> {
+        shard.tryWithEngineOrNull(engine -> {
             assertNull(engine);
             assertThat(engineResetLock.isReadLocked(), equalTo(false));
             return null;
@@ -5389,7 +5315,7 @@ public void prepareForEngineReset() throws IOException {
         safeGet(reset);
         assertThat(engineResetLock.isWriteLocked(), equalTo(false));
 
-        shard.withEngineOrNullIfBeingReset(engine -> {
+        shard.tryWithEngineOrNull(engine -> {
             assertThat(engine, instanceOf(ReadOnlyEngine.class));
             assertThat(engineResetLock.isReadLocked(), equalTo(true));
             return null;