Close correctly

src/main/java/com/google/devtools/build/lib/remote/Chunker.java

@@ -27,6 +27,7 @@
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.protobuf.ByteString;
 import java.io.ByteArrayInputStream;
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PushbackInputStream;
@@ -141,7 +142,6 @@ public void reset() throws IOException {
     close();
     offset = 0;
     initialized = false;
-    dataSupplier.close();
   }
 
   /**
@@ -184,6 +184,10 @@ private void close() throws IOException {
     chunkCache = null;
   }
 
+  public void closeQuietly() {
+    dataSupplier.close();
+  }
+
   /** Attempts reading at most a full chunk and stores it in the chunkCache buffer */
   private int read() throws IOException {
     int count = 0;

src/main/java/com/google/devtools/build/lib/remote/GrpcCacheClient.java

@@ -501,6 +501,7 @@ ListenableFuture<Void> uploadChunker(
         () -> {
           try {
             chunker.reset();
+            chunker.closeQuietly();
           } catch (IOException e) {
             logger.atWarning().withCause(e).log(
                 "failed to reset chunker uploading %s/%d", digest.getHash(), digest.getSizeBytes());

src/main/java/com/google/devtools/build/lib/remote/http/HttpCacheClient.java

@@ -639,7 +639,7 @@ private ListenableFuture<Void> uploadAsync(
           public void close() {
             // Ensure that the InputStream can't be closed somewhere in the Netty
             // pipeline, so that we can support retries. The InputStream is closed in
-            // the finally block below.
+            // the listener block below.
           }
         };
     UploadCommand upload = new UploadCommand(uri, casUpload, key, wrappedIn, length);
@@ -720,8 +720,12 @@ private void uploadAfterCredentialRefresh(UploadCommand upload, SettableFuture<V
   public ListenableFuture<Void> uploadBlob(
       RemoteActionExecutionContext context, Digest digest, CloseableBlobSupplier in) {
     return retrier.executeAsync(
-        () ->
-            uploadAsync(digest.getHash(), digest.getSizeBytes(), in.get(), /* casUpload= */ true));
+        () -> {
+          var result =
+              uploadAsync(digest.getHash(), digest.getSizeBytes(), in.get(), /* casUpload= */ true);
+          result.addListener(in::close, MoreExecutors.directExecutor());
+          return result;
+        });
   }
 
   @Override