Repository: googleapis/java-storage
Pull request #621 (2020-11-13 18:27:24)

chore: regenerate README
This PR was generated using Autosynth. :rainbow:


<details><summary>Log from Synthtool</summary>

```
2020-11-13 18:27:18,931 synthtool [DEBUG] > Executing /root/.cache/synthtool/java-storage/.github/readme/synth.py.
On branch autosynth-readme
nothing to commit, working directory clean
2020-11-13 18:27:19,903 synthtool [DEBUG] > Wrote metadata to .github/readme/synth.metadata/synth.metadata.

```
</details>

Full log will be available here:
https://source.cloud.google.com/results/invocations/f0dd7465-c636-4368-8b08-9fc0d9ffd1f7/targets

- [ ] To automatically regenerate this PR, check this box.

codecov[bot]: # [Codecov](https://codecov.io/gh/googleapis/java-storage/pull/621?src=pr&el=h1) Report
> Merging [#621](https://codecov.io/gh/googleapis/java-storage/pull/621?src=pr&el=desc) (308e7c5) into [master](https://codecov.io/gh/googleapis/java-storage/commit/48e404fedd5450a63b8ea25017acfaac4c6d5849?el=desc) (48e404f) will **not change** coverage.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff            @@
##             master     #621   +/-   ##
=========================================
  Coverage     64.68%   64.68%           
+ Complexity      629      620    -9     
=========================================
  Files            32       32           
  Lines          5255     5255           
  Branches        510      511    +1     
=========================================
  Hits           3399     3399           
  Misses         1697     1697           
  Partials        159      159           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/googleapis/java-storage/pull/621?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/621?src=pr&el=footer). Last update [48e404f...308e7c5](https://codecov.io/gh/googleapis/java-storage/pull/621?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/.github/readme/synth.metadata/synth.metadata b/.github/readme/synth.metadata/synth.metadata
index 4a44631a..f8683da8 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": "3ac92c97d371756d7abe0c56419092047b4a4c1b"
+        "sha": "48e404fedd5450a63b8ea25017acfaac4c6d5849"
       }
     },
     {
       "git": {
         "name": "synthtool",
         "remote": "https://github.com/googleapis/synthtool.git",
-        "sha": "7db8a6c5ffb12a6e4c2f799c18f00f7f3d60e279"
+        "sha": "d5fc0bcf9ea9789c5b0e3154a9e3b29e5cea6116"
       }
     }
   ]
diff --git a/.github/workflows/approve-readme.yaml b/.github/workflows/approve-readme.yaml
index e2d841d6..7513acae 100644
--- a/.github/workflows/approve-readme.yaml
+++ b/.github/workflows/approve-readme.yaml
@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     if: github.repository_owner == 'googleapis' && github.head_ref == 'autosynth-readme'
     steps:
-      - uses: actions/github-script@v3.0.0
+      - uses: actions/github-script@v3
         with:
           github-token: ${{secrets.YOSHI_APPROVER_TOKEN}}
           script: |
diff --git a/.github/workflows/auto-release.yaml b/.github/workflows/auto-release.yaml
index bc1554ae..2b6cdbc9 100644
--- a/.github/workflows/auto-release.yaml
+++ b/.github/workflows/auto-release.yaml
@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     if: contains(github.head_ref, 'release-v')
     steps:
-    - uses: actions/github-script@v3.0.0
+    - uses: actions/github-script@v3
       with:
         github-token: ${{secrets.YOSHI_APPROVER_TOKEN}}
         debug: true
diff --git a/.github/workflows/formatting.yaml b/.github/workflows/formatting.yaml
index d4d367cf..6844407b 100644
--- a/.github/workflows/formatting.yaml
+++ b/.github/workflows/formatting.yaml
@@ -18,7 +18,7 @@ jobs:
         with:
           java-version: 11
       - run: "mvn com.coveo:fmt-maven-plugin:format"
