Use _refresh to shrink the version map on inactivity (#27918)

We used to shrink the version map under an external lock. This is
quite ambigious and instead we can simply issue an empty refresh to
shrink it.

Closes #27852

Author: s1monw

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

@@ -981,6 +981,9 @@ public SyncedFlushResult syncFlush(String syncId, CommitId expectedCommitId) thr
         try (ReleasableLock lock = writeLock.acquire()) {
             ensureOpen();
             ensureCanFlush();
+            // lets do a refresh to make sure we shrink the version map. This refresh will be either a no-op (just shrink the version map)
+            // or we also have uncommitted changes and that causes this syncFlush to fail.
+            refresh("sync_flush");
             if (indexWriter.hasUncommittedChanges()) {
                 logger.trace("can't sync commit [{}]. have pending changes", syncId);
                 return SyncedFlushResult.PENDING_OPERATIONS;
@@ -993,8 +996,6 @@ public SyncedFlushResult syncFlush(String syncId, CommitId expectedCommitId) thr
             commitIndexWriter(indexWriter, translog, syncId);
             logger.debug("successfully sync committed. sync id [{}].", syncId);
             lastCommittedSegmentInfos = store.readLastCommittedSegmentsInfo();
-            // we are guaranteed to have no operations in the version map here!
-            versionMap.adjustMapSizeUnderLock();
             return SyncedFlushResult.SUCCESS;
         } catch (IOException ex) {
             maybeFailEngine("sync commit", ex);

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

@@ -34,18 +34,6 @@
 /** Maps _uid value to its version information. */
 class LiveVersionMap implements ReferenceManager.RefreshListener, Accountable {
 
-    /**
-     * Resets the internal map and adjusts it's capacity as if there were no indexing operations.
-     * This must be called under write lock in the engine
-     */
-    void adjustMapSizeUnderLock() {
-        if (maps.current.isEmpty() == false || maps.old.isEmpty() == false) {
-            assert false : "map must be empty"; // fail hard if not empty and fail with assertion in tests to ensure we never swallow it
-            throw new IllegalStateException("map must be empty");
-        }
-        maps = new Maps();
-    }
-
     private static class Maps {
 
         // All writes (adds and deletes) go into here:

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

@@ -23,7 +23,6 @@
 import org.apache.lucene.util.BytesRefBuilder;
 import org.apache.lucene.util.RamUsageTester;
 import org.apache.lucene.util.TestUtil;
-import org.elasticsearch.Assertions;
 import org.elasticsearch.bootstrap.JavaVersion;
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.util.concurrent.KeyedLock;
@@ -97,32 +96,6 @@ public void testBasics() throws IOException {
         assertNull(map.getUnderLock(uid("test")));
     }
 
-
-    public void testAdjustMapSizeUnderLock() throws IOException {
-        LiveVersionMap map = new LiveVersionMap();
-        map.putUnderLock(uid("test"), new VersionValue(1));
-        boolean withinRefresh = randomBoolean();
-        if (withinRefresh) {
-            map.beforeRefresh();
-        }
-        assertEquals(new VersionValue(1), map.getUnderLock(uid("test")));
-        final String msg;
-        if (Assertions.ENABLED) {
-            msg = expectThrows(AssertionError.class, map::adjustMapSizeUnderLock).getMessage();
-        } else {
-            msg = expectThrows(IllegalStateException.class, map::adjustMapSizeUnderLock).getMessage();
-        }
-        assertEquals("map must be empty", msg);
-        assertEquals(new VersionValue(1), map.getUnderLock(uid("test")));
-        if (withinRefresh == false) {
-            map.beforeRefresh();
-        }
-        map.afterRefresh(randomBoolean());
-        Map<BytesRef, VersionValue> allCurrent = map.getAllCurrent();
-        map.adjustMapSizeUnderLock();
-        assertNotSame(allCurrent, map.getAllCurrent());
-    }
-
     public void testConcurrently() throws IOException, InterruptedException {
         HashSet<BytesRef> keySet = new HashSet<>();
         int numKeys = randomIntBetween(50, 200);