DirectReadRequestChain and fallback-read might end up reading from same inputStream

[shubhamtagra]

There is a chance of reading wrong data in following case:

1. First block of read was handled by DirectReadRequestChain
2. Other ReadRequestChains are also involved in this read
3. One of the ReadRequestChains, other than DirectReadRequestChain, quickly hit an error
4. read will cancel all the other ReadRequestChains and fallback to reading from the parent-inputStream
5. This parent-inputStream is shared with DirectReadRequestChain
6. In case DirectReadRequestChain was on the first block and was still reading it, position in stream would match what fallback read would want to read from hence it will just start a read on the inputStream (all other cases will see a backward seek and re-open connection in parent-inputStream)
7. This will cause bad reads as both DirectReadRequestChain and fallback-read are reading from the same parent-inputStream

[raunaqmorarka]

We've now switched to using readFully in DirectReadRequestChain which is guaranteed to be thread-safe. Even if the parent FSInputStream implementation doesn't implement positioned read explicitly, the default implementation (FSInputStream#read(long position, byte[] buffer, int offset, int length)) takes lock on the stream before doing the usual seek and non-positioned read. The other option is to make DirectReadRequestChain open it's own stream on the remote FS and avoid sharing.

[shubhamtagra]

There are two issues now around sharing the stream: this and https://github.com/qubole/rubix/issues/375. Even though the intent in sharing stream was that it would never be used in parallel but it either does get used in cases like current one or causes other issues line in #375

Positional reads are saving us in both these cases. As a fix, opening streams per Chain should be avoided as explained in the comment of #375. Another solution we can try is removing the parallel execution of the Chains, given that engines are already parallelising work enough we might not be gaining anything with this additional parallelism (needs some perf validation though)