Repository: googleapis/java-storage (java)

Issue:
google cloud storage: NPE from BlobId.java:119
#### Environment details

1. google cloud storage
2. OS type and version: Ubuntu 18.04.4, running in docker container on k8s. 
3. Java version:
java version "1.8.0_241"
Java(TM) SE Runtime Environment (build 1.8.0_241-b07)
Java HotSpot(TM) 64-Bit Server VM (build 25.241-b07, mixed mode)
4. storage version(s):
google-cloud-storage-1.113.9.jar
google-cloud-core-1.94.0.jar
google-cloud-core-http-1.94.0.jar

#### Steps to reproduce

  1. Upload content per code below

#### Code example

```java
    BlobId blobId = BlobId.of(bucketName, key);
    String md5;
    try (InputStream inputStream = getInputStream()) {
      md5 = Base64.encodeBase64String(DigestUtils.md5(inputStream));
    }
    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType(getContentType())
      .setContentDisposition(String.format("attachment; filename=%s", getDownloadName()))
      .setMd5(md5)
      .build();
    storage.createFrom(blobInfo, getInputStream(), Storage.BlobWriteOption.md5Match());
```

#### Stack trace
```
Caused by: java.lang.NullPointerException
    at com.google.cloud.storage.BlobId.fromPb(BlobId.java:119)
    at com.google.cloud.storage.BlobInfo.fromPb(BlobInfo.java:1160)
    at com.google.cloud.storage.Blob.fromPb(Blob.java:958)
    at com.google.cloud.storage.StorageImpl.createFrom(StorageImpl.java:267)
    at com.google.cloud.storage.StorageImpl.createFrom(StorageImpl.java:253)
    at org.ambraproject.rhino.service.impl.GCSObjectStorageServiceImpl.storeFile(GCSObjectStorageServiceImpl.java:72)
    at org.ambraproject.rhino.service.impl.GCSObjectStorageServiceImpl.lambda$null$0(GCSObjectStorageServiceImpl.java:78)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
```

This is an intermittent issue - I’ve only seen it once, as far as I know. I can’t see how the NPE could be coming from our code. I can’t seem to find any similar issue in the issues list or online.

Thanks for any help you can provide!

Pull request:
fix: retrying get remote offset and recover from last chunk failures.
Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)


I added three things in this PR:
1. Retries around getting remote offset from GCS
2. Retrying last chunk failures correctly. 
3. Updated documentation for the retry cases


Fixes #709 #687 ☕️


Patch:
diff --git a/.github/readme/synth.metadata/synth.metadata b/.github/readme/synth.metadata/synth.metadata
index 8a9b167a..35c50eb6 100644
--- a/.github/readme/synth.metadata/synth.metadata
+++ b/.github/readme/synth.metadata/synth.metadata
@@ -4,14 +4,14 @@
       "git": {
         "name": ".",
         "remote": "https://github.com/googleapis/java-storage.git",
-        "sha": "6b1f0361376167719ec5456181134136d27d1d3c"
+        "sha": "4c1a739e7314eddc73cfb9df004e65a10a614086"
       }
     },
     {
       "git": {
         "name": "synthtool",
         "remote": "https://github.com/googleapis/synthtool.git",
-        "sha": "79ab0b44a2cc7d803d07c107f9faf07729fc4012"
+        "sha": "0199c79b8324fba66476300824aa931788c47e2d"
       }
     }
   ]
diff --git a/README.md b/README.md
index f06e1b30..98325732 100644
--- a/README.md
+++ b/README.md
@@ -17,7 +17,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>16.4.0</version>
+      <version>18.0.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -38,14 +38,14 @@ If you are using Maven without BOM, add this to your dependencies:
 <dependency>
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage</artifactId>
-  <version>1.113.10</version>
+  <version>1.113.11</version>
 </dependency>
 
 ```
 
 If you are using Gradle 5.x or later, add this to your dependencies
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:16.4.0')
+implementation platform('com.google.cloud:libraries-bom:18.0.0')
 
 compile 'com.google.cloud:google-cloud-storage'
 ```
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 8adbbd42..eeb635e4 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -16,7 +16,7 @@
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
-    <kms.version>0.87.6</kms.version>
+    <kms.version>0.87.7</kms.version>
   </properties>
   <dependencies>
     <dependency>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
