fix: add retry on ABORTED errors #286

thebrianchen · 2020-07-07T07:44:47Z

Porting over from node.

codecov · 2020-07-07T07:49:39Z

Codecov Report

Merging #286 into bc/bulk-master will increase coverage by 1.07%.
The diff coverage is 82.39%.

@@                 Coverage Diff                  @@
##             bc/bulk-master     #286      +/-   ##
====================================================
+ Coverage             72.94%   74.02%   +1.07%     
- Complexity             1015     1123     +108     
====================================================
  Files                    63       67       +4     
  Lines                  5407     6109     +702     
  Branches                617      672      +55     
====================================================
+ Hits                   3944     4522     +578     
- Misses                 1266     1373     +107     
- Partials                197      214      +17

Impacted Files	Coverage Δ	Complexity Δ
.../com/google/cloud/firestore/BulkWriterOptions.java	`0.00% <0.00%> (ø)`	`0.00 <0.00> (?)`
.../java/com/google/cloud/firestore/ResourcePath.java	`85.71% <0.00%> (ø)`	`20.00 <0.00> (ø)`
...oogle/cloud/firestore/spi/v1/GrpcFirestoreRpc.java	`0.00% <0.00%> (ø)`	`0.00 <0.00> (ø)`
...gle/cloud/firestore/v1/FirestoreAdminSettings.java	`13.20% <0.00%> (-2.35%)`	`2.00 <0.00> (ø)`
...le/cloud/firestore/v1/stub/FirestoreAdminStub.java	`5.88% <0.00%> (-2.46%)`	`1.00 <0.00> (ø)`
.../google/cloud/firestore/v1/stub/FirestoreStub.java	`5.26% <0.00%> (-0.99%)`	`1.00 <0.00> (ø)`
...gle/cloud/firestore/v1beta1/FirestoreSettings.java	`13.20% <0.00%> (ø)`	`2.00 <0.00> (ø)`
...le/cloud/firestore/v1beta1/stub/FirestoreStub.java	`6.25% <0.00%> (ø)`	`1.00 <0.00> (ø)`
...m/google/cloud/firestore/v1/FirestoreSettings.java	`19.29% <25.00%> (+2.31%)`	`4.00 <0.00> (ø)`
...om/google/cloud/firestore/CollectionReference.java	`47.72% <33.33%> (-2.28%)`	`10.00 <1.00> (ø)`
... and 31 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 05c59dc...78ea617. Read the comment docs.

