ServerStreaming retries by igorbernstein2 · Pull Request #449 · googleapis/gax-java

igorbernstein2 · 2017-12-29T15:35:40Z

Add retries and resume support. This is last part of #433

This PR consists of 3 parts split by commit:

minor refactoring of test helpers to enable tests on the later parts
Implement retries w. resume
Implement support for timeouts

Conceptually, retries for streams can be split into 2 phases: before the first response is observed and after. Before the first response is observed, retries can be handled automatically similar to unary RPCs. After the first response is observed, manual work needs to done to recalculate the retry request. This work is delegated to a StreamTracker, which is notified of incoming responses and can be asked to build a resume request on failure. Gax should offer out of box support for the first kind of retries and allow client developers to manually add support for resumes via handwritten code. I envision the workflow for client developers as: add retry codes & settings to GAPIC yaml to enable simple retries, then implement a StreamTracker and set it on the ServerStreamingCallSettings.

This PR also implements timeouts for RPCs. The existing concepts of RPC & total timeouts are extended: RPC timeout now limits the time interval between a consumer signaling demand for the next response via StreamController#request() and receiving a response in ResponseObserver#onResponse. Total timeout limits the total duration of the stream from the initial call() until the last onComplete/onError of the last retry attempt. Furthermore, the concept of idle timeout is added to streams with manual flow control, this limits the time between last observed consumer activity (the time between finishing processing a response in onResponse to the next call to request). This is meant to address the possibility of the caller forgetting to cancel a partially read stream.

Since GRPC does not implement the concept of RPC timeouts for streams, this PR implements it as a watchdog helper. The helper tracks activity per stream and schedules periodic garbage collection sweeps. So stream RPC timeouts are a lot looser than their unary counterparts. Furthermore, the helper needs to receive more information from its callers than is currently available: it needs to be able to accept both an rpc timeout and a total stream deadline. I couldn't figure out how to generalize this and extend the ApiContext to carry this information. So the watchdog helper steps outside the callable chain pattern and is implemented as a factory for ResponseObserver decorators.

codecov-io · 2017-12-29T15:39:50Z

Codecov Report

Merging #449 into master will decrease coverage by 0.56%.
The diff coverage is 64.98%.

@@ Coverage Diff @@## master #449 +/- ##
============================================
- Coverage 71.7% 71.13% -0.57% - Complexity 751 787 +36 
============================================
Files 155 159 +4 Lines 3354 3707 +353 Branches 261 300 +39 ============================================
+ Hits 2405 2637 +232 - Misses 847 945 +98 - Partials 102 125 +23

Impacted Files	Coverage Δ	Complexity Δ
...e/api/gax/rpc/RetryingServerStreamingCallable.java	`0% <0%> (ø)`	`0 <0> (?)`
...rc/main/java/com/google/api/gax/rpc/Callables.java	`70% <0%> (-26.56%)`	`6 <0> (ø)`
...oogle/api/gax/rpc/ServerStreamingCallSettings.java	`0% <0%> (ø)`	`0 <0> (ø)`	⬇️
...a/com/google/api/gax/grpc/GrpcCallableFactory.java	`65.45% <25%> (-3.78%)`	`8 <0> (ø)`
...src/main/java/com/google/api/gax/rpc/Watchdog.java	`83.76% <83.76%> (ø)`	`6 <6> (?)`
...i/gax/retrying/SimpleStreamResumptionStrategy.java	`85.71% <85.71%> (ø)`	`4 <4> (?)`
.../google/api/gax/retrying/RetryingServerStream.java	`90.07% <90.07%> (ø)`	`25 <25> (?)`
... and 2 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 bdcd4d8...0572d42. Read the comment docs.

Sign in to view