index ca87933c..aa5c3a81 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
@@ -55,6 +55,7 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
   // TODO: I don't think this is thread safe, and there's probably a better way to detect a retry
   // occuring.
   private boolean retrying = false;
+  private boolean checkingForLastChunk = false;
 
   boolean isRetrying() {
     return retrying;
@@ -64,129 +65,141 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
     return storageObject;
   }
 
+  private StorageObject transmitChunk(
+      int chunkOffset, int chunkLength, long position, boolean last) {
+    return getOptions()
+        .getStorageRpcV1()
+        .writeWithResponse(getUploadId(), getBuffer(), chunkOffset, position, chunkLength, last);
+  }
+
+  private long getRemotePosition() {
+    return getOptions().getStorageRpcV1().getCurrentUploadOffset(getUploadId());
+  }
+
+  private StorageObject getRemoteStorageObject() {
+    return getOptions().getStorageRpcV1().get(getEntity().toPb(), null);
+  }
+
+  private StorageException unrecoverableState(
+      int chunkOffset, int chunkLength, long localPosition, long remotePosition, boolean last) {
+    StringBuilder sb = new StringBuilder();
+    sb.append("Unable to recover in upload.\n");
+    sb.append(
+        "This may be a symptom of multiple clients uploading to the same upload session.\n\n");
+    sb.append("For debugging purposes:\n");
+    sb.append("uploadId: ").append(getUploadId()).append('\n');
+    sb.append("chunkOffset: ").append(chunkOffset).append('\n');
+    sb.append("chunkLength: ").append(chunkLength).append('\n');
+    sb.append("localOffset: ").append(localPosition).append('\n');
+    sb.append("remoteOffset: ").append(remotePosition).append('\n');
+    sb.append("lastChunk: ").append(last).append("\n\n");
+    return new StorageException(0, sb.toString());
+  }
+
+  // Retriable interruption occurred.
+  // Variables:
+  // chunk = getBuffer()
+  // localNextByteOffset == getPosition()
+  // chunkSize = getChunkSize()
+  //
+  // Case 1: localNextByteOffset == remoteNextByteOffset:
+  // Retrying the entire chunk
+  //
+  // Case 2: localNextByteOffset < remoteNextByteOffset
+  //             && driftOffset < chunkSize:
+  // Upload progressed and localNextByteOffset is not in-sync with
+  // remoteNextByteOffset and driftOffset is less than chunkSize.
+  // driftOffset must be less than chunkSize for it to retry using
+  // chunk maintained in memory.
+  // Find the driftOffset by subtracting localNextByteOffset from
+  // remoteNextByteOffset.
+  // Use driftOffset to determine where to restart from using the chunk in
+  // memory.
+  //
+  // Case 3: localNextByteOffset < remoteNextByteOffset
+  //            && driftOffset == chunkSize:
+  // Special case of Case 2.
+  // If chunkSize is equal to driftOffset then remoteNextByteOffset has moved on
+  // to the next chunk.
+  //
+  // Case 4: localNextByteOffset < remoteNextByteOffset
+  //            && driftOffset > chunkSize:
+  // Throw exception as remoteNextByteOffset has drifted beyond the retriable
+  // chunk maintained in memory. This is not possible unless there's multiple
+  // clients uploading to the same resumable upload session.
+  //
+  // Case 5: localNextByteOffset > remoteNextByteOffset:
+  // For completeness, this case is not possible because it would require retrying
+  // a 400 status code which is not allowed.
+  //
+  // Case 6: remoteNextByteOffset==-1 && last == true
+  // Upload is complete and retry occurred in the "last" chunk. Data sent was
+  // received by the service.
+  //
+  // Case 7: remoteNextByteOffset==-1 && last == false && !checkingForLastChunk
+  // Not last chunk and are not checkingForLastChunk, allow for the client to
+  // catch up to final chunk which meets
+  // Case 6.
+  //
+  // Case 8: remoteNextByteOffset==-1 && last == false && checkingForLastChunk
+  // Not last chunk and checkingForLastChunk means this is the second time we
+  // hit this case, meaning the upload was completed by a different client.
+  //
+  // Case 9: Only possible if the client local offset continues beyond the remote
+  // offset which is not possible.
+  //
   @Override
