Fix reindex from remote clearing scroll (#22525)

Reindex-from-remote had a race when it tried to clear the scroll. It
first starts the request to clear the scroll and then submits a task
to the generic threadpool to shutdown the client. These two things
race and, in my experience, closing the scroll generally loses. That
means that most of the time reindex-from-remote isn't clearing the
scrolls that it uses. This isn't the end of the world because we
flush old scroll contexts after a while but this isn't great.

Noticed while experimenting with #22514.

Author: nik9000

modules/reindex/src/main/java/org/elasticsearch/index/reindex/ClientScrollableHitSource.java

@@ -90,11 +90,7 @@ protected void doStartNextScroll(String scrollId, TimeValue extraKeepAlive, Cons
     }
 
     @Override
-    public void clearScroll(String scrollId) {
-        /*
-         * Fire off the clear scroll but don't wait for it it return before
-         * we send the use their response.
-         */
+    public void clearScroll(String scrollId, Runnable onCompletion) {
         ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
         clearScrollRequest.addScrollId(scrollId);
         /*
@@ -105,15 +101,22 @@ public void clearScroll(String scrollId) {
             @Override
             public void onResponse(ClearScrollResponse response) {
                 logger.debug("Freed [{}] contexts", response.getNumFreed());
+                onCompletion.run();
             }
 
             @Override
             public void onFailure(Exception e) {
                 logger.warn((Supplier<?>) () -> new ParameterizedMessage("Failed to clear scroll [{}]", scrollId), e);
+                onCompletion.run();
             }
         });
     }
 
+    @Override
+    protected void cleanup() {
+        // Nothing to do
+    }
+
     /**
      * Run a search action and call onResponse when a the response comes in, retrying if the action fails with an exception caused by
      * rejected execution.

modules/reindex/src/main/java/org/elasticsearch/index/reindex/ScrollableHitSource.java

@@ -83,13 +83,24 @@ public final void startNextScroll(TimeValue extraKeepAlive, Consumer<Response> o
     protected abstract void doStartNextScroll(String scrollId, TimeValue extraKeepAlive, Consumer<? super Response> onResponse);
 
     @Override
-    public void close() {
+    public final void close() {
         String scrollId = this.scrollId.get();
         if (Strings.hasLength(scrollId)) {
-            clearScroll(scrollId);
+            clearScroll(scrollId, this::cleanup);
+        } else {
+            cleanup();
         }
     }
-    protected abstract void clearScroll(String scrollId);
+    /**
+     * Called to clear a scroll id.
+     * @param scrollId the id to clear
+     * @param onCompletion implementers must call this after completing the clear whether they are successful or not
+     */
+    protected abstract void clearScroll(String scrollId, Runnable onCompletion);
+    /**
+     * Called after the process has been totally finished to clean up any resources the process needed like remote connections.
+     */
+    protected abstract void cleanup();
 
     /**
      * Set the id of the last scroll. Used for debugging.

modules/reindex/src/main/java/org/elasticsearch/index/reindex/remote/RemoteScrollableHitSource.java

@@ -81,21 +81,6 @@ public RemoteScrollableHitSource(Logger logger, BackoffPolicy backoffPolicy, Thr
         this.client = client;
     }
 
-    @Override
-    public void close() {
-        super.close();
-        /* This might be called on the RestClient's thread pool and attempting to close the client on its own threadpool causes it to fail
-         * to close. So we always shutdown the RestClient asynchronously on a thread in Elasticsearch's generic thread pool. That way we
-         * never close the client in its own thread pool. */
-        threadPool.generic().submit(() -> {
-            try {
-                client.close();
-            } catch (IOException e) {
-                logger.error("Failed to shutdown the remote connection", e);
-            }
-        });
-    }
-
     @Override
     protected void doStart(Consumer<? super Response> onResponse) {
         lookupRemoteVersion(version -> {
@@ -125,17 +110,32 @@ protected void doStartNextScroll(String scrollId, TimeValue extraKeepAlive, Cons
     }
 
     @Override
-    protected void clearScroll(String scrollId) {
-        // Need to throw out response....
+    protected void clearScroll(String scrollId, Runnable onCompletion) {
         client.performRequestAsync("DELETE", scrollPath(), emptyMap(), scrollEntity(scrollId), new ResponseListener() {
             @Override
             public void onSuccess(org.elasticsearch.client.Response response) {
                 logger.debug("Successfully cleared [{}]", scrollId);
+                onCompletion.run();
             }
 
             @Override
             public void onFailure(Exception t) {
                 logger.warn((Supplier<?>) () -> new ParameterizedMessage("Failed to clear scroll [{}]", scrollId), t);
+                onCompletion.run();
+            }
+        });
+    }
+
+    @Override
+    protected void cleanup() {
+        /* This is called on the RestClient's thread pool and attempting to close the client on its own threadpool causes it to fail to
+         * close. So we always shutdown the RestClient asynchronously on a thread in Elasticsearch's generic thread pool. */
+        threadPool.generic().submit(() -> {
+            try {
+                client.close();
+                logger.info("Shut down remote connection");
+            } catch (IOException e) {
+                logger.error("Failed to shutdown the remote connection", e);
             }
         });
     }