@@ -0,0 +1,405 @@
+/*
+ * Copyright 2017, Google LLC All rights reserved.


Sign in to view

 *
- * <p>This class includes settings that are applicable to all server streaming calls
+ * <p>This class includes settings that are applicable to all server streaming calls, which
+ * currently is retries.


Sign in to view

+ * <p>Retry configuration allows for the stream to be restarted and resumed. it is composed of 3
+ * parts: the retryable codes, the retry settings and the stream tracker. The retryable codes
+ * indicate which codes cause a retry to occur, the retry settings configure the retry logic when
+ * the retry needs to happen and the stream tracker composes the request to resume the stream. To


Sign in to view

+ * parts: the retryable codes, the retry settings and the stream tracker. The retryable codes
+ * indicate which codes cause a retry to occur, the retry settings configure the retry logic when
+ * the retry needs to happen and the stream tracker composes the request to resume the stream. To
+ * turn off retries, set the retryable codes needs to be set to the empty set.


Sign in to view

+ * <p>Implementations don't have to be threadsafe because all of the calls will be serialized.
+ */
+@BetaApi("The surface for streaming is not stable yet and may change in the future.")
+public interface StreamTracker<RequestT, ResponseT> {


Sign in to view

+ }
+
+ /**
+ * Checks if this stream has over run any of its timeouts and cancels it if it does.


Sign in to view

+ clientContext.getClock(),
+ callSettings.getTimeoutCheckInterval(),
+ callSettings.getIdleTimeout());
+ watchdog.start();


Sign in to view

+ }
+
+ /** Wraps the target observer with timing constraints. */
+ public ResponseObserver<ResponseT> wrapWithTimeout(


Sign in to view

+ Preconditions.checkArgument(count > 0, "count must be > 0");
+ Preconditions.checkState(!autoAutoFlowControl, "Auto flow control is enabled");
+
+ // Only reset the request water mark if there is no outstanding requests.


Sign in to view

+ @Override
+ public void onResponse(ResponseT response) {
+ synchronized (lock) {
+ state = State.DELIVERING;


igorbernstein2 · 2018-01-04T06:58:37Z

PTAL, all feedback should be addressed

Sign in to view


+ // NOTE: This creates a Watchdog per streaming API method. Ideally, there should only be a
+ // single Watchdog for the entire process, however that change would be fairly invasive and
+ // the cost of multiple Watchdogs is fairly small, they all use the same executor. If this


Sign in to view

+import com.google.api.core.BetaApi;
+
+/**
+ * This is part of the server streaming retry api. It's implementers are responsible for tracking


Sign in to view

+ /** Creates a new instance of this StreamResumptionStrategy without accumulated state */
+ StreamResumptionStrategy<RequestT, ResponseT> createNew();
+
+ /** Called by the {@link RetryingServerStream} to notify of a successfully received response. */


Sign in to view

+ /**
+ * Called when a stream needs to be restarted, the implementation should generate a request that
+ * will yield a stream whose first response would come right after the last response in
+ * onProgress.


garrettjonesgoogle · 2018-01-05T16:28:00Z

@vam-google could you take a look too since you did the unary retry implementation?

igorbernstein2 · 2018-01-07T00:37:41Z

All feedback addressed

igorbernstein2 · 2018-01-09T20:14:42Z

rebased

… started the call

vam-google

(Assuming that Reframer portion of streaming api is moved fom gax to google-cloud-java directly).

Also please check ReframingResponseObserver's thread safety implementation, some of @GuardedBy fields look not guarded by lock, like numPending on L136.

Sign in to view

+import java.util.concurrent.TimeUnit;
+import org.threeten.bp.Duration;
+
+/**


Sign in to view

+ private final ResponseObserver<ResponseT> outerObserver;
+
+ // Start state
+ private boolean autoFlowControl = true;


Sign in to view

+ final StreamController localInnerController;
+
+ synchronized (lock) {
+ int maxInc = Integer.MAX_VALUE - pendingRequests;


Sign in to view

+ pendingRequests = Integer.MAX_VALUE;
+ }
+
+ timedAttemptSettings = retryAlgorithm.createFirstAttempt();


Sign in to view

+ *
+ * @see ResponseObserver#onError(Throwable)
+ */
+ private void onAttemptError(Throwable t) {


Sign in to view

+ Duration checkInterval,
+ Duration idleTimeout) {
+
+ Preconditions.checkNotNull(executor, "executor can't be null");


Sign in to view

+ WatchdogStream(ResponseObserver<ResponseT> responseObserver, Duration waitTimeout) {
+ this.waitTimeout = waitTimeout;
+ this.outerResponseObserver = responseObserver;
+ this.lastActivityAt = clock.millisTime();


Sign in to view

+ }
+
+ // Increment the request count without overflow
+ int maxIncrement = Integer.MAX_VALUE - pendingCount;


Sign in to view

+ }
+
+ /** The marker exception thrown when a timeout is exceeded. */
+ public static class IdleConnectionException extends ApiException {


Sign in to view

+ }
+ }
+
+ public static final StatusCode LOCAL_ABORTED_STATUS_CODE =


igorbernstein2 · 2018-01-26T22:55:01Z

I've implemented most feedback, still outstanding on my part:

factor out retry business logic into RetryAlgorithm, whose design is awaiting review
factor a RetryExecutor-like from RetryStream

Things that I addressed, but don't think need code changes:

watchdog threadpool
watchdog self start
if Callables.retryabe should be responsible for excluding the callable

Discussion of pendingRequestCount and ServerStreamingCallable call method naming should be moved to a separate issue

This reverts commit 195a7f5.

igorbernstein2 · 2018-02-01T23:41:28Z

Closing in favor of #463

igorbernstein2 mentioned this pull request Dec 29, 2017
Server streaming callable improvements #433
Closed

igorbernstein2 force-pushed the ssc-7-retries branch 2 times, most recently from 7016b42 to c665770Compare December 29, 2017 22:17

garrettjonesgoogle reviewed Jan 3, 2018
View reviewed changes

igorbernstein2 force-pushed the ssc-7-retries branch from 842f2cd to 429d445Compare January 4, 2018 06:53

garrettjonesgoogle reviewed Jan 5, 2018
View reviewed changes

igorbernstein2 force-pushed the ssc-7-retries branch 2 times, most recently from ce5390d to 7349b38Compare January 9, 2018 20:14

igorbernstein2 added 7 commits January 9, 2018 23:54

Update test server streaming test helpers to include the request that…
ca9b4ff
… started the call

retries
8f80a40

watchdog
1e78857

address feedback
cd22646

update docs
7c7cd06

fix rebase
0e80cc2

update after rebase
c1e5f32

igorbernstein2 force-pushed the ssc-7-retries branch from 7367c0c to c1e5f32Compare January 10, 2018 05:01

vam-google reviewed Jan 16, 2018
View reviewed changes

address some feedback
6961ad6

igorbernstein2 mentioned this pull request Jan 29, 2018
Expose retryableCodes on ServerStreamingCallables. #461
Merged

igorbernstein2 added 2 commits January 29, 2018 18:28

integrate with RetryAlgorithm
195a7f5

Revert "integrate with RetryAlgorithm"
0572d42
This reverts commit 195a7f5.

igorbernstein2 mentioned this pull request Jan 30, 2018
Server streaming retries take 2 #463
Merged

igorbernstein2 closed this Feb 1, 2018

igorbernstein2 deleted the ssc-7-retries branch March 18, 2018 06:29

		@@ -0,0 +1,405 @@
		/*
		* Copyright 2017, Google LLC All rights reserved.

Uh oh!

Conversation

igorbernstein2 commented Dec 29, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov-io commented Dec 29, 2017• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

igorbernstein2 commented Jan 4, 2018

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

garrettjonesgoogle commented Jan 5, 2018

Uh oh!

igorbernstein2 commented Jan 7, 2018

Uh oh!

igorbernstein2 commented Jan 9, 2018

Uh oh!

vam-google left a comment

Choose a reason for hiding this comment

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

This comment was marked as spam.

Uh oh!

igorbernstein2 commented Dec 29, 2017•
edited
Loading

codecov-io commented Dec 29, 2017•
edited
Loading