thebrianchen · 2020-07-07T07:47:01Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+            });
+  }
+
+  private void bulkCommit(BulkCommitBatch batch) {


What do you think about having the retry logic live in BulkWriter instead of inside UpdateBuilder? I originally wanted to take care of everything in UpdateBuilder, but then I realized that doing so would prohibit creating a new BulkCommitBatch via constructor.

This matches Node so I am fully supportive.

thebrianchen · 2020-07-07T07:47:57Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+                                    null,
+                                    ((FirestoreException) e).getStatus(),
+                                    e.getMessage()));
+                          } else {


I wasn't sure how to handle the case where a non-FirestoreException was thrown. I added an additional field for an Exception in the event that the Exception thrown does not have a Status, but I wonder if this could be cleaner.

The code in bulkCommit should wrap all exceptions in FirestoreException.

Post conversation: We can't wrap some of the java exceptions that getting Futures will throw. However, the exceptions will still be surfaced to the user accordingly.

thebrianchen · 2020-07-07T07:51:26Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

+   * Removes all operations not specified in documentsToRetry from the batch. Marks the batch as
+   * READY_TO_SEND in order to allow the next batch to be sent.
+   */
+  void sliceBatchForRetry(final Set<String> documentsToRetry) {


Oh man. I really tried to use the BulkCommitBatch constructor directly to properly imitate Node. However, since the BulkCommitBatch logic lives inside UpdateBuilder, it got pretty ugly. I had to add a bunch of getters, and managing the state on a BulkCommitBatch becomes a tricky problem / potential for future bugs. I think slicing the batch here is the lesser of the two evils, and I'm open to separating out the batch state override.

How about https://gist.github.com/schmidt-sebastian/48ed97084b74bf7e3dd5936eddbfc001, or is this only addressing half of the problem?

I ran into two main issues: first, my currently implementation requires the batch in bulkCommit() to be marked final, which means it can't be changed directly, though I guess we could put it in an array. The second is that the completeFuture used to track awaitBulkCommit() will never resolve, unless we pipe it through somehow (is that possible?).

Update: Ended up updating to use BulkCommitBatch as constructor instead of slicing, pending successfulAsList

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BatchWriteResult.java

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

schmidt-sebastian · 2020-07-07T19:57:35Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+        .execute(
+            new Runnable() {
+              public void run() {
+                bulkCommit(batch);


We previously waited for bulkCommit before removing the batch from the batch queue. Do we not need to do this anymore?

I believe we are still waiting on bulkCommit to finish before removing the batch. We're waiting for attemptBulkCommit in bulkCommit(), which means that bulkCommit() should finish its logic before we remove batches from the batch queue.

Update: N/A now that we are chaining with ApiFutures.

schmidt-sebastian · 2020-07-07T19:58:17Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

-                batch.processResults(commitFuture.get(), null);
-              } catch (Exception e) {
-                batch.processResults(new ArrayList<BatchWriteResult>(), e);
+    MoreExecutors.directExecutor()


MoreExecutors.directExecutor(). execute(...) seems equivalent to just running the code in the Runnable. Let me know if I am mistaken, maybe by adding a comment.

I thought it was always good practice to execute runnables via an executor, as I haven't seen any Runnables being executed directly. Done.

google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java

schmidt-sebastian · 2020-07-07T20:33:27Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java

+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("Mock batchWrite failed in test"));
+      ++opCount;
+    }
+    assertEquals(1, opCount);


Suggested change

} catch (Exception e) {

assertTrue(e.getMessage().contains("Mock batchWrite failed in test"));

++opCount;

}

assertEquals(1, opCount);

Assert.fail(...);

} catch (Exception e) {

assertTrue(e.getMessage().contains("Mock batchWrite failed in test"));

}

schmidt-sebastian · 2020-07-07T20:34:09Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java

+
+  @Test
+  public void failsWritesAfterAllRetryAttemptsFail() throws Exception {
+    doThrow(


Does the mocking layer allow us to specify that we want this function to be called 10 times?

Changed to use doAnswer.

schmidt-sebastian · 2020-07-07T20:37:47Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

+    }
+    Set<Code> codes = FirestoreSettings.newBuilder().batchWriteSettings().getRetryableCodes();
+    for (Code code : codes) {
+      if (code.toString().equals(status.getCode().toString())) {


Suggested change

if (code.toString().equals(status.getCode().toString())) {

if (code.equals(Code.valueOf(status.getCode().name()))) {

that's cleaner, thanks.

schmidt-sebastian · 2020-07-07T20:44:39Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

+   * Removes all operations not specified in documentsToRetry from the batch. Marks the batch as
+   * READY_TO_SEND in order to allow the next batch to be sent.
+   */
+  void sliceBatchForRetry(final Set<String> documentsToRetry) {


How about https://gist.github.com/schmidt-sebastian/48ed97084b74bf7e3dd5936eddbfc001, or is this only addressing half of the problem?

This change should be cleaned up after com.google.com:google-cloud-shared-dependencies v0.8.4 is available

schmidt-sebastian · 2020-08-04T22:33:21Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BatchWriteResult.java


-  BatchWriteResult(@Nullable Timestamp timestamp, Status status, String message) {
+  BatchWriteResult(
+      DocumentReference documentReference, @Nullable Timestamp timestamp, Exception exception) {


Exception should be @Nullable

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

schmidt-sebastian · 2020-08-04T22:42:51Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+    commitFuture.addListener(
+        new Runnable() {
+          public void run() {
+            System.out.println("removing batch from batch queue. size: " + batchQueue.size());


Please remove.

schmidt-sebastian · 2020-08-04T22:44:20Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+  }
+
+  private ApiFuture<List<BatchWriteResult>> invokeBulkCommit(final BulkCommitBatch batch) {
+    return batch.bulkCommit();


This can be inlined.

schmidt-sebastian · 2020-08-04T22:45:47Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+  private ApiFuture<Void> bulkCommit(final BulkCommitBatch batch, final int attempt) {
+    final SettableApiFuture<Void> backoffFuture = SettableApiFuture.create();
+
+    class ProcessBulkCommitCallback implements ApiAsyncFunction<List<BatchWriteResult>, Void> {


Can you move these inner classes out of this method? Static classes within methods feel strange. :)

I originally wanted to avoid adding class members to the callback classes, but readability/style is more important. Thanks!

schmidt-sebastian · 2020-08-04T22:51:12Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+      @Override
+      public ApiFuture<Void> apply(Void ignored) {
+
+        // If the BatchWrite RPC fails, map the exception to each individual result.


Move comment closer to the code that does this.

schmidt-sebastian · 2020-08-04T22:53:18Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

+                  new BatchWriteResult(
+                      writes.get(i).documentReference,
+                      updateTime,
+                      updateTime == null


Should this be based on the Status?

updateTime is null if the Status is not OK from above, but I can see how using status again is more readable. Changed to an if/else block that hopefully increases readability.

schmidt-sebastian · 2020-08-04T22:53:33Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

+    }
+
+    // Mark the batch as ready to send in order to allow the batch to be retried again.
+    state = BatchState.READY_TO_SEND;


Is this now unused?

schmidt-sebastian · 2020-08-04T23:43:58Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+        batch.processResults(results);
+        if (batch.getPendingOperationCount() > 0) {
+          logger.log(
+              Level.WARNING,
+              String.format(
+                  "Current batch failed at retry #%d. Num failures: %d",
+                  attempt, batch.getPendingOperationCount()));
+
+          if (attempt < MAX_RETRY_ATTEMPTS) {
+            nextAttempt = backoff.createNextAttempt(nextAttempt);
+            BulkCommitBatch newBatch =
+                new BulkCommitBatch(firestore, batch, batch.getPendingDocuments());
+            return bulkCommit(newBatch, attempt + 1);
+          } else {
+            batch.failRemainingOperations(results);
+          }
+        }
+        return ApiFutures.immediateFuture(null);


This confused me during the review. There is a strange interaction model going on here.

batch.processResults changes the internal state, which reduces the number of pending operations (this is not obvious from the callsite). new BulkCommitBatch then uses this filtered list of pending operations, but also uses the unfiltered list of writes.

While this matches what Node does, a lot of the back and forth in Node happens in the same codepath, and here these interactions all cross into and out of UpdateBuilder.

One improvement suggestion:

Suggested change

batch.processResults(results);

if (batch.getPendingOperationCount() > 0) {

logger.log(

Level.WARNING,

String.format(

"Current batch failed at retry #%d. Num failures: %d",

attempt, batch.getPendingOperationCount()));

if (attempt < MAX_RETRY_ATTEMPTS) {

nextAttempt = backoff.createNextAttempt(nextAttempt);

BulkCommitBatch newBatch =

new BulkCommitBatch(firestore, batch, batch.getPendingDocuments());

return bulkCommit(newBatch, attempt + 1);

} else {

batch.failRemainingOperations(results);

}

}

return ApiFutures.immediateFuture(null);

batch.processResults(results);

List<DocumentReference> pendingOps = batch.getRemainingOperations();

if (!pendingOps.isEmpty()) {

logger.log(

Level.WARNING,

String.format(

"Current batch failed at retry #%d. Num failures: %d",

attempt,pendingOps.size())));

if (attempt < MAX_RETRY_ATTEMPTS) {

nextAttempt = backoff.createNextAttempt(nextAttempt);

BulkCommitBatch newBatch =

new BulkCommitBatch(firestore, batch, pendingOps);

return bulkCommit(newBatch, attempt + 1);

} else {

batch.failRemainingOperations(results);

}

}

return ApiFutures.immediateFuture(null);

Thanks for taking the time to write it out and explain it. I remember thinking the code was kind of tricky and had a smell to it, but I couldn't pinpoint what it was. Changed and updated comments for processResults.

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

schmidt-sebastian · 2020-08-05T23:15:38Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+  /**
+   * A queue of batches to be written. Use a synchronized list to avoid multi-thread concurrent
+   * modification errors.
+   */


Suggested change

/**

* A queue of batches to be written. Use a synchronized list to avoid multi-thread concurrent

* modification errors.

*/

/**

* A queue of batches to be written. Use a synchronized list to avoid multi-thread concurrent

* modification errors (as this list is modified from both the user thread and the network thread).

*/

Done. Thanks for the clarification

schmidt-sebastian · 2020-08-05T23:21:25Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

+      batch.processResults(results);
+      Set<DocumentReference> pendingOps = batch.getPendingDocuments();


The logic is much cleaner now, but still not self-explanatory. I think this can be solved by naming though:

batch.processResults(results); Set<DocumentReference> remainingOp = batch.getRemainingOperations();

Or:

batch.processResultsAndRemoveSuccesfulResultsFromPendingOperationsBernie2016(results); Set<DocumentReference> remainingOp = batch. getPendingDocuments();

sad reacts only for 2020

schmidt-sebastian · 2020-08-05T23:22:56Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

+              @Nullable Timestamp updateTime;
+              @Nullable Exception exception;
+              if (code == Status.OK) {
+                updateTime = Timestamp.fromProto(writeResult.getUpdateTime());
+                exception = null;
+              } else {
+                updateTime = null;
+                exception = FirestoreException.serverRejected(code, status.getMessage());
+              }
+              result.add(
+                  new BatchWriteResult(writes.get(i).documentReference, updateTime, exception));


Suggested change

@Nullable Timestamp updateTime;

@Nullable Exception exception;

if (code == Status.OK) {

updateTime = Timestamp.fromProto(writeResult.getUpdateTime());

exception = null;

} else {

updateTime = null;

exception = FirestoreException.serverRejected(code, status.getMessage());

}

result.add(

new BatchWriteResult(writes.get(i).documentReference, updateTime, exception));

Timestamp updateTime = null;

Exception exception = null;

if (Status.OK.equals(code)) {

updateTime = Timestamp.fromProto(writeResult.getUpdateTime());

} else {

exception = FirestoreException.serverRejected(code, status.getMessage());

}

result.add(

new BatchWriteResult(writes.get(i).documentReference, updateTime, exception));

schmidt-sebastian · 2020-08-05T23:24:00Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java

-        convertBatchWriteResult(results.get(i), future);
-      } else {
-        resultsMap.get(i).setException(error);
+  /**


Missing empty line.

schmidt-sebastian · 2020-08-05T23:25:04Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java

+  }
+
+  @Test
+  public void allWritesCompleteWhenFlushCompletes() throws Exception {


Is this more "flushCompletesWhenallWritesComplete"?

schmidt-sebastian · 2020-08-05T23:27:35Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java

+      assertTrue(e.getMessage().contains("Mock batchWrite failed in test"));
+      assertEquals(retryAttempts[0], BulkWriter.MAX_RETRY_ATTEMPTS + 1);
+    }
+  }


Can we add a test that:

Adds two writes

Calls flush

Fails one write with retryable error, let's the other succeed

Retries write which succeeds

Test verifies that flush() only finishes now

I am not 100% convinced that this works right now.

Added test.

* fix: add retry on ABORTED errors * WIP: test pass individually, concurrency error * WIP: tests passing but hidden concurrency bug * use BulkCommitBatch constructor, remove completeFuture * use firestoreExecutor and update rate limiter test * add todo for successfulAsList * Update BulkWriterTest to check number of retry attempts * build: manually bump com.google.api:api-common to v1.10.0 This change should be cleaned up after com.google.com:google-cloud-shared-dependencies v0.8.4 is available * test: add junit timeout rule to BulkWriterTest (#312) * update to use successfulAsList * add test to verify successfulAsList * lint * resolve comments, fix concurrency issue? * resolve comments, add additional test for flush Co-authored-by: BenWhitehead <BenWhitehead@users.noreply.github.com>

fix: add retry on ABORTED errors

eb099d5

thebrianchen self-assigned this Jul 7, 2020

googlebot added the cla: yes This human has signed the Contributor License Agreement. label Jul 7, 2020

thebrianchen commented Jul 7, 2020

View reviewed changes

thebrianchen requested a review from schmidt-sebastian July 7, 2020 16:26

thebrianchen assigned schmidt-sebastian and unassigned thebrianchen Jul 7, 2020

schmidt-sebastian reviewed Jul 7, 2020

View reviewed changes

schmidt-sebastian removed their assignment Jul 7, 2020

WIP: test pass individually, concurrency error

95c099f

thebrianchen force-pushed the bc/bulk-retry branch from 60dcfbd to 95c099f Compare July 9, 2020 18:29

Brian Chen added 5 commits July 9, 2020 19:31

WIP: tests passing but hidden concurrency bug

2e8d5ff

use BulkCommitBatch constructor, remove completeFuture

a9d3180

use firestoreExecutor and update rate limiter test

556aa66

add todo for successfulAsList

5658d46

Update BulkWriterTest to check number of retry attempts

6966092

thebrianchen self-assigned this Jul 20, 2020

BenWhitehead and others added 5 commits July 30, 2020 18:05

build: manually bump com.google.api:api-common to v1.10.0

320c54c

This change should be cleaned up after com.google.com:google-cloud-shared-dependencies v0.8.4 is available

test: add junit timeout rule to BulkWriterTest (#312)

15343d5

update to use successfulAsList

f2b0d5b

add test to verify successfulAsList

127c862

lint

e1a6378

thebrianchen requested a review from schmidt-sebastian August 4, 2020 21:45

thebrianchen assigned schmidt-sebastian and unassigned thebrianchen Aug 4, 2020

schmidt-sebastian reviewed Aug 4, 2020

View reviewed changes

schmidt-sebastian assigned thebrianchen and unassigned schmidt-sebastian Aug 4, 2020

resolve comments, fix concurrency issue?

0300c3e

thebrianchen requested a review from schmidt-sebastian August 5, 2020 15:45

thebrianchen assigned schmidt-sebastian and thebrianchen and unassigned thebrianchen Aug 5, 2020

schmidt-sebastian reviewed Aug 5, 2020

View reviewed changes

schmidt-sebastian assigned thebrianchen and unassigned schmidt-sebastian Aug 5, 2020

resolve comments, add additional test for flush

78ea617

thebrianchen added the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Aug 6, 2020

thebrianchen requested a review from schmidt-sebastian August 6, 2020 18:55

thebrianchen assigned schmidt-sebastian and unassigned thebrianchen Aug 6, 2020

yoshi-kokoro removed the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Aug 6, 2020

schmidt-sebastian approved these changes Aug 7, 2020

View reviewed changes

schmidt-sebastian assigned thebrianchen and unassigned schmidt-sebastian Aug 7, 2020

BenWhitehead merged commit c06ae8a into bc/bulk-master Aug 7, 2020

BenWhitehead deleted the bc/bulk-retry branch August 7, 2020 17:33

BenWhitehead restored the bc/bulk-retry branch August 7, 2020 17:37

BenWhitehead deleted the bc/bulk-retry branch August 7, 2020 18:24

	if (code.toString().equals(status.getCode().toString())) {
	if (code.equals(Code.valueOf(status.getCode().name()))) {

		batch.processResults(results);
		Set<DocumentReference> pendingOps = batch.getPendingDocuments();

fix: add retry on ABORTED errors #286

fix: add retry on ABORTED errors #286

Conversation

thebrianchen commented Jul 7, 2020 • edited

codecov bot commented Jul 7, 2020 • edited

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

thebrianchen commented Jul 7, 2020 •

edited

codecov bot commented Jul 7, 2020 •

edited