-  protected void flushBuffer(final int length, final boolean last) {
+  protected void flushBuffer(final int length, final boolean lastChunk) {
     try {
       runWithRetries(
           callable(
               new Runnable() {
                 @Override
                 public void run() {
+                  // Get remote offset from API
+                  final long localPosition = getPosition();
+                  // For each request it should be possible to retry from its location in this code
+                  final long remotePosition = isRetrying() ? getRemotePosition() : getPosition();
+                  final int chunkOffset = (int) (remotePosition - localPosition);
+                  final int chunkLength = length - chunkOffset;
+                  final boolean uploadAlreadyComplete = remotePosition == -1;
+                  // Enable isRetrying state to reduce number of calls to getRemotePosition()
                   if (!isRetrying()) {
-                    // Enable isRetrying state to reduce number of calls to getCurrentUploadOffset()
                     retrying = true;
+                  }
+                  if (uploadAlreadyComplete && lastChunk) {
+                    // Case 6
+                    // Request object metadata if not available
+                    if (storageObject == null) {
+                      storageObject = getRemoteStorageObject();
+                    }
+                    // Verify that with the final chunk we match the blob length
+                    if (storageObject.getSize().longValue() != getPosition() + length) {
+                      throw unrecoverableState(
+                          chunkOffset, chunkLength, localPosition, remotePosition, lastChunk);
+                    }
+                    retrying = false;
+                  } else if (uploadAlreadyComplete && !lastChunk && !checkingForLastChunk) {
+                    // Case 7
+                    // Make sure this is the second to last chunk.
+                    checkingForLastChunk = true;
+                    // Continue onto next chunk in case this is the last chunk
+                  } else if (localPosition <= remotePosition && chunkOffset < getChunkSize()) {
+                    // Case 1 && Case 2
+                    // We are in a position to send a chunk
                     storageObject =
-                        getOptions()
-                            .getStorageRpcV1()
-                            .writeWithResponse(
-                                getUploadId(), getBuffer(), 0, getPosition(), length, last);
+                        transmitChunk(chunkOffset, chunkLength, remotePosition, lastChunk);
+                    retrying = false;
+                  } else if (localPosition < remotePosition && chunkOffset == getChunkSize()) {
+                    // Case 3
+                    // Continue to next chunk to catch up with remotePosition we are one chunk
+                    // behind
+                    retrying = false;
                   } else {
-                    // Retriable interruption occurred.
-                    // Variables:
-                    // chunk = getBuffer()
-                    // localNextByteOffset == getPosition()
-                    // chunkSize = getChunkSize()
-                    //
-                    // Case 1: localNextByteOffset == 0 && remoteNextByteOffset == 0:
-                    // we are retrying from first chunk start from 0 offset.
-                    //
-                    // Case 2: localNextByteOffset == remoteNextByteOffset:
-                    // Special case of Case 1 when a chunk is retried.
-                    //
-                    // Case 3: localNextByteOffset < remoteNextByteOffset
-                    //             && driftOffset < chunkSize:
-                    // Upload progressed and localNextByteOffset is not in-sync with
-                    // remoteNextByteOffset and driftOffset is less than chunkSize.
-                    // driftOffset must be less than chunkSize for it to retry using
-                    // chunk maintained in memory.
-                    // Find the driftOffset by subtracting localNextByteOffset from
-                    // remoteNextByteOffset.
-                    // Use driftOffset to determine where to restart from using the chunk in
-                    // memory.
-                    //
-                    // Case 4: localNextByteOffset < remoteNextByteOffset
-                    //            && driftOffset == chunkSize:
-                    // Special case of Case 3.
-                    // If chunkSize is equal to driftOffset then remoteNextByteOffset has moved on
-                    // to the next chunk.
-                    //
-                    // Case 5: localNextByteOffset < remoteNextByteOffset
-                    //            && driftOffset > chunkSize:
-                    // Throw exception as remoteNextByteOffset has drifted beyond the retriable
-                    // chunk maintained in memory. This is not possible unless there's multiple
-                    // clients uploading to the same resumable upload session.
-                    //
-                    // Case 6: localNextByteOffset > remoteNextByteOffset:
-                    // For completeness, this case is not possible because it would require retrying
-                    // a 400 status code which is not allowed.
-                    //
-                    // Case 7: remoteNextByteOffset==-1 && last == true
-                    // Upload is complete and retry occurred in the "last" chunk. Data sent was
-                    // received by the service.
-                    //
-                    // Case 8: remoteNextByteOffset==-1 && last == false
-                    // Upload was completed by another client because this retry did not occur
-                    // during the last chunk.
-                    //
-                    // Get remote offset from API
-                    long remoteNextByteOffset =
-                        getOptions().getStorageRpcV1().getCurrentUploadOffset(getUploadId());
-                    long localNextByteOffset = getPosition();
-                    int driftOffset = (int) (remoteNextByteOffset - localNextByteOffset);
-                    int retryChunkLength = length - driftOffset;
-
-                    if (localNextByteOffset == 0 && remoteNextByteOffset == 0
-                        || localNextByteOffset == remoteNextByteOffset) {
-                      // Case 1 and 2
-                      storageObject =
-                          getOptions()
-                              .getStorageRpcV1()
-                              .writeWithResponse(
-                                  getUploadId(), getBuffer(), 0, getPosition(), length, last);
-                    } else if (localNextByteOffset < remoteNextByteOffset
-                        && driftOffset < getChunkSize()) {
-                      // Case 3
-                      storageObject =
-                          getOptions()
-                              .getStorageRpcV1()
-                              .writeWithResponse(
-                                  getUploadId(),
-                                  getBuffer(),
-                                  driftOffset,
-                                  remoteNextByteOffset,
-                                  retryChunkLength,
-                                  last);
-                    } else if (localNextByteOffset < remoteNextByteOffset
-                        && driftOffset == getChunkSize()) {
-                      // Case 4
-                      // Continue to next chunk
-                      retrying = false;
-                      return;
-                    } else if (localNextByteOffset < remoteNextByteOffset
-                        && driftOffset > getChunkSize()) {
-                      // Case 5
-                      StringBuilder sb = new StringBuilder();
-                      sb.append(
-                          "Remote offset has progressed beyond starting byte offset of next chunk.");
-                      sb.append(
-                          "This may be a symptom of multiple clients uploading to the same upload session.\n\n");
-                      sb.append("For debugging purposes:\n");
-                      sb.append("uploadId: ").append(getUploadId()).append('\n');
-                      sb.append("localNextByteOffset: ").append(localNextByteOffset).append('\n');
-                      sb.append("remoteNextByteOffset: ").append(remoteNextByteOffset).append('\n');
-                      sb.append("driftOffset: ").append(driftOffset).append("\n\n");
-                      throw new StorageException(0, sb.toString());
-                    } else if (remoteNextByteOffset == -1 && last) {
-                      // Case 7
-                      retrying = false;
-                      return;
-                    } else if (remoteNextByteOffset == -1 && !last) {
-                      // Case 8
-                      throw new StorageException(0, "Resumable upload is already complete.");
-                    }
+                    // Case 4 && Case 8 && Case 9
+                    throw unrecoverableState(
+                        chunkOffset, chunkLength, localPosition, remotePosition, lastChunk);
                   }
-                  // Request was successful and retrying state is now disabled.
-                  retrying = false;
                 }
               }),
           getOptions().getRetrySettings(),
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
index 74977dc6..6df86cb6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -32,6 +32,7 @@ import com.google.api.client.http.HttpRequestFactory;
 import com.google.api.client.http.HttpRequestInitializer;
 import com.google.api.client.http.HttpResponse;
 import com.google.api.client.http.HttpResponseException;
+import com.google.api.client.http.HttpStatusCodes;
 import com.google.api.client.http.HttpTransport;
 import com.google.api.client.http.InputStreamContent;
 import com.google.api.client.http.json.JsonHttpContent;
@@ -765,7 +766,8 @@ public class HttpStorageRpc implements StorageRpc {
       try {
         response = httpRequest.execute();
         int code = response.getStatusCode();
-        if (code == 201 || code == 200) {
+        if (HttpStatusCodes.isSuccess(code)) {
+          // Upload completed successfully
           return -1;
         }
         StringBuilder sb = new StringBuilder();
@@ -774,20 +776,18 @@ public class HttpStorageRpc implements StorageRpc {
         throw new StorageException(0, sb.toString());
       } catch (HttpResponseException ex) {
         int code = ex.getStatusCode();
-        if (code == 308 && ex.getHeaders().getRange() == null) {
-          // No progress has been made.
-          return 0;
-        } else if (code == 308 && ex.getHeaders().getRange() != null) {
+        if (code == 308) {
+          if (ex.getHeaders().getRange() == null) {
+            // No progress has been made.
+            return 0;
+          }
           // API returns last byte received offset
           String range = ex.getHeaders().getRange();
           // Return next byte offset by adding 1 to last byte received offset
           return Long.parseLong(range.substring(range.indexOf("-") + 1)) + 1;
         } else {
-          // Not certain what went wrong
-          StringBuilder sb = new StringBuilder();
-          sb.append("Not sure what occurred. Here's debugging information:\n");
-          sb.append("Response:\n").append(ex.toString()).append("\n\n");
-          throw new StorageException(0, sb.toString());
+          // Something else occurred like a 5xx so translate and throw.
+          throw translate(ex);
         }
       } finally {
         if (response != null) {
diff --git a/pom.xml b/pom.xml
index 4959b62b..92c5ac21 100644
--- a/pom.xml
+++ b/pom.xml
@@ -14,7 +14,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-shared-config</artifactId>
-    <version>0.10.0</version>
+    <version>0.11.0</version>
   </parent>
 
   <developers>
@@ -63,7 +63,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <github.global.server>github</github.global.server>
     <site.installationModule>google-cloud-storage-parent</site.installationModule>
-    <google.cloud.shared-dependencies.version>0.19.0</google.cloud.shared-dependencies.version>
+    <google.cloud.shared-dependencies.version>0.20.0</google.cloud.shared-dependencies.version>
   </properties>
 
   <dependencyManagement>
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 6e2f3c7b..02f4f5e9 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -29,7 +29,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>1.113.10</version>
+      <version>1.113.11</version>
     </dependency>
     <!-- [END storage_install_without_bom] -->
 
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index 749bc35b..76a81652 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -28,7 +28,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>1.113.10</version>
+      <version>1.113.11</version>
     </dependency>
 
     <dependency>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index f67951b5..fb546e66 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -30,7 +30,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>16.4.0</version>
+        <version>18.0.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java
index eefee772..ba60a3c4 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java
@@ -41,6 +41,7 @@ import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
+import java.math.BigInteger;
 import java.net.MalformedURLException;
 import java.net.SocketException;
 import java.net.URL;
@@ -69,7 +70,8 @@ public class BlobWriteChannelTest {
   private static final Random RANDOM = new Random();
   private static final String SIGNED_URL =
       "http://www.test.com/test-bucket/test1.txt?GoogleAccessId=testClient-test@test.com&Expires=1553839761&Signature=MJUBXAZ7";
-
+  private static final StorageException socketClosedException =
+      new StorageException(new SocketException("Socket closed"));
   private StorageOptions options;
   private StorageRpcFactory rpcFactoryMock;
   private StorageRpc storageRpcMock;
@@ -104,8 +106,8 @@ public class BlobWriteChannelTest {
 
   @Test
   public void testCreateRetryableError() {
-    StorageException exception = new StorageException(new SocketException("Socket closed"));
-    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andThrow(exception);
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS))
+        .andThrow(socketClosedException);
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     replay(storageRpcMock);
     writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
@@ -136,7 +138,6 @@ public class BlobWriteChannelTest {
 
   @Test
   public void testWriteWithFlushRetryChunk() throws IOException {
-    StorageException exception = new StorageException(new SocketException("Socket closed"));
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -148,7 +149,7 @@ public class BlobWriteChannelTest {
                 eq(0L),
                 eq(MIN_CHUNK_SIZE),
                 eq(false)))
-        .andThrow(exception);
+        .andThrow(socketClosedException);
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(0L);
     expect(
             storageRpcMock.writeWithResponse(
@@ -169,8 +170,45 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithFlushRetryChunkWithDrift() throws IOException {
-    StorageException exception = new StorageException(new SocketException("Socket closed"));
+  public void testWriteWithRetryFullChunk() throws IOException {
+    ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
+    Capture<byte[]> capturedBuffer = Capture.newInstance();
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID), (byte[]) anyObject(), eq(0), eq(0L), eq(MIN_CHUNK_SIZE), eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(0L);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andReturn(null);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                (byte[]) anyObject(),
+                eq(0),
+                eq((long) MIN_CHUNK_SIZE),
+                eq(0),
+                eq(true)))
+        .andReturn(BLOB_INFO.toPb());
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
+    assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
+    writer.close();
+    assertFalse(writer.isOpen());
+    assertNotNull(writer.getStorageObject());
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+  }
+
+  @Test
+  public void testWriteWithRemoteProgressMade() throws IOException {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -182,7 +220,8 @@ public class BlobWriteChannelTest {
                 eq(0L),
                 eq(MIN_CHUNK_SIZE),
                 eq(false)))
-        .andThrow(exception);
+        .andThrow(socketClosedException);
+    // Simulate GCS received 10 bytes but not the rest of the chunk
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(10L);
     expect(
             storageRpcMock.writeWithResponse(
@@ -203,8 +242,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithLastFlushRetryChunkButCompleted() throws IOException {
-    StorageException exception = new StorageException(new SocketException("Socket closed"));
+  public void testWriteWithDriftRetryCase4() throws IOException {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -215,25 +253,102 @@ public class BlobWriteChannelTest {
                 eq(0),
                 eq(0L),
                 eq(MIN_CHUNK_SIZE),
-                eq(true)))
-        .andThrow(exception);
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn((long) MIN_CHUNK_SIZE);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq((long) MIN_CHUNK_SIZE),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andReturn(null);
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
+    assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+    capturedBuffer.reset();
+    buffer.rewind();
+    assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+    assertTrue(writer.isOpen());
+    assertNull(writer.getStorageObject());
+  }
+
+  @Test
+  public void testWriteWithUnreachableRemoteOffset() throws IOException {
+    ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
+    Capture<byte[]> capturedBuffer = Capture.newInstance();
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(MIN_CHUNK_SIZE + 10L);
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
+    try {
+      writer.write(buffer);
+      fail("Expected StorageException");
+    } catch (StorageException storageException) {
+      // expected storageException
+    }
+    assertTrue(writer.isOpen());
+    assertNull(writer.getStorageObject());
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+  }
+
+  @Test
+  public void testWriteWithRetryAndObjectMetadata() throws IOException {
+    ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
+    Capture<byte[]> capturedBuffer = Capture.newInstance();
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(10L);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(10),
+                eq(10L),
+                eq(MIN_CHUNK_SIZE - 10),
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
+    expect(storageRpcMock.get(BLOB_INFO.toPb(), null)).andThrow(socketClosedException);
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
+    expect(storageRpcMock.get(BLOB_INFO.toPb(), null))
+        .andReturn(BLOB_INFO.toPb().setSize(BigInteger.valueOf(MIN_CHUNK_SIZE)));
     replay(storageRpcMock);
     writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     writer.close();
-    assertFalse(writer.isRetrying());
     assertFalse(writer.isOpen());
-    // Capture captures entire buffer of a chunk even when not completely used.
-    // Making assert selective up to the size of MIN_CHUNK_SIZE
-    assertArrayEquals(Arrays.copyOf(capturedBuffer.getValue(), MIN_CHUNK_SIZE), buffer.array());
+    assertNotNull(writer.getStorageObject());
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
   }
 
   @Test
-  public void testWriteWithFlushRetryChunkButCompletedByAnotherClient() throws IOException {
-    StorageException exception = new StorageException(new SocketException("Socket closed"));
-    StorageException completedException =
-        new StorageException(0, "Resumable upload is already complete.");
+  public void testWriteWithUploadCompletedByAnotherClient() throws IOException {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -245,21 +360,145 @@ public class BlobWriteChannelTest {
                 eq(0L),
                 eq(MIN_CHUNK_SIZE),
                 eq(false)))
-        .andThrow(exception);
+        .andReturn(null);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq((long) MIN_CHUNK_SIZE),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
     replay(storageRpcMock);
     writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
     writer.setChunkSize(MIN_CHUNK_SIZE);
     try {
       writer.write(buffer);
+      buffer.rewind();
+      writer.write(buffer);
+      buffer.rewind();
+      writer.write(buffer);
+      fail("Expected completed exception.");
+    } catch (StorageException ex) {
+
+    }
+    assertTrue(writer.isOpen());
+  }
+
+  @Test
+  public void testWriteWithLocalOffsetGoingBeyondRemoteOffset() throws IOException {
+    ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
+    Capture<byte[]> capturedBuffer = Capture.newInstance();
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andReturn(null);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq((long) MIN_CHUNK_SIZE),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(0L);
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
+    try {
+      writer.write(buffer);
+      buffer.rewind();
+      writer.write(buffer);
+      writer.close();
       fail("Expected completed exception.");
     } catch (StorageException ex) {
-      assertEquals(ex, completedException);
     }
-    assertTrue(writer.isRetrying());
     assertTrue(writer.isOpen());
   }
 
+  @Test
+  public void testGetCurrentUploadOffset() throws IOException {
+    ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
+    Capture<byte[]> capturedBuffer = Capture.newInstance();
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(0L);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andReturn(null);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                (byte[]) anyObject(),
+                eq(0),
+                eq((long) MIN_CHUNK_SIZE),
+                eq(0),
+                eq(true)))
+        .andReturn(BLOB_INFO.toPb());
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
+    assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
+    writer.close();
+    assertFalse(writer.isOpen());
+    assertNotNull(writer.getStorageObject());
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+  }
+
+  @Test
+  public void testWriteWithLastFlushRetryChunkButCompleted() throws IOException {
+    ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
+    Capture<byte[]> capturedBuffer = Capture.newInstance();
+    expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(true)))
+        .andThrow(socketClosedException);
+    expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
+    expect(storageRpcMock.get(BLOB_INFO.toPb(), null))
+        .andReturn(BLOB_INFO.toPb().setSize(BigInteger.valueOf(MIN_CHUNK_SIZE)));
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
+    writer.close();
+    assertFalse(writer.isRetrying());
+    assertFalse(writer.isOpen());
+    assertNotNull(writer.getStorageObject());
+    // Capture captures entire buffer of a chunk even when not completely used.
+    // Making assert selective up to the size of MIN_CHUNK_SIZE
+    assertArrayEquals(Arrays.copyOf(capturedBuffer.getValue(), MIN_CHUNK_SIZE), buffer.array());
+  }
+
   @Test
   public void testWriteWithFlush() throws IOException {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);