Determine remote cluster version

modules/reindex/src/main/java/org/elasticsearch/reindex/BulkByPaginatedSearchParallelizationHelper.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.reindex;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsRequest;
@@ -17,6 +18,7 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.reindex.AbstractBulkByScrollRequest;
@@ -35,6 +37,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.index.reindex.AbstractBulkByScrollRequest.AUTO_SLICES;
@@ -73,19 +76,24 @@ static <Request extends AbstractBulkByScrollRequest<Request>> void startSlicedAc
             task,
             request,
             client,
-            listener.delegateFailure((l, v) -> executeSlicedAction(task, request, action, l, client, node, workerAction))
+            listener.delegateFailure(
+                (l, v) -> executeSlicedAction(task, request, action, l, client, node, null, version -> workerAction.run())
+            )
         );
     }
 
     /**
      * Takes an action and a {@link BulkByScrollTask} and runs it with regard to whether this task is a
      * leader or worker.
      *
-     * If this task is a worker, the worker action in the given {@link Runnable} will be started on the local
-     * node. If the task is a leader (i.e. the number of slices is more than 1), then a subrequest will be
-     * created for each slice and sent.
+     * If this task is a worker, the worker action is invoked with the given {@code remoteVersion} (may be null
+     * for local reindex). If the task is a leader (i.e. the number of slices is more than 1), then a subrequest
+     * will be created for each slice and sent.
      *
      * This method can only be called after the task state is initialized {@link #initTaskState}.
+     *
+     * @param remoteVersion the version of the remote cluster when reindexing from remote, or null for local reindex
+     * @param workerAction  invoked when this task is a worker, with the remote version (or null)
      */
     static <Request extends AbstractBulkByScrollRequest<Request>> void executeSlicedAction(
         BulkByScrollTask task,
@@ -94,12 +102,13 @@ static <Request extends AbstractBulkByScrollRequest<Request>> void executeSliced
         ActionListener<BulkByScrollResponse> listener,
         Client client,
         DiscoveryNode node,
-        Runnable workerAction
+        @Nullable Version remoteVersion,
+        Consumer<Version> workerAction
     ) {
         if (task.isLeader()) {
             sendSubRequests(client, action, node.getId(), task, request, listener);
         } else if (task.isWorker()) {
-            workerAction.run();
+            workerAction.accept(remoteVersion);
         } else {
             throw new AssertionError("Task should have been initialized at this point.");
         }

modules/reindex/src/main/java/org/elasticsearch/reindex/Reindexer.java

@@ -19,6 +19,7 @@
 import org.apache.http.message.BasicHeader;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.DocWriteRequest;
@@ -53,11 +54,13 @@
 import org.elasticsearch.index.reindex.PaginatedHitSource;
 import org.elasticsearch.index.reindex.ReindexAction;
 import org.elasticsearch.index.reindex.ReindexRequest;
+import org.elasticsearch.index.reindex.RejectAwareActionListener;
 import org.elasticsearch.index.reindex.RemoteInfo;
 import org.elasticsearch.index.reindex.ResumeBulkByScrollRequest;
 import org.elasticsearch.index.reindex.ResumeBulkByScrollResponse;
 import org.elasticsearch.index.reindex.ResumeReindexAction;
 import org.elasticsearch.index.reindex.WorkerBulkByScrollTaskState;
+import org.elasticsearch.reindex.remote.RemoteReindexingUtils;
 import org.elasticsearch.reindex.remote.RemoteScrollablePaginatedHitSource;
 import org.elasticsearch.script.CtxMap;
 import org.elasticsearch.script.ReindexMetadata;
@@ -83,12 +86,15 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.synchronizedList;
+import static org.elasticsearch.common.BackoffPolicy.exponentialBackoff;
 import static org.elasticsearch.index.VersionType.INTERNAL;
+import static org.elasticsearch.reindex.ReindexPlugin.REINDEX_PIT_SEARCH_ENABLED;
 
 public class Reindexer {
 
@@ -143,33 +149,113 @@ public void execute(BulkByScrollTask task, ReindexRequest request, Client bulkCl
         // for update-by-query and delete-by-query
         final ActionListener<BulkByScrollResponse> listenerWithRelocations = listenerWithRelocations(task, request, listener);
 
-        BulkByPaginatedSearchParallelizationHelper.executeSlicedAction(
-            task,
-            request,
-            ReindexAction.INSTANCE,
-            listenerWithRelocations,
-            client,
-            clusterService.localNode(),
-            () -> {
-                ParentTaskAssigningClient assigningClient = new ParentTaskAssigningClient(client, clusterService.localNode(), task);
-                ParentTaskAssigningClient assigningBulkClient = new ParentTaskAssigningClient(bulkClient, clusterService.localNode(), task);
-                AsyncIndexBySearchAction searchAction = new AsyncIndexBySearchAction(
-                    task,
-                    logger,
-                    assigningClient,
-                    assigningBulkClient,
-                    threadPool,
-                    scriptService,
-                    projectResolver.getProjectState(clusterService.state()),
-                    reindexSslConfig,
-                    request,
-                    workerListenerWithRelocationAndMetrics(listenerWithRelocations, startTime, request.getRemoteInfo() != null)
+        Consumer<Version> workerAction = remoteVersion -> {
+            ParentTaskAssigningClient assigningClient = new ParentTaskAssigningClient(client, clusterService.localNode(), task);
+            ParentTaskAssigningClient assigningBulkClient = new ParentTaskAssigningClient(bulkClient, clusterService.localNode(), task);
+            AsyncIndexBySearchAction searchAction = new AsyncIndexBySearchAction(
+                task,
+                logger,
+                assigningClient,
+                assigningBulkClient,
+                threadPool,
+                scriptService,
+                projectResolver.getProjectState(clusterService.state()),
+                reindexSslConfig,
+                request,
+                workerListenerWithRelocationAndMetrics(listenerWithRelocations, startTime, request.getRemoteInfo() != null),
+                remoteVersion
+            );
+            searchAction.start();
+        };
+
+        /**
+         * If this is a request to reindex from remote, then we need to determine the remote version prior to execution
+         * NB {@link ReindexRequest} forbids remote requests and slices > 1, so we're guaranteed to be running on the only slice
+         */
+        if (REINDEX_PIT_SEARCH_ENABLED && request.getRemoteInfo() != null) {
+            lookupRemoteVersionAndExecute(task, request, listenerWithRelocations, workerAction);
+        } else {
+            BulkByPaginatedSearchParallelizationHelper.executeSlicedAction(
+                task,
+                request,
+                ReindexAction.INSTANCE,
+                listenerWithRelocations,
+                client,
+                clusterService.localNode(),
+                null,
+                workerAction
+            );
+        }
+    }
+
+    /**
+     * Looks up the remote cluster version when reindexing from a remote source, then runs the sliced action with that version.
+     * The RestClient used for the lookup is closed after the callback; closing must happen on a thread other than the
+     * RestClient's own thread pool to avoid shutdown failures.
+     */
+    private void lookupRemoteVersionAndExecute(
+        BulkByScrollTask task,
+        ReindexRequest request,
+        ActionListener<BulkByScrollResponse> listenerWithRelocations,
+        Consumer<Version> workerAction
+    ) {
+        RemoteInfo remoteInfo = request.getRemoteInfo();
+        assert reindexSslConfig != null : "Reindex ssl config must be set";
+        RestClient restClient = buildRestClient(remoteInfo, reindexSslConfig, task.getId(), synchronizedList(new ArrayList<>()));
+        RejectAwareActionListener<Version> rejectAwareListener = new RejectAwareActionListener<>() {
+            @Override
+            public void onResponse(Version version) {
+                closeRestClientAndRun(
+                    restClient,
+                    () -> BulkByPaginatedSearchParallelizationHelper.executeSlicedAction(
+                        task,
+                        request,
+                        ReindexAction.INSTANCE,
+                        listenerWithRelocations,
+                        client,
+                        clusterService.localNode(),
+                        version,
+                        workerAction
+                    )
                 );
-                searchAction.start();
             }
+
+            @Override
+            public void onFailure(Exception e) {
+                closeRestClientAndRun(restClient, () -> listenerWithRelocations.onFailure(e));
+            }
+
+            @Override
+            public void onRejection(Exception e) {
+                closeRestClientAndRun(restClient, () -> listenerWithRelocations.onFailure(e));
+            }
+        };
+        RemoteReindexingUtils.lookupRemoteVersionWithRetries(
+            logger,
+            exponentialBackoff(request.getRetryBackoffInitialTime(), request.getMaxRetries()),
+            threadPool,
+            restClient,
+            // TODO - Do we want to pass in a countRetry runnable here to count the number of times we retry?
+            // https://github.com/elastic/elasticsearch-team/issues/2382
+            rejectAwareListener
         );
     }
 
+    /**
+     * Closes the RestClient on the generic thread pool (to avoid closing from the client's own thread), then runs the given action.
+     */
+    private void closeRestClientAndRun(RestClient restClient, Runnable onCompletion) {
+        threadPool.generic().submit(() -> {
+            try {
+                restClient.close();
+            } catch (IOException e) {
+                logger.warn("Failed to close RestClient after version lookup", e);
+            } finally {
+                onCompletion.run();
+            }
+        });
+    }
+
     /** Wraps the listener with metrics tracking and relocation handling (if applicable). Visible for testing. */
     ActionListener<BulkByScrollResponse> workerListenerWithRelocationAndMetrics(
         ActionListener<BulkByScrollResponse> potentiallyWrappedRelocationListener,
@@ -413,6 +499,11 @@ static class AsyncIndexBySearchAction extends AbstractAsyncBulkByScrollAction<Re
          */
         private List<Thread> createdThreads = emptyList();
 
+        /**
+         * Version of the remote cluster when reindexing from remote, or null when reindexing locally.
+         */
+        private final Version remoteVersion;
+
         AsyncIndexBySearchAction(
             BulkByScrollTask task,
             Logger logger,
@@ -423,7 +514,8 @@ static class AsyncIndexBySearchAction extends AbstractAsyncBulkByScrollAction<Re
             ProjectState state,
             ReindexSslConfig sslConfig,
             ReindexRequest request,
-            ActionListener<BulkByScrollResponse> listener
+            ActionListener<BulkByScrollResponse> listener,
+            @Nullable Version remoteVersion
         ) {
             super(
                 task,
@@ -444,6 +536,7 @@ static class AsyncIndexBySearchAction extends AbstractAsyncBulkByScrollAction<Re
                 sslConfig
             );
             this.destinationIndexIdMapper = destinationIndexMode(state).idFieldMapperWithoutFieldData();
+            this.remoteVersion = remoteVersion;
         }
 
         private IndexMode destinationIndexMode(ProjectState state) {
@@ -477,7 +570,8 @@ protected PaginatedHitSource buildScrollableResultSource(BackoffPolicy backoffPo
                     this::finishHim,
                     restClient,
                     remoteInfo,
-                    searchRequest
+                    searchRequest,
+                    remoteVersion
                 );
             }
             return super.buildScrollableResultSource(backoffPolicy, searchRequest);