[#291] feat(client): Introduce PrefetchableClientReadHandler to support async read

[zuston]

What changes were proposed in this pull request?

1. Introduce PrefetchableClientReadHandler to support async read. And this will be disabled by default.
2. Apply for the memory/localfile/hdfs read handler

Why are the changes needed?

Recently I found some important spark jobs are slow due to the lots of shuffle read operations. If we could support async read, the job's performance will be improved.

So this PR is the callback for #291. almost 3 years ago!

Does this PR introduce any user-facing change?

Yes. Some configs are introduced

1. rss.client.read.prefetch.enabled
2. rss.client.read.prefetch.capacity
3. rss.client.read.prefetch.timeoutSec

How was this patch tested?

1. Unit tests

[zuston]

cc @jerqi

[jerqi]

1. What will it happen if the operation of read fails? Will it break the order of the data? Because it maybe ordered if mr sort data in the server side.

[zuston]

1. What will it happen if the operation of read fails? Will it break the order of the data? Because it maybe ordered if mr sort data in the server side.

No, the real concurrency number is 1.

[github-actions]

Test Results

 2 996 files  +15   2 996 suites  +15   6h 28m 22s ⏱️ -2s
 1 105 tests + 4   1 103 ✅ + 4   2 💤 ±0  0 ❌ ±0 
13 849 runs  +60  13 819 ✅ +60  30 💤 ±0  0 ❌ ±0 

Results for commit 4e7c659. ± Comparison against base commit e5cfc4a.

♻️ This comment has been updated with latest results.

[jerqi]

Could you add document for these config options?
Do we have read timeout option? What's the relationship between it and prefetchTimeout?

[zuston]

What's the relationship between it and prefetchTimeout?

If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail

[jerqi]

What's the relationship between it and prefetchTimeout?

If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail

PrefetchTimeout may be equal to read time out. Isn't right? They are all the one time read max time.

[zuston]

What's the relationship between it and prefetchTimeout?

If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail

PrefetchTimeout may be equal to read time out. Isn't right? They are all the one time read max time.

Yes. But I hope it could be configured (< grpc read timeout)

[jerqi]

What's the relationship between it and prefetchTimeout?

If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail

PrefetchTimeout may be equal to read time out. Isn't right? They are all the one time read max time.

Yes. But I hope it could be configured (< grpc read timeout)

OK.