[Remote Cluster State] Ensure proper closure of IndexInput created in ChecksumBlobStoreFormat

[vikasvb90]

Is your feature request related to a problem? Please describe.
In the following code, IndexInput is opened and closed right after remote async calls are triggered. Currently, these streams have no-op close but if in future these are replaced with streams which require close then they can potentially throw errors (segment faults in case of MMap directories) because by the time async calls are made streams would have been already closed.

        try (IndexInput input = new ByteArrayIndexInput(resourceDescription, BytesReference.toBytes(bytes))) {
            long expectedChecksum;
            try {
                expectedChecksum = checksumOfChecksum(input.clone(), 8);
            } catch (Exception e) {
                throw new ChecksumCombinationException(
                    "Potentially corrupted file: Checksum combination failed while combining stored checksum "
                        + "and calculated checksum of stored checksum",
                    resourceDescription,
                    e
                );
            }

            try (
                RemoteTransferContainer remoteTransferContainer = new RemoteTransferContainer(
                    blobName,
                    blobName,
                    bytes.length(),
                    true,
                    priority,
                    (size, position) -> new OffsetRangeIndexInputStream(input, size, position),
                    expectedChecksum,
                    ((AsyncMultiStreamBlobContainer) blobContainer).remoteIntegrityCheckSupported()
                )
            ) {
                ((AsyncMultiStreamBlobContainer) blobContainer).asyncBlobUpload(remoteTransferContainer.createWriteContext(), listener);
            }
        }

Describe the solution you'd like
All close calls should be made in the listener passed to asyncBlobUpload. Reference code from RemoteDirectory.java

ActionListener<Void> completionListener = ActionListener.wrap(resp -> {
            try {
                postUploadRunner.run();
                listener.onResponse(null);
            } catch (Exception e) {
                logger.error(() -> new ParameterizedMessage("Exception in segment postUpload for file [{}]", src), e);
                listener.onFailure(e);
            }
        }, ex -> {
            logger.error(() -> new ParameterizedMessage("Failed to upload blob {}", src), ex);
            IOException corruptIndexException = ExceptionsHelper.unwrapCorruption(ex);
            if (corruptIndexException != null) {
                listener.onFailure(corruptIndexException);
                return;
            }
            Throwable throwable = ExceptionsHelper.unwrap(ex, CorruptFileException.class);
            if (throwable != null) {
                CorruptFileException corruptFileException = (CorruptFileException) throwable;
                listener.onFailure(new CorruptIndexException(corruptFileException.getMessage(), corruptFileException.getFileName()));
                return;
            }
            listener.onFailure(ex);
        });

        completionListener = ActionListener.runBefore(completionListener, () -> {
            try {
                remoteTransferContainer.close();
            } catch (Exception e) {
                logger.warn("Error occurred while closing streams", e);
            }
        });

        WriteContext writeContext = remoteTransferContainer.createWriteContext();
        ((AsyncMultiStreamBlobContainer) blobContainer).asyncBlobUpload(writeContext, completionListener);

[Bukhtawar]

Close is no-op today, if the stream close before the upload completes, it might lead to grievous bugs. Need better protection.

[Storage Triage - attendees 1 2 3 4 5 6 7 8 9 10