Return MergeObserver from IndexWriter.forceMergeDeletes() (#15378)

Return MergeObserver from IndexWriter.forceMergeDeletes()

IndexWriter.forceMergeDeletes() now returns MergePolicy.MergeObserver
instead of void, allowing applications to monitor merge progress and
wait for completion. Addresses (#14515).

Author: salvatorecampagna

lucene/CHANGES.txt

@@ -17,6 +17,12 @@ API Changes
 * GITHUB#15234: Remove unused method which does unsafe XML parsing.
   (Uwe Schindler, Isaac David)
 
+* GITHUB#14515: IndexWriter.forceMergeDeletes() now returns MergePolicy.MergeObserver,
+  allowing applications to monitor merge progress and wait for completion. The observer
+  provides methods to query merge count, check completion status, and wait synchronously
+  (with optional timeout) or asynchronously via CompletableFuture. Backward compatible -
+  existing code that ignores the return value works unchanged. (Salvatore Campagna)
+
 New Features
 ---------------------
 * GITHUB#15328: VectorSimilarityFunction.getValues() now implements doubleVal allowing its

lucene/core/src/java/org/apache/lucene/index/IndexWriter.java

@@ -2232,8 +2232,10 @@ private synchronized boolean maxNumSegmentsMergesPending() {
    * Just like {@link #forceMergeDeletes()}, except you can specify whether the call should block
    * until the operation completes. This is only meaningful with a {@link MergeScheduler} that is
    * able to run merges in background threads.
+   *
+   * @return a {@link MergePolicy.MergeObserver} to monitor merge progress and wait for completion
    */
-  public void forceMergeDeletes(boolean doWait) throws IOException {
+  public MergePolicy.MergeObserver forceMergeDeletes(boolean doWait) throws IOException {
     ensureOpen();
 
     flush(true, true);
@@ -2293,6 +2295,7 @@ public void forceMergeDeletes(boolean doWait) throws IOException {
     // NOTE: in the ConcurrentMergeScheduler case, when
     // doWait is false, we can return immediately while
     // background threads accomplish the merging
+    return new MergePolicy.MergeObserver(spec);
   }
 
   /**
@@ -2307,9 +2310,12 @@ public void forceMergeDeletes(boolean doWait) throws IOException {
    *
    * <p><b>NOTE</b>: this method first flushes a new segment (if there are indexed documents), and
    * applies all buffered deletes.
+   *
+   * @return a {@link MergePolicy.MergeObserver} to monitor merge progress. Since this method blocks
+   *     until completion, merges will already be complete when it returns.
    */
-  public void forceMergeDeletes() throws IOException {
-    forceMergeDeletes(true);
+  public MergePolicy.MergeObserver forceMergeDeletes() throws IOException {
+    return forceMergeDeletes(true);
   }
 
   /**

lucene/core/src/java/org/apache/lucene/index/MergePolicy.java

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.EnumMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -939,4 +940,94 @@ static final class MergeReader {
       this.hardLiveDocs = hardLiveDocs;
     }
   }
+
+  /**
+   * Observer for merge operations returned by {@link IndexWriter#forceMergeDeletes(boolean)}.
+   * Provides methods to query merge status and wait for completion.
+   *
+   * <p>When no merges are needed, {@link #numMerges()} returns 0. In this case, {@link #await()}
+   * returns {@code true} immediately since there is nothing to wait for.
+   *
+   * @lucene.experimental
+   */
+  public static final class MergeObserver {
+    private final MergePolicy.MergeSpecification spec;
+
+    MergeObserver(MergePolicy.MergeSpecification spec) {
+      this.spec = spec;
+    }
+
+    /**
+     * Returns the number of merges in this specification.
+     *
+     * @return number of merges, or 0 if no merges were scheduled
+     */
+    public int numMerges() {
+      return spec == null ? 0 : spec.merges.size();
+    }
+
+    /**
+     * Returns the number of completed merges in this specification. Useful for tracking merge
+     * progress: {@code numCompletedMerges() / numMerges()}.
+     *
+     * @return number of completed merges
+     */
+    public int numCompletedMerges() {
+      if (spec == null) {
+        return 0;
+      }
+      int completed = 0;
+      for (OneMerge merge : spec.merges) {
+        if (merge.mergeCompleted.isDone()) {
+          completed++;
+        }
+      }
+      return completed;
+    }
+
+    /**
+     * Waits for all merges in this specification to complete. Returns immediately if no merges were
+     * scheduled.
+     *
+     * @return {@code true} if all merges completed successfully or no merges were needed, {@code
+     *     false} on error
+     */
+    public boolean await() {
+      return spec == null || spec.await();
+    }
+
+    /**
+     * Waits for all merges in this specification to complete, with timeout. Returns immediately if
+     * no merges were scheduled.
+     *
+     * @param timeout maximum time to wait
+     * @param unit time unit for timeout
+     * @return {@code true} if all merges completed within timeout or no merges were needed, {@code
+     *     false} on timeout or error
+     */
+    public boolean await(long timeout, TimeUnit unit) {
+      return spec == null || spec.await(timeout, unit);
+    }
+
+    /**
+     * Returns a {@link CompletableFuture} that completes when all merges finish. Returns an
+     * already-completed future if no merges were scheduled.
+     *
+     * @return future that completes when merges finish
+     */
+    public CompletableFuture<Void> awaitAsync() {
+      return spec == null
+          ? CompletableFuture.completedFuture(null)
+          : spec.getMergeCompletedFutures();
+    }
+
+    @Override
+    public String toString() {
+      if (spec == null) {
+        return "MergeObserver: no merges";
+      }
+      return String.format(
+          Locale.ROOT, "MergeObserver: %d merges\n%s", numMerges(), spec.toString());
+    }
+  }
 }