Close Input Streams in HDFS and Local Exchange storage

[sanchitkashyap]

Description

When reading slices in HDFS and Local exchange storage, all input streams are not getting closed leading to connection leaks.

Additional context and related issues

Release notes

(x) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

[linzebing]

I might be missing something obvious. sliceInput is closed here, right?

[sanchitkashyap]

Every time getSliceInput is called, a new InputStreamSliceInput is created.. so we are not closing every input stream.

[linzebing]

Oh, then should we delete the code of:

if (sliceInput != null) {
                sliceInput.close();
                sliceInput = null;
}

? It becomes redundant now, right?

[sanchitkashyap]

Done

[sanchitkashyap]

The failing tests look unrelated.
@linzebing Can we merge this change?

[losipiuk]

It looks like we do not need to accumulate all the sliceInputs in closer.
Instead we can do change

            if (sliceInput != null && sliceInput.isReadable()) {
                return sliceInput.readSlice(sliceInput.readInt());
            }

to

            if (sliceInput != null) {
                if (sliceInput.isReadable()) {
                    return sliceInput.readSlice(sliceInput.readInt());
                }
                else {
                    sliceInput.close();
                }
            }