fix: close pending zero-copy responses when Storage#close is called (#…

…2696)

Update gRPC based Storage instances to close out in progress reads when zero-copy is used and Storage#close() is called.

google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java

@@ -70,7 +70,6 @@
 import com.google.storage.v2.StorageSettings;
 import com.google.storage.v2.stub.GrpcStorageCallableFactory;
 import com.google.storage.v2.stub.GrpcStorageStub;
-import com.google.storage.v2.stub.StorageStub;
 import com.google.storage.v2.stub.StorageStubSettings;
 import io.grpc.ClientInterceptor;
 import io.grpc.Detachable;
@@ -89,9 +88,9 @@
 import java.time.Clock;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.IdentityHashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -908,9 +907,27 @@ private Object readResolve() {
 
   private static final class InternalStorageClient extends StorageClient {
 
-    private InternalStorageClient(StorageStub stub) {
+    private InternalStorageClient(InternalZeroCopyGrpcStorageStub stub) {
       super(stub);
     }
+
+    @Override
+    public void shutdownNow() {
+      try {
+        // GrpcStorageStub#close() is final and we can't override it
+        // instead hook in here to close out the zero-copy marshaller
+        getStub().getObjectMediaResponseMarshaller.close();
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      } finally {
+        super.shutdownNow();
+      }
+    }
+
+    @Override
+    public InternalZeroCopyGrpcStorageStub getStub() {
+      return (InternalZeroCopyGrpcStorageStub) super.getStub();
+    }
   }
 
   private static final class InternalZeroCopyGrpcStorageStub extends GrpcStorageStub
@@ -1071,30 +1088,21 @@ public void close() throws IOException {
      * them all as suppressed exceptions on the first occurrence.
      */
     @VisibleForTesting
-    static void closeAllStreams(Collection<InputStream> inputStreams) throws IOException {
-      IOException ioException =
-          inputStreams.stream()
-              .map(
-                  stream -> {
-                    try {
-                      stream.close();
-                      return null;
-                    } catch (IOException e) {
-                      return e;
-                    }
-                  })
-              .filter(Objects::nonNull)
-              .reduce(
-                  null,
-                  (l, r) -> {
-                    if (l != null) {
-                      l.addSuppressed(r);
-                      return l;
-                    } else {
-                      return r;
-                    }
-                  },
-                  (l, r) -> l);
+    static void closeAllStreams(Iterable<InputStream> inputStreams) throws IOException {
+      Iterator<InputStream> iterator = inputStreams.iterator();
+      IOException ioException = null;
+      while (iterator.hasNext()) {
+        InputStream next = iterator.next();
+        try {
+          next.close();
+        } catch (IOException e) {
+          if (ioException == null) {
+            ioException = e;
+          } else {
+            ioException.addSuppressed(e);
+          }
+        }
+      }
 
       if (ioException != null) {
         throw ioException;

google-cloud-storage/src/main/java/com/google/cloud/storage/ResponseContentLifecycleManager.java

@@ -16,10 +16,15 @@
 package com.google.cloud.storage;
 
 import com.google.storage.v2.ReadObjectResponse;
+import java.io.Closeable;
+import java.io.IOException;
 
-interface ResponseContentLifecycleManager {
+interface ResponseContentLifecycleManager extends Closeable {
   ResponseContentLifecycleHandle get(ReadObjectResponse response);
 
+  @Override
+  default void close() throws IOException {}
+
   static ResponseContentLifecycleManager noop() {
     return response ->
         new ResponseContentLifecycleHandle(