Repository: googleapis/java-storage
Pull request #733 (2021-02-25 23:23:05)

deps: update kms.version to v0.87.8
[![WhiteSource Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com)

This PR contains the following updates:

| Package | Change | Age | Adoption | Passing | Confidence |
|---|---|---|---|---|---|
| [com.google.api.grpc:proto-google-cloud-kms-v1](https://togithub.com/googleapis/java-kms) | `0.87.7` -> `0.87.8` | [![age](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.87.8/age-slim)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.87.8/adoption-slim)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.87.8/compatibility-slim/0.87.7)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:proto-google-cloud-kms-v1/0.87.8/confidence-slim/0.87.7)](https://docs.renovatebot.com/merge-confidence/) |
| [com.google.api.grpc:grpc-google-cloud-kms-v1](https://togithub.com/googleapis/java-kms) | `0.87.7` -> `0.87.8` | [![age](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:grpc-google-cloud-kms-v1/0.87.8/age-slim)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:grpc-google-cloud-kms-v1/0.87.8/adoption-slim)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:grpc-google-cloud-kms-v1/0.87.8/compatibility-slim/0.87.7)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://badges.renovateapi.com/packages/maven/com.google.api.grpc:grpc-google-cloud-kms-v1/0.87.8/confidence-slim/0.87.7)](https://docs.renovatebot.com/merge-confidence/) |

---

### Renovate configuration

:date: **Schedule**: At any time (no schedule defined).

:vertical_traffic_light: **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

:recycle: **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

:no_bell: **Ignore**: Close this PR and you won't be reminded about these updates again.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box

---

This PR has been generated by [WhiteSource Renovate](https://renovate.whitesourcesoftware.com). View repository job log [here](https://app.renovatebot.com/dashboard#github/googleapis/java-storage).

codecov[bot]: # [Codecov](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=h1) Report
> Merging [#733](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=desc) (c8bf3f8) into [master](https://codecov.io/gh/googleapis/java-storage/commit/c98413df9d9514340aed78b5a4d5e596760bb616?el=desc) (c98413d) will **decrease** coverage by `0.18%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/googleapis/java-storage/pull/733/graphs/tree.svg?width=650&height=150&src=pr&token=ssm139YOdp)](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #733      +/-   ##
============================================
- Coverage     64.33%   64.15%   -0.19%     
+ Complexity      630      620      -10     
============================================
  Files            32       32              
  Lines          5322     5300      -22     
  Branches        520      488      -32     
============================================
- Hits           3424     3400      -24     
+ Misses         1733     1731       -2     
- Partials        165      169       +4     
```


| [Impacted Files](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...in/java/com/google/cloud/storage/StorageBatch.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2UvU3RvcmFnZUJhdGNoLmphdmE=) | `88.00% <0.00%> (-4.00%)` | `13.00% <0.00%> (ø%)` | |
| [...gle/cloud/storage/testing/RemoteStorageHelper.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2UvdGVzdGluZy9SZW1vdGVTdG9yYWdlSGVscGVyLmphdmE=) | `63.41% <0.00%> (-0.59%)` | `9.00% <0.00%> (ø%)` | |
| [...main/java/com/google/cloud/storage/BucketInfo.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2UvQnVja2V0SW5mby5qYXZh) | `83.59% <0.00%> (-0.39%)` | `86.00% <0.00%> (ø%)` | |
| [...in/java/com/google/cloud/storage/PostPolicyV4.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2UvUG9zdFBvbGljeVY0LmphdmE=) | `86.11% <0.00%> (-0.23%)` | `5.00% <0.00%> (ø%)` | |
| [...ain/java/com/google/cloud/storage/StorageImpl.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2UvU3RvcmFnZUltcGwuamF2YQ==) | `81.18% <0.00%> (-0.14%)` | `138.00% <0.00%> (-9.00%)` | |
| [...e/src/main/java/com/google/cloud/storage/Blob.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2UvQmxvYi5qYXZh) | `83.61% <0.00%> (-0.10%)` | `29.00% <0.00%> (-1.00%)` | |
| [...om/google/cloud/storage/spi/v1/HttpStorageRpc.java](https://codecov.io/gh/googleapis/java-storage/pull/733/diff?src=pr&el=tree#diff-Z29vZ2xlLWNsb3VkLXN0b3JhZ2Uvc3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3N0b3JhZ2Uvc3BpL3YxL0h0dHBTdG9yYWdlUnBjLmphdmE=) | `1.56% <0.00%> (+<0.01%)` | `2.00% <0.00%> (ø%)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=footer). Last update [c98413d...ff0b786](https://codecov.io/gh/googleapis/java-storage/pull/733?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index eeb635e4..619544c4 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.7</kms.version>
+    <kms.version>0.87.8</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) {


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);