[VL] Celeborn shuffle reader OOM with many empty input stream

[marin-ma]

Below code piece will create a buffered input stream on native side. The buffer is released on close() called when EOS is reached in readValue(). For empty input stream, the readValue function is never called, and the buffered memory can only be released at task end.

      ShuffleReaderJniWrapper
        .create(runtime)
        .readStream(shuffleReaderHandle, byteIn))

[github-actions]

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/apache/incubator-gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

• Other pull requests