Use IndexWriter.getFlushingBytes() rather than tracking it ourselves (#33582)

Currently we keep track of how many bytes are currently being written to disk
in an AtomicLong within InternalEngine, updating it on refresh. The IndexWriter
has its own accounting for this, and exposes it via a getFlushingBytes method
in the latest lucene 8 snapshot. This commit removes the InternalEngine tracking
in favour of just using the IndexWriter method.

Author: romseygeek

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

@@ -156,12 +156,6 @@ public class InternalEngine extends Engine {
     private final SoftDeletesPolicy softDeletesPolicy;
     private final LastRefreshedCheckpointListener lastRefreshedCheckpointListener;
 
-    /**
-     * How many bytes we are currently moving to disk, via either IndexWriter.flush or refresh.  IndexingMemoryController polls this
-     * across all shards to decide if throttling is necessary because moving bytes to disk is falling behind vs incoming documents
-     * being indexed/deleted.
-     */
-    private final AtomicLong writingBytes = new AtomicLong();
     private final AtomicBoolean trackTranslogLocation = new AtomicBoolean(false);
 
     @Nullable
@@ -543,7 +537,7 @@ public String getHistoryUUID() {
     /** Returns how many bytes we are currently moving from indexing buffer to segments on disk */
     @Override
     public long getWritingBytes() {
-        return writingBytes.get();
+        return indexWriter.getFlushingBytes() + versionMap.getRefreshingBytes();
     }
 
     /**
@@ -1517,9 +1511,6 @@ final void refresh(String source, SearcherScope scope) throws EngineException {
         // pass the new reader reference to the external reader manager.
         final long localCheckpointBeforeRefresh = getLocalCheckpoint();
 
-        // this will also cause version map ram to be freed hence we always account for it.
-        final long bytes = indexWriter.ramBytesUsed() + versionMap.ramBytesUsedForRefresh();
-        writingBytes.addAndGet(bytes);
         try (ReleasableLock lock = readLock.acquire()) {
             ensureOpen();
             if (store.tryIncRef()) {
@@ -1545,8 +1536,6 @@ final void refresh(String source, SearcherScope scope) throws EngineException {
                 e.addSuppressed(inner);
             }
             throw new RefreshFailedEngineException(shardId, e);
-        }  finally {
-            writingBytes.addAndGet(-bytes);
         }
         assert lastRefreshedCheckpoint() >= localCheckpointBeforeRefresh : "refresh checkpoint was not advanced; " +
             "local_checkpoint=" + localCheckpointBeforeRefresh + " refresh_checkpoint=" + lastRefreshedCheckpoint();

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

@@ -434,6 +434,14 @@ long ramBytesUsedForRefresh() {
         return maps.current.ramBytesUsed.get();
     }
 
+    /**
+     * Returns how much RAM is current being freed up by refreshing.  This is {@link #ramBytesUsed()}
+     * except does not include tombstones because they don't clear on refresh.
+     */
+    long getRefreshingBytes() {
+        return maps.old.ramBytesUsed.get();
+    }
+
     @Override
     public Collection<Accountable> getChildResources() {
         // TODO: useful to break down RAM usage here?

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

@@ -41,6 +41,7 @@
 
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.nullValue;
 
 public class LiveVersionMapTests extends ESTestCase {
@@ -91,6 +92,19 @@ public void testRamBytesUsed() throws Exception {
         assertEquals(actualRamBytesUsed, estimatedRamBytesUsed, tolerance);
     }
 
+    public void testRefreshingBytes() throws IOException {
+        LiveVersionMap map = new LiveVersionMap();
+        BytesRefBuilder uid = new BytesRefBuilder();
+        uid.copyChars(TestUtil.randomSimpleString(random(), 10, 20));
+        try (Releasable r = map.acquireLock(uid.toBytesRef())) {
+            map.putIndexUnderLock(uid.toBytesRef(), randomIndexVersionValue());
+        }
+        map.beforeRefresh();
+        assertThat(map.getRefreshingBytes(), greaterThan(0L));
+        map.afterRefresh(true);
+        assertThat(map.getRefreshingBytes(), equalTo(0L));
+    }
+
     private BytesRef uid(String string) {
         BytesRefBuilder builder = new BytesRefBuilder();
         builder.copyChars(string);