KAFKA-14915: allow read remote storage for multiple partitions in one fetchRequest

[showuon]

This PR enables reading remote storage for multiple partitions in one
fetchRequest. The main changes are:

1. In DelayedRemoteFetch, we accept multiple remoteFetchTasks and
   other metadata now.
2. In DelayedRemoteFetch, we'll wait until all remoteFetch done,
   either succeeded or failed.
3. In ReplicaManager#fetchMessage, we'll create one
   DelayedRemoteFetch and pass multiple remoteFetch metadata to it, and
   watch all of them.
4. Add tests

[showuon]

@satishd @kamalcph , please take a look. Thanks.

[kamalcph]

Thanks for the patch! Overall LGTM.

Few points:

1. The FETCH request now has to wait to get response from all the remote partitions. And, it can have tail-latency. This tradeoff is fine considering the consumer might have to read the same data in the upcoming FETCH requests. This can be documented in the PR.
2. When minOneMessage set to false (from second partition in the request), and there are multiple remote reads in the given FETCH request. We may read the data from remote and then drop those when the size of the message does not fit the partition limit. This was fine for local-reads but remote-reads might fetch chunk of data and multiple responses might be dropped. Shall we add some DEBUG logs when the remote fetched data gets dropped?

[showuon]

1. The FETCH request now has to wait to get response from all the remote partitions. And, it can have tail-latency. This tradeoff is fine considering the consumer might have to read the same data in the upcoming FETCH requests. This can be documented in the PR.

Good point. I've updated the doc in remote.fetch.max.wait to mention the waiting all fetch behavior, and reverted the change in this PR about the limitation.

2. When minOneMessage set to false (from second partition in the request), and there are multiple remote reads in the given FETCH request. We may read the data from remote and then drop those when the size of the message does not fit the partition limit. This was fine for local-reads but remote-reads might fetch chunk of data and multiple responses might be dropped. Shall we add some DEBUG logs when the remote fetched data gets dropped?

I agree with you. Added a debug log.

Thanks for the review!

[satishd]

Thanks @showuon for the PR, I will review this PR by Monday.

[kamalcph]

LGTM! Thanks for the patch!

[fvaleri]

LGTM. Thanks.

I tried on my local machine and I can see remote partitions being read in parallel and increased throughput.

[satishd]

Thanks @showuon for the PR, overall LGTM.

Left a couple of comments on minor typos int he test.

[satishd]

Thanks @showuon for addressing the comments. LGTM.