class RemoteCollectorFactory {

    private static final Logger LOGGER = LogManager.getLogger(RemoteCollectorFactory.class);
    private static final int SENDER_PHASE_ID = 0;

    private final ClusterService clusterService;
    private final TasksService tasksService;
    private final ElasticsearchClient elasticsearchClient;
    private final IndicesService indicesService;
    private final Executor searchTp;

    @Inject
    public RemoteCollectorFactory(ClusterService clusterService,
                                  TasksService tasksService,
                                  Node node,
                                  IndicesService indicesService,
                                  ThreadPool threadPool) {
        this.clusterService = clusterService;
        this.tasksService = tasksService;
        this.elasticsearchClient = node.client();
        this.indicesService = indicesService;
        searchTp = threadPool.executor(ThreadPool.Names.SEARCH);
    }

    /**
     * create a RemoteCollector
     * The RemoteCollector will collect data from another node using a wormhole as if it was collecting on this node.
     * <p>
     * This should only be used if a shard is not available on the current node due to a relocation
     */
    public CompletableFuture<BatchIterator<Row>> createCollector(ShardId shardId,
                                                                 RoutedCollectPhase collectPhase,
                                                                 CollectTask collectTask,
                                                                 ShardCollectorProviderFactory shardCollectorProviderFactory,
                                                                 boolean requiresScroll) {<FILL_FUNCTION_BODY>}

    private CompletableFuture<BatchIterator<Row>> localOrRemoteCollect(
            ShardRouting primaryRouting,
            RoutedCollectPhase collectPhase,
            CollectTask collectTask,
            ShardCollectorProviderFactory collectorFactory,
            boolean requiresScroll) {
        String nodeId = primaryRouting.currentNodeId();
        String localNodeId = clusterService.localNode().getId();
        if (localNodeId.equalsIgnoreCase(nodeId)) {
            var indexShard = indicesService.indexServiceSafe(primaryRouting.index())
                .getShard(primaryRouting.shardId().id());
            var collectorProvider = collectorFactory.create(indexShard);
            try {
                return collectorProvider.awaitShardSearchActive().thenApply(
                    biFactory -> biFactory.getIterator(collectPhase, requiresScroll, collectTask));
            } catch (Exception e) {
                throw Exceptions.toRuntimeException(e);
            }
        } else {
            return remoteBatchIterator(primaryRouting, collectPhase, collectTask, requiresScroll);
        }
    }

    private CompletableFuture<BatchIterator<Row>> remoteBatchIterator(ShardRouting primaryRouting,
                                                                      RoutedCollectPhase collectPhase,
                                                                      CollectTask collectTask,
                                                                      boolean requiresScroll) {
        CapturingRowConsumer consumer = new CapturingRowConsumer(requiresScroll, collectTask.completionFuture());
        String remoteNodeId = primaryRouting.currentNodeId();
        String localNodeId = clusterService.localNode().getId();
        UUID childJobId = UUIDs.dirtyUUID();
        LOGGER.trace(
            "Creating child remote collect with id={} for parent job={}",
            childJobId,
            collectPhase.jobId()
        );
        RemoteCollector remoteCollector = new RemoteCollector(
            childJobId,
            collectTask.txnCtx().sessionSettings(),
            localNodeId,
            remoteNodeId,
            req -> elasticsearchClient.execute(JobAction.INSTANCE, req),
            req -> elasticsearchClient.execute(KillJobsNodeAction.INSTANCE, req),
            searchTp,
            tasksService,
            collectTask.getRamAccounting(),
            consumer,
            createRemoteCollectPhase(childJobId, collectPhase, primaryRouting.shardId(), remoteNodeId)
        );
        collectTask.completionFuture().exceptionally(err -> {
            remoteCollector.kill(err);
            consumer.capturedBatchIterator().whenComplete((bi, ignored) -> {
                if (bi != null) {
                    bi.kill(err);
                }
            });
            return null;
        });
        remoteCollector.doCollect();
        return consumer.capturedBatchIterator();
    }

    private static RoutedCollectPhase createRemoteCollectPhase(UUID childJobId,
                                                               RoutedCollectPhase collectPhase,
                                                               ShardId shardId,
                                                               String nodeId) {

        Routing routing = new Routing(
            Map.of(
                nodeId, Map.of(shardId.getIndexName(), IntArrayList.from(shardId.id()))
            )
        );
        return new RoutedCollectPhase(
            childJobId,
            SENDER_PHASE_ID,
            collectPhase.name(),
            routing,
            collectPhase.maxRowGranularity(),
            collectPhase.toCollect(),
            Projections.shardProjections(collectPhase.projections()),
            collectPhase.where(),
            DistributionInfo.DEFAULT_BROADCAST
        );
    }
}

        ShardStateObserver shardStateObserver = new ShardStateObserver(clusterService);
        return shardStateObserver.waitForActiveShard(shardId).thenCompose(primaryRouting -> localOrRemoteCollect(
            primaryRouting,
            collectPhase,
            collectTask,
            shardCollectorProviderFactory,
            requiresScroll
        ));
    
<no_super_class>