Revert "fix: update implementation of readAllBytes and downloadTo to …

…be more robust to retryable errors (#2305)"

This reverts commit 21821da.

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

@@ -101,10 +101,12 @@ public final synchronized int write(ByteBuffer src) throws IOException {
       }
       int write = tmp.write(src);
       return write;
+    } catch (StorageException e) {
+      throw new IOException(e);
     } catch (IOException e) {
       throw e;
     } catch (Exception e) {
-      throw StorageException.coalesce(e);
+      throw new IOException(StorageException.coalesce(e));
     }
   }
 

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

@@ -256,34 +256,11 @@ public Blob create(
 
   @Override
   public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) {
-    requireNonNull(blobInfo, "blobInfo must be non null");
-
-    Opts<ObjectTargetOpt> opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts);
-    GrpcCallContext grpcCallContext =
-        opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault());
-    WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
-
-    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
-        ResumableMedia.gapic()
-            .write()
-            .byteChannel(
-                storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext))
-            .setHasher(Hasher.enabled())
-            .setByteStringStrategy(ByteStringStrategy.noCopy())
-            .direct()
-            .unbuffered()
-            .setRequest(req)
-            .build();
-
-    // Specifically not in the try-with, so we don't close the provided stream
-    ReadableByteChannel src =
-        Channels.newChannel(firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)));
-    try (UnbufferedWritableByteChannel dst = session.open()) {
-      ByteStreams.copy(src, dst);
-    } catch (Exception e) {
+    try {
+      return createFrom(blobInfo, content, options);
+    } catch (IOException e) {
       throw StorageException.coalesce(e);
     }
-    return getBlob(session.getResult());
   }
 
   @Override
@@ -338,7 +315,7 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetOpt> o
       }
       return codecs.blobInfo().decode(object).asBlob(this);
     } catch (InterruptedException | ExecutionException e) {
-      throw StorageException.coalesce(e.getCause());
+      throw StorageException.coalesce(e);
     }
   }
 
@@ -388,14 +365,7 @@ public Blob createFrom(
   @Override
   public Bucket get(String bucket, BucketGetOption... options) {
     Opts<BucketSourceOpt> unwrap = Opts.unwrap(options);
-    try {
-      return internalBucketGet(bucket, unwrap);
-    } catch (StorageException e) {
-      if (e.getCode() == 404) {
-        return null;
-      }
-      throw e;
-    }
+    return internalBucketGet(bucket, unwrap);
   }
 
   @Override

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

@@ -21,6 +21,7 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.concurrent.Executors.callable;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.paging.Page;
@@ -36,14 +37,12 @@
 import com.google.cloud.Policy;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
-import com.google.cloud.storage.ApiaryUnbufferedReadableByteChannel.ApiaryReadRequest;
 import com.google.cloud.storage.BlobReadChannelV2.BlobReadChannelContext;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.PostPolicyV4.ConditionV4Type;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.PostPolicyV4.PostPolicyV4Document;
-import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
 import com.google.cloud.storage.UnifiedOpts.ObjectSourceOpt;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
@@ -60,10 +59,9 @@
 import com.google.common.collect.Maps;
 import com.google.common.hash.Hashing;
 import com.google.common.io.BaseEncoding;
-import com.google.common.io.ByteStreams;
+import com.google.common.io.CountingOutputStream;
 import com.google.common.primitives.Ints;
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -75,7 +73,6 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.text.SimpleDateFormat;
@@ -617,25 +614,9 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
     Opts<ObjectSourceOpt> unwrap = Opts.unwrap(options);
     Opts<ObjectSourceOpt> resolve = unwrap.resolveFrom(blob);
     ImmutableMap<StorageRpc.Option, ?> optionsMap = resolve.getRpcOptions();
-    boolean autoGzipDecompression =
-        Utils.isAutoGzipDecompression(resolve, /*defaultWhenUndefined=*/ true);
-    UnbufferedReadableByteChannelSession<StorageObject> session =
-        ResumableMedia.http()
-            .read()
-            .byteChannel(BlobReadChannelContext.from(this))
-            .setAutoGzipDecompression(autoGzipDecompression)
-            .unbuffered()
-            .setApiaryReadRequest(
-                new ApiaryReadRequest(storageObject, optionsMap, ByteRangeSpec.nullRange()))
-            .build();
-    ByteArrayOutputStream baos = new ByteArrayOutputStream();
-    try (UnbufferedReadableByteChannel r = session.open();
-        WritableByteChannel w = Channels.newChannel(baos)) {
-      ByteStreams.copy(r, w);
-    } catch (IOException e) {
-      throw StorageException.translate(e);
-    }
-    return baos.toByteArray();
+    ResultRetryAlgorithm<?> algorithm =
+        retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap);
+    return run(algorithm, () -> storageRpc.load(storageObject, optionsMap), Function.identity());
   }
 
   @Override