-      - uses: googleapis/code-suggester@v1.8.0
+      - uses: googleapis/code-suggester@v1
         with:
           command: review
           pull_number: ${{ github.event.pull_request.number }}
diff --git a/.kokoro/common.sh b/.kokoro/common.sh
index a8d0ea04..ace89f45 100644
--- a/.kokoro/common.sh
+++ b/.kokoro/common.sh
@@ -25,7 +25,6 @@ function retry_with_backoff {
     
     # allow a failures to continue
     set +e
-    echo "${command}"
     ${command}
     exit_code=$?
 
diff --git a/.kokoro/release/stage.sh b/.kokoro/release/stage.sh
index 3c482cbc..16126d2e 100755
--- a/.kokoro/release/stage.sh
+++ b/.kokoro/release/stage.sh
@@ -20,19 +20,22 @@ python3 -m pip install gcp-releasetool
 python3 -m releasetool publish-reporter-script > /tmp/publisher-script; source /tmp/publisher-script
 
 source $(dirname "$0")/common.sh
+source $(dirname "$0")/../common.sh
 MAVEN_SETTINGS_FILE=$(realpath $(dirname "$0")/../../)/settings.xml
 pushd $(dirname "$0")/../../
 
 setup_environment_secrets
 create_settings_xml_file "settings.xml"
 
-mvn clean install deploy -B \
-  --settings ${MAVEN_SETTINGS_FILE} \
-  -DskipTests=true \
-  -DperformRelease=true \
-  -Dgpg.executable=gpg \
-  -Dgpg.passphrase=${GPG_PASSPHRASE} \
-  -Dgpg.homedir=${GPG_HOMEDIR}
+# attempt to stage 3 times with exponential backoff (starting with 10 seconds)
+retry_with_backoff 3 10 \
+  mvn clean install deploy -B \
+    --settings ${MAVEN_SETTINGS_FILE} \
+    -DskipTests=true \
+    -DperformRelease=true \
+    -Dgpg.executable=gpg \
+    -Dgpg.passphrase=${GPG_PASSPHRASE} \
+    -Dgpg.homedir=${GPG_HOMEDIR}
 
 if [[ -n "${AUTORELEASE_PR}" ]]
 then
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c7bb999d..0ccaff6c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,17 @@
 # Changelog
 
+### [1.113.4](https://www.github.com/googleapis/java-storage/compare/v1.113.3...v1.113.4) (2020-11-13)
+
+
+### Bug Fixes
+
+* retry using remote offset ([#604](https://www.github.com/googleapis/java-storage/issues/604)) ([216b52c](https://www.github.com/googleapis/java-storage/commit/216b52c54d34eaf1307788809a3512c461adf381))
+
+
+### Dependencies
+
+* update dependency com.google.cloud:google-cloud-shared-dependencies to v0.15.0 ([#610](https://www.github.com/googleapis/java-storage/issues/610)) ([ac65e5b](https://www.github.com/googleapis/java-storage/commit/ac65e5b0bd324d5726504bb3405c758675a56ddc))
+
 ### [1.113.3](https://www.github.com/googleapis/java-storage/compare/v1.113.2...v1.113.3) (2020-11-06)
 
 
diff --git a/README.md b/README.md
index 4b797d17..6343b8ce 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>15.0.0</version>
+      <version>15.1.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
diff --git a/google-cloud-storage/clirr-ignored-differences.xml b/google-cloud-storage/clirr-ignored-differences.xml
index ba97a975..02722fac 100644
--- a/google-cloud-storage/clirr-ignored-differences.xml
+++ b/google-cloud-storage/clirr-ignored-differences.xml
@@ -26,4 +26,9 @@
         <method>com.google.cloud.storage.BucketInfo$Builder setUpdateTime(java.lang.Long)</method>
         <differenceType>7013</differenceType>
     </difference>
+    <difference>
+        <className>com/google/cloud/storage/spi/v1/StorageRpc</className>
+        <method>long getCurrentUploadOffset(java.lang.String)</method>
+        <differenceType>7012</differenceType>
+    </difference>
 </differences>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 8d2a9675..7a1b2226 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -2,7 +2,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <artifactId>google-cloud-storage</artifactId>
-  <version>1.113.4-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>1.113.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <packaging>jar</packaging>
   <name>Google Cloud Storage</name>
   <url>https://github.com/googleapis/java-storage</url>
@@ -12,11 +12,11 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-storage-parent</artifactId>
-    <version>1.113.4-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+    <version>1.113.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   </parent>
   <properties>
     <site.installationModule>google-cloud-storage</site.installationModule>
-    <kms.version>0.87.1</kms.version>
+    <kms.version>0.87.2</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 0c952084..2e9c0a38 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
@@ -51,6 +51,15 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
   // Contains metadata of the updated object or null if upload is not completed.
   private StorageObject storageObject;
 
+  // Detect if flushBuffer() is being retried or not.
+  // 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;
+
+  boolean isRetrying() {
+    return retrying;
+  }
+
   StorageObject getStorageObject() {
     return storageObject;
   }
@@ -63,11 +72,105 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
               new Runnable() {
                 @Override
                 public void run() {
-                  storageObject =
-                      getOptions()
-                          .getStorageRpcV1()
-                          .writeWithResponse(
-                              getUploadId(), getBuffer(), 0, getPosition(), length, last);
+                  if (!isRetrying()) {
+                    // Enable isRetrying state to reduce number of calls to getCurrentUploadOffset()
+                    retrying = true;
+                    storageObject =
+                        getOptions()
+                            .getStorageRpcV1()
+                            .writeWithResponse(
+                                getUploadId(), getBuffer(), 0, getPosition(), length, last);
+                  } 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.
+                    //
+                    // 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 {
+                      // 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());
+                    }
+                  }
+                  // 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 0960f91f..faba82b1 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
@@ -747,6 +747,51 @@ public class HttpStorageRpc implements StorageRpc {
     writeWithResponse(uploadId, toWrite, toWriteOffset, destOffset, length, last);
   }
 
+  @Override
+  public long getCurrentUploadOffset(String uploadId) {
+    try {
+      GenericUrl url = new GenericUrl(uploadId);
+      HttpRequest httpRequest = storage.getRequestFactory().buildPutRequest(url, null);
+
+      httpRequest.getHeaders().setContentRange("bytes */*");
+      // Turn off automatic redirects.
+      // HTTP 308 are returned if upload is incomplete.
+      // See: https://cloud.google.com/storage/docs/performing-resumable-uploads
+      httpRequest.setFollowRedirects(false);
+
+      HttpResponse response = null;
+      try {
+        response = httpRequest.execute();
+        int code = response.getStatusCode();
+        String message = response.getStatusMessage();
+        if (code == 201 || code == 200) {
+          throw new StorageException(0, "Resumable upload is already complete.");
+        }
+        StringBuilder sb = new StringBuilder();
+        sb.append("Not sure what occurred. Here's debugging information:\n");
+        sb.append("Response:\n").append(response.toString()).append("\n\n");
+        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 {
+          // 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;
+        }
+      } finally {
+        if (response != null) {
+          response.disconnect();
+        }
+      }
+    } catch (IOException ex) {
+      throw translate(ex);
+    }
+  }
+
   @Override
   public StorageObject writeWithResponse(
       String uploadId,
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
index 2088c15b..cc4dd574 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
@@ -328,6 +328,15 @@ public interface StorageRpc extends ServiceRpc {
       int length,
       boolean last);
 
+  /**
+   * Requests current byte offset from Cloud Storage API. Used to recover from a failure in some
+   * bytes were committed successfully to the open resumable session.
+   *
+   * @param uploadId resumable upload ID URL
+   * @throws StorageException upon failure
+   */
+  long getCurrentUploadOffset(String uploadId);
+
   /**
    * Writes the provided bytes to a storage object at the provided location. If {@code last=true}
    * returns metadata of the updated object, otherwise returns null.
diff --git a/pom.xml b/pom.xml
index a09b4b13..010b03fe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
   <groupId>com.google.cloud</groupId>
   <artifactId>google-cloud-storage-parent</artifactId>
   <packaging>pom</packaging>
-  <version>1.113.4-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
+  <version>1.113.5-SNAPSHOT</version><!-- {x-version-update:google-cloud-storage:current} -->
   <name>Storage Parent</name>
   <url>https://github.com/googleapis/java-storage</url>
   <description>
@@ -79,7 +79,7 @@
       <dependency>
         <groupId>com.google.apis</groupId>
         <artifactId>google-api-services-storage</artifactId>
-        <version>v1-rev20200927-1.30.10</version>
+        <version>v1-rev20201112-1.30.10</version>
       </dependency>
       <dependency>
         <groupId>org.easymock</groupId>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 7910681e..71b94ba2 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>15.1.0</version>
+        <version>16.1.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
diff --git a/synth.metadata b/synth.metadata
index 675b68c0..04c96f13 100644
--- a/synth.metadata
+++ b/synth.metadata
@@ -4,14 +4,14 @@
       "git": {
         "name": ".",
         "remote": "https://github.com/googleapis/java-storage.git",
-        "sha": "1e399cd33755e647bf08f4a82af932320cab655d"
+        "sha": "243a3cb1506b2e2d609210dc4e9608637c06d7f3"
       }
     },
     {
       "git": {
         "name": "synthtool",
         "remote": "https://github.com/googleapis/synthtool.git",
-        "sha": "7db8a6c5ffb12a6e4c2f799c18f00f7f3d60e279"
+        "sha": "68742c6016bf0d16948a572633d17955a8737414"
       }
     }
   ],
diff --git a/versions.txt b/versions.txt
index d2b8fea4..19c540ba 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,4 +1,4 @@
 # Format:
 # module:released-version:current-version
 
-google-cloud-storage:1.113.3:1.113.4-SNAPSHOT
\ No newline at end of file
+google-cloud-storage:1.113.4:1.113.5-SNAPSHOT
\ No newline at end of file


Test patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java
index 7733f13e..b4479682 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java
@@ -139,6 +139,11 @@ public class StorageRpcTestBase implements StorageRpc {
     throw new UnsupportedOperationException("Not implemented yet");
   }
 
+  @Override
+  public long getCurrentUploadOffset(String uploadId) {
+    throw new UnsupportedOperationException("Not implemented yet");
+  }
+
   @Override
   public StorageObject writeWithResponse(
       String uploadId,
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 a18345be..b6d7bbf3 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
@@ -134,6 +134,74 @@ public class BlobWriteChannelTest {
     assertEquals(MIN_CHUNK_SIZE, writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE)));
   }
 
+  @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);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(exception);
+    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);
+    replay(storageRpcMock);
+    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer.setChunkSize(MIN_CHUNK_SIZE);
+    assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
+    assertTrue(writer.isOpen());
+    assertNull(writer.getStorageObject());
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+  }
+
+  @Test
+  public void testWriteWithFlushRetryChunkWithDrift() 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);
+    expect(
+            storageRpcMock.writeWithResponse(
+                eq(UPLOAD_ID),
+                capture(capturedBuffer),
+                eq(0),
+                eq(0L),
+                eq(MIN_CHUNK_SIZE),
+                eq(false)))
+        .andThrow(exception);
+    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)))
+        .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));
+    assertTrue(writer.isOpen());
+    assertNull(writer.getStorageObject());
+    assertArrayEquals(buffer.array(), capturedBuffer.getValue());
+  }
+
   @Test
   public void testWriteWithFlush() throws IOException {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);