fix: Unary Callables Deadline values respect the TotalTimeout in RetrySettings

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallOptions.java

@@ -34,6 +34,7 @@
 import com.google.auto.value.AutoValue;
 import com.google.protobuf.TypeRegistry;
 import javax.annotation.Nullable;
+import org.threeten.bp.Duration;
 import org.threeten.bp.Instant;
 
 /** Options for an http-json call, including deadline and credentials. */
@@ -45,6 +46,9 @@ public abstract class HttpJsonCallOptions {
   @Nullable
   public abstract Instant getDeadline();
 
+  @Nullable
+  public abstract Duration getTimeout();
+
   @Nullable
   public abstract Credentials getCredentials();
 
@@ -69,6 +73,11 @@ public HttpJsonCallOptions merge(HttpJsonCallOptions inputOptions) {
       builder.setDeadline(newDeadline);
     }
 
+    Duration newTimeout = inputOptions.getTimeout();
+    if (newTimeout != null) {
+      builder.setTimeout(newTimeout);
+    }
+
     Credentials newCredentials = inputOptions.getCredentials();
     if (newCredentials != null) {
       builder.setCredentials(newCredentials);
@@ -86,6 +95,8 @@ public HttpJsonCallOptions merge(HttpJsonCallOptions inputOptions) {
   public abstract static class Builder {
     public abstract Builder setDeadline(Instant value);
 
+    public abstract Builder setTimeout(Duration value);
+
     public abstract Builder setCredentials(Credentials value);
 
     public abstract Builder setTypeRegistry(TypeRegistry value);

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonClientCallImpl.java

@@ -33,6 +33,7 @@
 import com.google.api.gax.httpjson.ApiMethodDescriptor.MethodType;
 import com.google.api.gax.httpjson.HttpRequestRunnable.ResultListener;
 import com.google.api.gax.httpjson.HttpRequestRunnable.RunnableResult;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -42,8 +43,12 @@
 import java.util.Queue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
+import org.threeten.bp.Duration;
 
 /**
  * This class serves as main implementation of {@link HttpJsonClientCall} for REST transport and is
@@ -88,6 +93,7 @@ final class HttpJsonClientCallImpl<RequestT, ResponseT>
   private final ApiMethodDescriptor<RequestT, ResponseT> methodDescriptor;
   private final HttpTransport httpTransport;
   private final Executor executor;
+  private final ScheduledExecutorService deadlineCancellationExecutor;
 
   //
   // Request-specific data (provided by client code) before we get a response.
@@ -128,6 +134,7 @@ final class HttpJsonClientCallImpl<RequestT, ResponseT>
     this.httpTransport = httpTransport;
     this.executor = executor;
     this.closed = false;
+    this.deadlineCancellationExecutor = Executors.newScheduledThreadPool(1);
   }
 
   @Override
@@ -161,6 +168,36 @@ public void start(Listener<ResponseT> responseListener, HttpJsonMetadata request
       this.listener = responseListener;
       this.requestHeaders = requestHeaders;
     }
+
+    // Use the timeout duration value instead of calculating the future Instant
+    Duration timeout = callOptions.getTimeout();
+    if (timeout != null) {
+      // If the future timeout amount has been calculated as a negative value,
+      // we cancel the call immediately as the deadline has already been exceeded.
+      // The RetryAlgorithm should for this value and not run schedule a run
+      // if this is negative.
+      long timeoutNanos = 0;
+      if (!timeout.isNegative()) {
+        timeoutNanos = timeout.toNanos();
+      }
+      this.deadlineCancellationExecutor.schedule(
+          this::closeAndNotifyListeners, timeoutNanos, TimeUnit.NANOSECONDS);
+    }
+  }
+
+  // No need to trigger the deliver() loop again as we have already closed the runnable
+  // task and added the OnCloseNotificationTask. We notify the FutureListener that the
+  // there is a timeout exception from this RPC call (DEADLINE_EXCEEDED)
+  private void closeAndNotifyListeners() {
+    synchronized (lock) {
+      close(
+          StatusCode.Code.DEADLINE_EXCEEDED.getHttpStatusCode(),
+          "Deadline exceeded",
+          new HttpJsonStatusRuntimeException(
+              StatusCode.Code.DEADLINE_EXCEEDED.getHttpStatusCode(), "Deadline exceeded", null),
+          true);
+    }
+    notifyListeners();
   }
 
   @Override

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonClientCalls.java

@@ -34,8 +34,6 @@
 import com.google.api.gax.rpc.ApiCallContext;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import javax.annotation.Nonnull;
-import org.threeten.bp.Instant;
 
 /**
  * {@code HttpJsonClientCalls} creates a new {@code HttpJsonClientCAll} from the given call context.
@@ -50,12 +48,16 @@ public static <RequestT, ResponseT> HttpJsonClientCall<RequestT, ResponseT> newC
 
     HttpJsonCallContext httpJsonContext = HttpJsonCallContext.createDefault().nullToSelf(context);
 
-    // Try to convert the timeout into a deadline and use it if it occurs before the actual deadline
+    // Use the context's timeout instead of calculating a future deadline with the System clock.
+    // The timeout value is calculated from TimedAttemptSettings which accounts for the
+    // TotalTimeout value set in the RetrySettings.
     if (httpJsonContext.getTimeout() != null) {
-      @Nonnull Instant newDeadline = Instant.now().plus(httpJsonContext.getTimeout());
       HttpJsonCallOptions callOptions = httpJsonContext.getCallOptions();
-      if (callOptions.getDeadline() == null || newDeadline.isBefore(callOptions.getDeadline())) {
-        callOptions = callOptions.toBuilder().setDeadline(newDeadline).build();
+      // HttpJsonChannel expects the HttpJsonCallOptions and we store the timeout duration
+      // inside the HttpJsonCallOptions
+      if (callOptions.getTimeout() == null
+          || httpJsonContext.getTimeout().compareTo(callOptions.getTimeout()) < 0) {
+        callOptions = callOptions.toBuilder().setTimeout(httpJsonContext.getTimeout()).build();
         httpJsonContext = httpJsonContext.withCallOptions(callOptions);
       }
     }

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpRequestRunnable.java

@@ -57,7 +57,6 @@
 import java.util.Map.Entry;
 import javax.annotation.Nullable;
 import org.threeten.bp.Duration;
-import org.threeten.bp.Instant;
 
 /** A runnable object that creates and executes an HTTP request. */
 class HttpRequestRunnable<RequestT, ResponseT> implements Runnable {
@@ -191,13 +190,27 @@ HttpRequest createHttpRequest() throws IOException {
 
     HttpRequest httpRequest = buildRequest(requestFactory, url, jsonHttpContent);
 
-    Instant deadline = httpJsonCallOptions.getDeadline();
-    if (deadline != null) {
-      long readTimeout = Duration.between(Instant.now(), deadline).toMillis();
+    Duration timeout = httpJsonCallOptions.getTimeout();
+    if (timeout != null) {
+      long timeoutMs = timeout.toMillis();
+
+      // Read timeout is the timeout between reading two data packets and not total timeout
+      // HttpJsonClientCallsImpl implements a deadlineCancellationExecutor to cancel the
+      // RPC when it exceeds the RPC timeout
       if (httpRequest.getReadTimeout() > 0
-          && httpRequest.getReadTimeout() < readTimeout
-          && readTimeout < Integer.MAX_VALUE) {
-        httpRequest.setReadTimeout((int) readTimeout);
+          && httpRequest.getReadTimeout() < timeoutMs
+          && timeoutMs < Integer.MAX_VALUE) {
+        httpRequest.setReadTimeout((int) timeoutMs);
+      }
+
+      // Connect timeout is the time allowed for establishing the connection.
+      // This is updated to match the RPC timeout as we do not want a shorter
+      // connect timeout to preemptively throw a ConnectExcepetion before
+      // we've reached the RPC timeout
+      if (httpRequest.getConnectTimeout() > 0
+          && httpRequest.getConnectTimeout() < timeoutMs
+          && timeoutMs < Integer.MAX_VALUE) {
+        httpRequest.setConnectTimeout((int) timeoutMs);
       }
     }
 

gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonDirectCallableTest.java

@@ -167,7 +167,10 @@ public void testSuccessfulMultipleResponsesForUnaryCall()
     HttpJsonDirectCallable<Field, Field> callable =
         new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
 
-    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withTimeout(Duration.ofSeconds(30));
 
     Field request = createTestMessage(2);
     Field expectedResponse = createTestMessage(2);
@@ -199,7 +202,10 @@ public void testErrorMultipleResponsesForUnaryCall()
     HttpJsonDirectCallable<Field, Field> callable =
         new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
 
-    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withTimeout(Duration.ofSeconds(30));
 
     Field request = createTestMessage(2);
     Field expectedResponse = createTestMessage(2);
@@ -228,7 +234,10 @@ public void testErrorUnaryResponse() throws InterruptedException {
     HttpJsonDirectCallable<Field, Field> callable =
         new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
 
-    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withTimeout(Duration.ofSeconds(30));
 
     ApiException exception =
         ApiExceptionFactory.createException(
@@ -257,7 +266,10 @@ public void testErrorNullContentSuccessfulResponse() throws InterruptedException
     HttpJsonDirectCallable<Field, Field> callable =
         new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
 
-    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withTimeout(Duration.ofSeconds(30));
 
     MOCK_SERVICE.addNullResponse();
 
@@ -283,7 +295,10 @@ public void testErrorNullContentFailedResponse() throws InterruptedException {
     HttpJsonDirectCallable<Field, Field> callable =
         new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
 
-    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withTimeout(Duration.ofSeconds(30));
     MOCK_SERVICE.addNullResponse(400);
 
     try {
@@ -306,7 +321,10 @@ public void testErrorNon2xxOr4xxResponse() throws InterruptedException {
     HttpJsonDirectCallable<Field, Field> callable =
         new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
 
-    HttpJsonCallContext callContext = HttpJsonCallContext.createDefault().withChannel(channel);
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault()
+            .withChannel(channel)
+            .withTimeout(Duration.ofSeconds(30));
 
     ApiException exception =
         ApiExceptionFactory.createException(
@@ -323,6 +341,34 @@ public void testErrorNon2xxOr4xxResponse() throws InterruptedException {
     }
   }
 
+  /**
+   * Expectation is that an RPC that exceeds the Timeout value set will receive a DEADLINE_EXCEEDED
+   * response back. In this test, the call has a timeout value that is smaller than the time it
+   * takes for the mock service to return a response.
+   *
+   * @throws InterruptedException
+   */
+  @Test
+  public void testDeadlineExceededResponse() throws InterruptedException {
+    HttpJsonDirectCallable<Field, Field> callable =
+        new HttpJsonDirectCallable<>(FAKE_METHOD_DESCRIPTOR);
+
+    HttpJsonCallContext callContext =
+        HttpJsonCallContext.createDefault().withChannel(channel).withTimeout(Duration.ofSeconds(3));
+
+    Field response = createTestMessage(10);
+    MOCK_SERVICE.addResponse(response, Duration.ofSeconds(5));
+
+    try {
+      callable.futureCall(createTestMessage(10), callContext).get();
+      Assert.fail("No exception raised");
+    } catch (ExecutionException e) {
+      HttpJsonStatusRuntimeException respExp = (HttpJsonStatusRuntimeException) e.getCause();
+      assertThat(respExp.getStatusCode()).isEqualTo(504);
+      assertThat(respExp.getMessage()).isEqualTo("Deadline exceeded");
+    }
+  }
+
   private Field createTestMessage(int number) {
     return Field.newBuilder() // "echo" service
         .setName("john/imTheBestField")

gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonDirectServerStreamingCallableTest.java

@@ -34,6 +34,7 @@
 import com.google.api.gax.httpjson.testing.MockHttpService;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ClientContext;
+import com.google.api.gax.rpc.DeadlineExceededException;
 import com.google.api.gax.rpc.ResponseObserver;
 import com.google.api.gax.rpc.ServerStream;
 import com.google.api.gax.rpc.ServerStreamingCallSettings;
@@ -65,6 +66,7 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.threeten.bp.Duration;
 
 @RunWith(JUnit4.class)
 public class HttpJsonDirectServerStreamingCallableTest {
@@ -144,7 +146,9 @@ public void setUp() {
     clientContext =
         ClientContext.newBuilder()
             .setTransportChannel(HttpJsonTransportChannel.create(channel))
-            .setDefaultCallContext(HttpJsonCallContext.of(channel, HttpJsonCallOptions.DEFAULT))
+            .setDefaultCallContext(
+                HttpJsonCallContext.of(channel, HttpJsonCallOptions.DEFAULT)
+                    .withTimeout(Duration.ofSeconds(3)))
             .build();
     streamingCallSettings = ServerStreamingCallSettings.<Color, Money>newBuilder().build();
     streamingCallable =
@@ -326,6 +330,26 @@ public void testBlockingServerStreaming() {
     Truth.assertThat(responseData).containsExactly(expected);
   }
 
+  // This test ensures that the server-side streaming does not exceed the timeout value
+  @Test
+  public void testDeadlineExceededServerStreaming() throws InterruptedException {
+    MOCK_SERVICE.addResponse(
+        new Money[] {DEFAULT_RESPONSE, DEFAULTER_RESPONSE}, Duration.ofSeconds(5));
+    Color request = Color.newBuilder().setRed(0.5f).build();
+    CountDownLatch latch = new CountDownLatch(1);
+    MoneyObserver moneyObserver = new MoneyObserver(false, latch);
+
+    streamingCallable.call(request, moneyObserver);
+
+    moneyObserver.controller.request(2);
+    // Set the latch's await time to above the context's timeout value to ensure that
+    // the latch has been released.
+    Truth.assertThat(latch.await(5000, TimeUnit.MILLISECONDS)).isTrue();
+
+    Truth.assertThat(moneyObserver.error).isInstanceOf(DeadlineExceededException.class);
+    Truth.assertThat(moneyObserver.error).hasMessageThat().isEqualTo("Deadline exceeded");
+  }
+
   static class MoneyObserver extends StateCheckingResponseObserver<Money> {
     private final boolean autoFlowControl;
     private final CountDownLatch latch;

gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/testing/MockHttpService.java

@@ -46,6 +46,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
+import org.threeten.bp.Duration;
 
 /**
  * Mocks an HTTPTransport. Expected responses and exceptions can be added to a queue from which this
@@ -87,6 +88,11 @@ public synchronized void addResponse(Object response) {
     responseHandlers.add(new MessageResponseFactory(endpoint, serviceMethodDescriptors, response));
   }
 
+  public synchronized void addResponse(Object response, Duration delay) {
+    responseHandlers.add(
+        new MessageResponseFactory(endpoint, serviceMethodDescriptors, response, delay));
+  }
+
   /** Add an expected null response (empty HTTP response body) with a custom status code. */
   public synchronized void addNullResponse(int statusCode) {
     responseHandlers.add(
@@ -182,16 +188,33 @@ private static class MessageResponseFactory implements HttpResponseFactory {
     private final List<ApiMethodDescriptor> serviceMethodDescriptors;
     private final Object response;
     private final String endpoint;
+    private final Duration delay;
 
     public MessageResponseFactory(
         String endpoint, List<ApiMethodDescriptor> serviceMethodDescriptors, Object response) {
+      this(endpoint, serviceMethodDescriptors, response, Duration.ofNanos(0));
+    }
+
+    public MessageResponseFactory(
+        String endpoint,
+        List<ApiMethodDescriptor> serviceMethodDescriptors,
+        Object response,
+        Duration delay) {
       this.endpoint = endpoint;
       this.serviceMethodDescriptors = ImmutableList.copyOf(serviceMethodDescriptors);
       this.response = response;
+      this.delay = delay;
     }
 
     @Override
     public MockLowLevelHttpResponse getHttpResponse(String httpMethod, String fullTargetUrl) {
+      // We use Thread.sleep to mimic a long server response. Most tests should not
+      // require a sleep and can return a response immediately.
+      try {
+        Thread.sleep(delay.toMillis());
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
       MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
 
       String relativePath = getRelativePath(fullTargetUrl);