Concurrent file chunk fetching for CCR restore

[ywelsch]

Adds the ability to fetch chunks from different files in parallel, configurable using the new ccr.indices.recovery.max_concurrent_file_chunks setting, which defaults to 5.

The implementation uses the parallel file writer functionality that is also used by peer recoveries.

[elasticmachine]

Pinging @elastic/es-distributed

[ywelsch]

@elasticmachine run elasticsearch-ci/2

[Tim-Brooks]

It seems like you have a good idea that this works based upon the benchmark usages. I don't see anything wrong with it. It comes across as pretty complicated. I don't know if this is like a first step towards making the whole process asynchronous. But in the world where there is essentially a dedicated generic thread waiting on everything, I think the synchronization and all the thread interactions are difficult to reason about.

In the world where there is a dedicated generic thread I would do something like this and the get file chunk callbacks would just enqueue the results of the request in the blocking queue. And then all the operations (writing locally, rate limiting, error handling, etc) would be performed by the generic thread.

            class Result {
                private final Exception exception;
                private final FileInfo fileInfo;
                private final BytesReference content;

                Result(FileInfo fileInfo, Exception exception, BytesReference content) {
                    this.exception = exception;
                    this.fileInfo = fileInfo;
                    this.content = content;
                }

                private boolean isError(){
                    return exception != null;
                }
            }

            int inflightRequests = 0;
            ArrayDeque<FileInfo> remainingFiles = new ArrayDeque<>(filesToRecover);
            ArrayBlockingQueue<Result> queue = new ArrayBlockingQueue<>(ccrSettings.getMaxConcurrentFileChunks());

But maybe there is a reason why you took dispatch to generic threads on the callback approach. So pursue whatever makes sense to you.

[Tim-Brooks]

I think we should assert that the file length is not greater? I think we should continue to expect that the follower has a correct understanding of the length of the files it is fetching.

[ywelsch]

see a000339

[Tim-Brooks]

I'm not wild about the read method slicing internally and returning a different bytes reference. I also don't really understand why it is necessary as the follower should never request more data than is remaining in a file (you continue to only request the remaining bytes in this PR).

[ywelsch]

The current API for getting a file is more iterator-oriented (i.e. read next, read next, read next), which is why I find it weird that the target will have to track information about where it is at in that file and make sure to request the right size. I have reverted this for now.

[Tim-Brooks]

We have lost the timeout. You can configure request options by overriding a method on the action.

Unfortunately it is not apparent to me if there is a way to get them back while using callbacks. We could always move the restore process into a local transport action and use the guice injections to get the RemoteClusterService and TransportService. But that is probably beyond the scope of this PR.

[ywelsch]

yeah, I would like us to investigate this in a follow-up.

[Tim-Brooks]

It feels like we should throttle after writing the file chunk? Since I think the rate limiting activity more about network activity, we might as well spend some time writing to disk before determining if we want to pause.

[ywelsch]

I've chosen to use the same way we do for peer recovery, to keep things consistent. I'm fine if we want to follow-up on this with a change to both.

[ywelsch]

But maybe there is a reason why you took dispatch to generic threads on the callback approach

I want to make the restore process fully async, and the current approach here gets us closer to that. I've refactored some of the code to make it hopefully clearer what's going on.

[Tim-Brooks]

LGTM