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

@@ -1342,6 +1342,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", SearcherScope.INTERNAL);
             if (indexWriter.hasUncommittedChanges()) {
                 logger.trace("can't sync commit [{}]. have pending changes", syncId);
                 return SyncedFlushResult.PENDING_OPERATIONS;
@@ -1354,8 +1357,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

@@ -38,18 +38,6 @@ final class LiveVersionMap implements ReferenceManager.RefreshListener, Accounta
 
     private final KeyedLock<BytesRef> keyedLock = new KeyedLock<>();
 
-    /**
-     * 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 final class VersionLookup {
 
         private static final VersionLookup EMPTY = new VersionLookup(Collections.emptyMap());

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

@@ -101,38 +101,6 @@ public void testBasics() throws IOException {
         }
     }
 
-
-    public void testAdjustMapSizeUnderLock() throws IOException {
-        LiveVersionMap map = new LiveVersionMap();
-        try (Releasable r = map.acquireLock(uid("test"))) {
-            map.putUnderLock(uid("test"), new VersionValue(1, 1, 1));
-        }
-        boolean withinRefresh = randomBoolean();
-        if (withinRefresh) {
-            map.beforeRefresh();
-        }
-        try (Releasable r = map.acquireLock(uid("test"))) {
-            assertEquals(new VersionValue(1, 1, 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);
-        try (Releasable r = map.acquireLock(uid("test"))) {
-            assertEquals(new VersionValue(1, 1, 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);