### What changes were proposed in this pull request?
Followup from #27831 , origin author chrysan.
Each request it will check `chunksBeingTransferred `
```
public long chunksBeingTransferred() {
long sum = 0L;
for (StreamState streamState: streams.values()) {
sum += streamState.chunksBeingTransferred.get();
}
return sum;
}
```
such as
```
long chunksBeingTransferred = streamManager.chunksBeingTransferred();
if (chunksBeingTransferred >= maxChunksBeingTransferred) {
logger.warn("The number of chunks being transferred {} is above {}, close the connection.",
chunksBeingTransferred, maxChunksBeingTransferred);
channel.close();
return;
}
```
It will traverse `streams` repeatedly and we know that fetch data chunk will access `stream` too, there cause two problem:
1. repeated traverse `streams`, the longer the length, the longer the time
2. lock race in ConcurrentHashMap `streams`
In this PR, when `maxChunksBeingTransferred` use default value, we avoid compute `chunksBeingTransferred ` since we don't care about this. If user want to set this configuration and meet performance problem, you can also backport PR #27831
### Why are the changes needed?
Speed up getting `chunksBeingTransferred` and avoid lock race in object `streams`
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Existed UT
Closes#30139 from AngersZhuuuu/SPARK-31069.
Lead-authored-by: angerszhu <angers.zhu@gmail.com>
Co-authored-by: chrysan <chrysanxia@gmail.com>
Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>
dd32f45d20