@@ -667,26 +648,19 @@ public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) {
 
   @Override
   public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
+    final CountingOutputStream countingOutputStream = new CountingOutputStream(outputStream);
     final StorageObject pb = codecs.blobId().encode(blob);
-    Opts<ObjectSourceOpt> resolve = Opts.unwrap(options).resolveFrom(blob);
-    ImmutableMap<StorageRpc.Option, ?> optionsMap = resolve.getRpcOptions();
-    boolean autoGzipDecompression =
-        Utils.isAutoGzipDecompression(resolve, /*defaultWhenUndefined=*/ true);
-    UnbufferedReadableByteChannelSession<StorageObject> session =
-        ResumableMedia.http()
-            .read()
-            .byteChannel(BlobReadChannelContext.from(this))
-            .setAutoGzipDecompression(autoGzipDecompression)
-            .unbuffered()
-            .setApiaryReadRequest(new ApiaryReadRequest(pb, optionsMap, ByteRangeSpec.nullRange()))
-            .build();
-    // don't close the provided stream
-    WritableByteChannel w = Channels.newChannel(outputStream);
-    try (UnbufferedReadableByteChannel r = session.open()) {
-      ByteStreams.copy(r, w);
-    } catch (IOException e) {
-      throw StorageException.translate(e);
-    }
+    ImmutableMap<StorageRpc.Option, ?> optionsMap =
+        Opts.unwrap(options).resolveFrom(blob).getRpcOptions();
+    ResultRetryAlgorithm<?> algorithm = retryAlgorithmManager.getForObjectsGet(pb, optionsMap);
+    run(
+        algorithm,
+        callable(
+            () -> {
+              storageRpc.read(
+                  pb, optionsMap, countingOutputStream.getCount(), countingOutputStream);
+            }),
+        Function.identity());
   }
 
   @Override

google-cloud-storage/src/test/java/com/google/cloud/storage/ITGrpcStorageImplUploadRetryTest.java

@@ -18,7 +18,6 @@
 
 import static com.google.cloud.storage.ByteSizeConstants._2MiB;
 import static com.google.common.truth.Truth.assertThat;
-import static org.junit.Assert.assertThrows;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
@@ -82,19 +81,16 @@ public void create_bytes() throws Exception {
 
   @Test
   public void create_inputStream() throws Exception {
-    Direct.FakeService service = Direct.FakeService.create();
+    Resumable.FakeService service = Resumable.FakeService.create();
     try (TmpFile tmpFile = DataGenerator.base64Characters().tempFile(baseDir, objectContentSize);
         FakeServer server = FakeServer.of(service);
         Storage s = server.getGrpcStorageOptions().getService();
         InputStream in = Channels.newInputStream(tmpFile.reader())) {
       BlobInfo info = BlobInfo.newBuilder("buck", "obj").build();
-      // create uses a direct upload, once the stream is consumed there is no means for us to retry
-      // if an error happens it should be surfaced
-      StorageException se =
-          assertThrows(
-              StorageException.class, () -> s.create(info, in, BlobWriteOption.doesNotExist()));
-      assertThat(se.getCode()).isEqualTo(500);
+      s.create(info, in, BlobWriteOption.doesNotExist());
     }
+
+    assertThat(service.returnError.get()).isFalse();
   }
 
   @Test