diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java index 46e4adca1c..f5e41f1e6a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java @@ -40,6 +40,9 @@ public final class GenericWorkflowClientImpl implements GenericWorkflowClient { + private static final ScheduledExecutorService executor = + new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "generic-wf-client-async-throttler")); + private final WorkflowServiceStubs service; private final Scope metricsScope; private final GrpcRetryer grpcRetryer; @@ -160,6 +163,7 @@ public GetWorkflowExecutionHistoryResponse longPollHistory( public CompletableFuture longPollHistoryAsync( @Nonnull GetWorkflowExecutionHistoryRequest request, @Nonnull Deadline deadline) { return grpcRetryer.retryWithResultAsync( + executor, () -> { CompletableFuture result = new CompletableFuture<>(); ListenableFuture resultFuture = diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java index c27a5bb074..6bbd3b6601 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/Poller.java @@ -62,7 +62,6 @@ interface ThrowingRunnable { private final AtomicReference suspendLatch = new AtomicReference<>(); - private BackoffThrottler pollBackoffThrottler; private Throttler pollRateThrottler; private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = @@ -113,11 +112,6 @@ public void start() { new ExecutorThreadFactory( pollerOptions.getPollThreadNamePrefix(), pollerOptions.getUncaughtExceptionHandler())); - pollBackoffThrottler = - new BackoffThrottler( - pollerOptions.getPollBackoffInitialInterval(), - pollerOptions.getPollBackoffMaximumInterval(), - pollerOptions.getPollBackoffCoefficient()); for (int i = 0; i < pollerOptions.getPollThreadCount(); i++) { pollExecutor.execute(new PollLoopTask(new PollExecutionTask())); workerMetricsScope.counter(MetricsType.POLLER_START_COUNTER).inc(1); @@ -206,15 +200,24 @@ public String toString() { private class PollLoopTask implements Runnable { private final Poller.ThrowingRunnable task; + private final BackoffThrottler pollBackoffThrottler; PollLoopTask(Poller.ThrowingRunnable task) { this.task = task; + this.pollBackoffThrottler = + new BackoffThrottler( + pollerOptions.getPollBackoffInitialInterval(), + pollerOptions.getPollBackoffMaximumInterval(), + pollerOptions.getPollBackoffCoefficient()); } @Override public void run() { try { - pollBackoffThrottler.throttle(); + long throttleMs = pollBackoffThrottler.getSleepTime(); + if (throttleMs > 0) { + Thread.sleep(throttleMs); + } if (pollRateThrottler != null) { pollRateThrottler.throttle(); } @@ -237,8 +240,10 @@ public void run() { if (e instanceof InterruptedException) { // we restore the flag here, so it can be checked and processed (with exit) in finally. Thread.currentThread().interrupt(); + } else { + // Don't increase throttle on InterruptedException + pollBackoffThrottler.failure(); } - pollBackoffThrottler.failure(); uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), e); } finally { if (!shouldTerminate()) { diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java b/temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java deleted file mode 100644 index dffff83383..0000000000 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/AsyncBackoffThrottler.java +++ /dev/null @@ -1,130 +0,0 @@ -/* - * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved. - * - * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. - * - * Modifications copyright (C) 2017 Uber Technologies, Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this material except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package io.temporal.internal; - -import io.grpc.Context; -import java.time.Duration; -import java.util.Objects; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; - -/** - * Used to throttle code execution in presence of failures using exponential backoff logic. The - * formula used to calculate the next sleep interval is: - * - *

- * - * 

- * min(pow(backoffCoefficient, failureCount - 1) * initialSleep, maxSleep);
- *
- * - *

Example usage: - * - *

- * - * 

- * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2);
- * while(!stopped) {
- *     try {
- *         Future<Void> t = throttler.throttle();
- *         t.get();
- *         // some code that can fail and should be throttled
- *         ...
- *         throttler.success();
- *     }
- *     catch (Exception e) {
- *         throttler.failure();
- *     }
- * }
- *
- * - * @author fateev - */ -public final class AsyncBackoffThrottler { - - private static final ScheduledExecutorService executor = - new ScheduledThreadPoolExecutor(1, r -> new Thread(r, "async-backoff-throttler")); - - private final Duration initialSleep; - - private final Duration maxSleep; - - private final double backoffCoefficient; - - private final AtomicLong failureCount = new AtomicLong(); - - /** - * Construct an instance of the throttler. - * - * @param initialSleep time to sleep on the first failure - * @param maxSleep maximum time to sleep independently of number of failures - * @param backoffCoefficient coefficient used to calculate the next time to sleep. - */ - public AsyncBackoffThrottler( - Duration initialSleep, Duration maxSleep, double backoffCoefficient) { - Objects.requireNonNull(initialSleep, "initialSleep"); - if (backoffCoefficient < 1.0) { - throw new IllegalArgumentException( - "backoff coefficient less than 1.0: " + backoffCoefficient); - } - this.initialSleep = initialSleep; - this.maxSleep = maxSleep; - this.backoffCoefficient = backoffCoefficient; - } - - private long calculateSleepTime() { - double sleepMillis = - Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis(); - if (maxSleep != null) { - return Math.min((long) sleepMillis, maxSleep.toMillis()); - } - return (long) sleepMillis; - } - - /** Result future is done after a delay if there were failures since the last success call. */ - public CompletableFuture throttle() { - if (failureCount.get() == 0) { - return CompletableFuture.completedFuture(null); - } - CompletableFuture result = new CompletableFuture<>(); - long delay = calculateSleepTime(); - @SuppressWarnings({"FutureReturnValueIgnored", "unused"}) - ScheduledFuture ignored = - executor.schedule( - // preserving gRPC context between threads - Context.current().wrap(() -> result.complete(null)), delay, TimeUnit.MILLISECONDS); - return result; - } - - /** Resent failure count to 0. */ - public void success() { - failureCount.set(0); - } - - /** Increment failure count. */ - public void failure() { - failureCount.incrementAndGet(); - } -} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java index e7503d769b..2e59513055 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/BackoffThrottler.java @@ -22,8 +22,8 @@ import java.time.Duration; import java.util.Objects; -import java.util.concurrent.atomic.AtomicLong; import javax.annotation.Nullable; +import javax.annotation.concurrent.NotThreadSafe; /** * Used to throttle code execution in presence of failures using exponential backoff logic. The @@ -43,7 +43,10 @@ * BackoffThrottler throttler = new BackoffThrottler(1000, 60000, 2); * while(!stopped) { * try { - * throttler.throttle(); + * long throttleMs = throttler.getSleepTime(); + * if (throttleMs > 0) { + * Thread.sleep(throttleMs); + * } * // some code that can fail and should be throttled * ... * throttler.success(); @@ -56,6 +59,7 @@ * * @author fateev */ +@NotThreadSafe public final class BackoffThrottler { private final Duration initialSleep; @@ -64,7 +68,7 @@ public final class BackoffThrottler { private final double backoffCoefficient; - private final AtomicLong failureCount = new AtomicLong(); + private int failureCount = 0; /** * Construct an instance of the throttler. @@ -81,33 +85,26 @@ public BackoffThrottler( this.backoffCoefficient = backoffCoefficient; } - private long calculateSleepTime() { - double sleepMillis = - Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis(); + public long getSleepTime() { + if (failureCount == 0) return 0; + double sleepMillis = Math.pow(backoffCoefficient, failureCount - 1) * initialSleep.toMillis(); if (maxSleep != null) { return Math.min((long) sleepMillis, maxSleep.toMillis()); } return (long) sleepMillis; } - /** - * Sleep if there were failures since the last success call. - * - * @throws InterruptedException - */ - public void throttle() throws InterruptedException { - if (failureCount.get() > 0) { - Thread.sleep(calculateSleepTime()); - } + public int getAttemptCount() { + return failureCount; } - /** Resent failure count to 0. */ + /** Reset failure count to 0. */ public void success() { - failureCount.set(0); + failureCount = 0; } /** Increment failure count. */ public void failure() { - failureCount.incrementAndGet(); + failureCount++; } } diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java index ec64df2c2a..ee5ebbe54d 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcAsyncRetryer.java @@ -24,126 +24,103 @@ import io.grpc.Deadline; import io.grpc.StatusRuntimeException; import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; -import io.temporal.internal.AsyncBackoffThrottler; +import io.temporal.internal.BackoffThrottler; import io.temporal.serviceclient.RpcRetryOptions; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.function.Supplier; -import javax.annotation.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -class GrpcAsyncRetryer { +class GrpcAsyncRetryer { private static final Logger log = LoggerFactory.getLogger(GrpcRetryer.class); - public CompletableFuture retry( + private final ScheduledExecutorService executor; + private final GrpcRetryer.GrpcRetryerOptions options; + private final GetSystemInfoResponse.Capabilities serverCapabilities; + private final Supplier> function; + private final BackoffThrottler throttler; + private final Deadline retriesExpirationDeadline; + private StatusRuntimeException lastMeaningfulException = null; + + public GrpcAsyncRetryer( + ScheduledExecutorService executor, Supplier> function, GrpcRetryer.GrpcRetryerOptions options, GetSystemInfoResponse.Capabilities serverCapabilities) { + options.validate(); + + this.executor = executor; + this.options = options; + this.serverCapabilities = serverCapabilities; + this.function = function; + RpcRetryOptions rpcOptions = options.getOptions(); - @Nullable Deadline deadline = options.getDeadline(); - @Nullable - Deadline retriesExpirationDeadline = - GrpcRetryerUtils.mergeDurationWithAnAbsoluteDeadline(rpcOptions.getExpiration(), deadline); - AsyncBackoffThrottler throttler = - new AsyncBackoffThrottler( + this.retriesExpirationDeadline = + GrpcRetryerUtils.mergeDurationWithAnAbsoluteDeadline( + rpcOptions.getExpiration(), options.getDeadline()); + this.throttler = + new BackoffThrottler( rpcOptions.getInitialInterval(), rpcOptions.getMaximumInterval(), rpcOptions.getBackoffCoefficient()); + } - int attempt = 1; + public CompletableFuture retry() { CompletableFuture resultCF = new CompletableFuture<>(); - retry( - options, - serverCapabilities, - function, - attempt, - retriesExpirationDeadline, - throttler, - null, - resultCF); + retry(resultCF); return resultCF; } - private void retry( - GrpcRetryer.GrpcRetryerOptions options, - GetSystemInfoResponse.Capabilities serverCapabilities, - Supplier> function, - int attempt, - @Nullable Deadline retriesExpirationDeadline, - AsyncBackoffThrottler throttler, - StatusRuntimeException previousException, - CompletableFuture resultCF) { - throttler - .throttle() - .thenAccept( - (ignore) -> { - if (previousException != null) { - log.debug("Retrying after failure", previousException); - } - - // try-catch is because get() call might throw. - CompletableFuture result; - - try { - result = function.get(); - } catch (Throwable e) { - throttler.failure(); - // function isn't supposed to throw exceptions, it should always return a - // CompletableFuture even if it's a failed one. - // But if this happens - process the same way as it would be an exception from - // completable future - // Do not retry if it's not StatusRuntimeException - failOrRetry( - options, - serverCapabilities, - function, - attempt, - retriesExpirationDeadline, - throttler, - previousException, - e, - resultCF); - return; - } - if (result == null) { - resultCF.complete(null); - return; - } - - result.whenComplete( - (r, e) -> { - if (e == null) { - throttler.success(); - resultCF.complete(r); - } else { - throttler.failure(); - failOrRetry( - options, - serverCapabilities, - function, - attempt, - retriesExpirationDeadline, - throttler, - previousException, - e, - resultCF); - } - }); - }); + private void retry(CompletableFuture resultCF) { + CompletableFuture throttleFuture = new CompletableFuture<>(); + @SuppressWarnings({"FutureReturnValueIgnored", "unused"}) + ScheduledFuture ignored = + executor.schedule( + // preserving gRPC context between threads + Context.current().wrap(() -> throttleFuture.complete(null)), + throttler.getSleepTime(), + TimeUnit.MILLISECONDS); + + throttleFuture.thenAccept( + (ignore) -> { + if (lastMeaningfulException != null) { + log.debug("Retrying after failure", lastMeaningfulException); + } + + // try-catch is because get() call might throw. + try { + CompletableFuture result = function.get(); + if (result == null) result = CompletableFuture.completedFuture(null); + + result.whenComplete( + (r, e) -> { + if (e == null) { + throttler.success(); + resultCF.complete(r); + } else { + throttler.failure(); + failOrRetry(e, resultCF); + } + }); + + } catch (Throwable e) { + throttler.failure(); + // function isn't supposed to throw exceptions, it should always return a + // CompletableFuture even if it's a failed one. + // But if this happens - process the same way as it would be an exception from + // completable future + // Do not retry if it's not StatusRuntimeException + failOrRetry(e, resultCF); + } + }); } - private void failOrRetry( - GrpcRetryer.GrpcRetryerOptions options, - GetSystemInfoResponse.Capabilities serverCapabilities, - Supplier> function, - int attempt, - @Nullable Deadline retriesExpirationDeadline, - AsyncBackoffThrottler throttler, - StatusRuntimeException previousException, - Throwable currentException, - CompletableFuture resultCF) { + private void failOrRetry(Throwable currentException, CompletableFuture resultCF) { // If exception is thrown from CompletionStage/CompletableFuture methods like compose or handle // - it gets wrapped into CompletionException, so here we need to unwrap it. We can get not @@ -168,25 +145,17 @@ private void failOrRetry( return; } - StatusRuntimeException lastMeaningfulException = - GrpcRetryerUtils.lastMeaningfulException(statusRuntimeException, previousException); + this.lastMeaningfulException = + GrpcRetryerUtils.lastMeaningfulException(statusRuntimeException, lastMeaningfulException); if (GrpcRetryerUtils.ranOutOfRetries( options.getOptions(), - attempt, - retriesExpirationDeadline, + this.throttler.getAttemptCount(), + this.retriesExpirationDeadline, Context.current().getDeadline())) { log.debug("Out of retries, throwing", lastMeaningfulException); resultCF.completeExceptionally(lastMeaningfulException); } else { - retry( - options, - serverCapabilities, - function, - attempt + 1, - retriesExpirationDeadline, - throttler, - lastMeaningfulException, - resultCF); + retry(resultCF); } } diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java index 346d56ce81..4fbfb567ab 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcRetryer.java @@ -25,13 +25,12 @@ import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; import io.temporal.serviceclient.RpcRetryOptions; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; import java.util.function.Supplier; import javax.annotation.Nonnull; import javax.annotation.Nullable; public final class GrpcRetryer { - private static final GrpcSyncRetryer SYNC = new GrpcSyncRetryer(); - private static final GrpcAsyncRetryer ASYNC = new GrpcAsyncRetryer(); private final Supplier serverCapabilities; @@ -58,12 +57,14 @@ public void retry(RetryableProc r, GrpcRetryerOptions o public R retryWithResult( RetryableFunc r, GrpcRetryerOptions options) throws T { - return SYNC.retry(r, options, serverCapabilities.get()); + return new GrpcSyncRetryer().retry(r, options, serverCapabilities.get()); } public CompletableFuture retryWithResultAsync( - Supplier> function, GrpcRetryerOptions options) { - return ASYNC.retry(function, options, serverCapabilities.get()); + ScheduledExecutorService executor, + Supplier> function, + GrpcRetryerOptions options) { + return new GrpcAsyncRetryer(executor, function, options, serverCapabilities.get()).retry(); } public static class GrpcRetryerOptions { diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java index 1fb20f3045..e80444a3b2 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/retryer/GrpcSyncRetryer.java @@ -57,7 +57,10 @@ public R retry( attempt++; try { - throttler.throttle(); + long throttleMs = throttler.getSleepTime(); + if (throttleMs > 0) { + Thread.sleep(throttleMs); + } if (lastMeaningfulException != null) { log.debug("Retrying after failure", lastMeaningfulException); } @@ -76,11 +79,10 @@ public R retry( } lastMeaningfulException = GrpcRetryerUtils.lastMeaningfulException(e, lastMeaningfulException); + throttler.failure(); } // No catch block for any other exceptions because we don't retry them, we pass them through. // It's designed this way because it's GrpcRetryer, not general purpose retryer. - - throttler.failure(); } while (!GrpcRetryerUtils.ranOutOfRetries( rpcOptions, attempt, retriesExpirationDeadline, Context.current().getDeadline())); diff --git a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java index d04f2064cb..ffe43c8d9a 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcAsyncRetryerTest.java @@ -37,8 +37,6 @@ public class GrpcAsyncRetryerTest { - private static final GrpcAsyncRetryer DEFAULT_ASYNC_RETRYER = new GrpcAsyncRetryer(); - private static ScheduledExecutorService scheduledExecutor; @BeforeClass @@ -62,14 +60,17 @@ public void testExpirationAsync() throws InterruptedException { .setExpiration(Duration.ofMillis(500)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { + attempts.incrementAndGet(); throw new StatusRuntimeException(Status.fromCode(STATUS_CODE)); }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); fail("unreachable"); } catch (ExecutionException e) { @@ -77,6 +78,7 @@ public void testExpirationAsync() throws InterruptedException { assertEquals(STATUS_CODE, ((StatusRuntimeException) e.getCause()).getStatus().getCode()); } + assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1); assertTrue(System.currentTimeMillis() - start > 500); } @@ -91,10 +93,12 @@ public void testExpirationFutureAsync() throws InterruptedException { .setExpiration(Duration.ofMillis(500)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { + attempts.incrementAndGet(); CompletableFuture result = new CompletableFuture<>(); result.completeExceptionally( new StatusRuntimeException(Status.fromCode(STATUS_CODE))); @@ -102,12 +106,15 @@ public void testExpirationFutureAsync() throws InterruptedException { }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); fail("unreachable"); } catch (ExecutionException e) { assertTrue(e.getCause() instanceof StatusRuntimeException); assertEquals(STATUS_CODE, ((StatusRuntimeException) e.getCause()).getStatus().getCode()); } + + assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1); assertTrue(System.currentTimeMillis() - start > 500); } @@ -117,15 +124,18 @@ public void testDoNotRetryAsync() throws InterruptedException { RpcRetryOptions options = RpcRetryOptions.newBuilder() - .setInitialInterval(Duration.ofMillis(10)) - .setMaximumInterval(Duration.ofMillis(100)) + .setInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) .addDoNotRetry(STATUS_CODE, null) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { + if (attempts.incrementAndGet() > 1) + fail("We should not retry on exception that we specified to don't retry"); CompletableFuture result = new CompletableFuture<>(); result.completeExceptionally( new StatusRuntimeException(Status.fromCode(STATUS_CODE))); @@ -133,6 +143,7 @@ public void testDoNotRetryAsync() throws InterruptedException { }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); fail("unreachable"); } catch (ExecutionException e) { @@ -141,27 +152,31 @@ public void testDoNotRetryAsync() throws InterruptedException { } assertTrue( "We should fail fast on exception that we specified to don't retry", - System.currentTimeMillis() - start < 10_000); + System.currentTimeMillis() - start < 1000); } @Test public void testInterruptedExceptionAsync() throws InterruptedException { RpcRetryOptions options = RpcRetryOptions.newBuilder() - .setInitialInterval(Duration.ofMillis(10)) - .setMaximumInterval(Duration.ofMillis(100)) + .setInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { + if (attempts.incrementAndGet() > 1) + fail("We should not retry on InterruptedException"); CompletableFuture result = new CompletableFuture<>(); result.completeExceptionally(new InterruptedException("simulated")); return result; }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); fail("unreachable"); } catch (ExecutionException e) { @@ -169,27 +184,31 @@ public void testInterruptedExceptionAsync() throws InterruptedException { assertEquals("simulated", e.getCause().getMessage()); } assertTrue( - "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 10_000); + "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 1000); } @Test public void testNotStatusRuntimeExceptionAsync() throws InterruptedException { RpcRetryOptions options = RpcRetryOptions.newBuilder() - .setInitialInterval(Duration.ofMillis(10)) - .setMaximumInterval(Duration.ofMillis(100)) + .setInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { + if (attempts.incrementAndGet() > 1) + fail("We should not retry if the exception is not StatusRuntimeException"); CompletableFuture result = new CompletableFuture<>(); result.completeExceptionally(new IllegalArgumentException("simulated")); return result; }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); fail("unreachable"); } catch (ExecutionException e) { @@ -198,7 +217,7 @@ public void testNotStatusRuntimeExceptionAsync() throws InterruptedException { } assertTrue( "If the exception is not StatusRuntimeException - we shouldn't retry", - System.currentTimeMillis() - start < 10_000); + System.currentTimeMillis() - start < 1000); } @Test @@ -217,8 +236,8 @@ public void testRetryDeadlineExceededException() { StatusRuntimeException.class, () -> { try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { attempts.incrementAndGet(); CompletableFuture future = new CompletableFuture<>(); @@ -229,6 +248,7 @@ public void testRetryDeadlineExceededException() { }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); } catch (ExecutionException ex) { throw ex.getCause(); @@ -239,7 +259,7 @@ public void testRetryDeadlineExceededException() { assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", - System.currentTimeMillis() - start > 400); + System.currentTimeMillis() - start > 500); assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", @@ -267,8 +287,8 @@ public void testRespectGlobalDeadlineExceeded() { StatusRuntimeException.class, () -> { try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { attempts.incrementAndGet(); CompletableFuture future = new CompletableFuture<>(); @@ -279,6 +299,7 @@ public void testRespectGlobalDeadlineExceeded() { }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); } catch (ExecutionException e) { throw e.getCause(); @@ -313,8 +334,8 @@ public void testGlobalDeadlineExceededAfterAnotherException() { StatusRuntimeException.class, () -> { try { - DEFAULT_ASYNC_RETRYER - .retry( + new GrpcAsyncRetryer<>( + scheduledExecutor, () -> { if (Context.current().getDeadline().isExpired()) { throw new StatusRuntimeException( @@ -326,6 +347,7 @@ public void testGlobalDeadlineExceededAfterAnotherException() { }, new GrpcRetryer.GrpcRetryerOptions(options, null), GetSystemInfoResponse.Capabilities.getDefaultInstance()) + .retry() .get(); } catch (ExecutionException e) { throw e.getCause(); diff --git a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java index 54ca31766b..3124dd7c5b 100644 --- a/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java +++ b/temporal-serviceclient/src/test/java/io/temporal/internal/retryer/GrpcSyncRetryerTest.java @@ -65,9 +65,11 @@ public void testExpiration() { .setExpiration(Duration.ofMillis(500)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { DEFAULT_SYNC_RETRYER.retry( () -> { + attempts.incrementAndGet(); throw new StatusRuntimeException(Status.fromCode(STATUS_CODE)); }, new GrpcRetryer.GrpcRetryerOptions(options, null), @@ -78,7 +80,8 @@ public void testExpiration() { assertEquals(STATUS_CODE, ((StatusRuntimeException) e).getStatus().getCode()); } - assertTrue(System.currentTimeMillis() - start > 400); + assertTrue("Should retry on DATA_LOSS failures.", attempts.get() > 1); + assertTrue(System.currentTimeMillis() - start > 500); } @Test @@ -87,14 +90,17 @@ public void testDoNotRetry() { RpcRetryOptions options = RpcRetryOptions.newBuilder() - .setInitialInterval(Duration.ofMillis(10)) - .setMaximumInterval(Duration.ofMillis(100)) + .setInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) .addDoNotRetry(STATUS_CODE, null) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { DEFAULT_SYNC_RETRYER.retry( () -> { + if (attempts.incrementAndGet() > 1) + fail("We should not retry on exception that we specified to don't retry"); throw new StatusRuntimeException(Status.fromCode(STATUS_CODE)); }, new GrpcRetryer.GrpcRetryerOptions(options, null), @@ -106,20 +112,22 @@ public void testDoNotRetry() { } assertTrue( "We should fail fast on exception that we specified to don't retry", - System.currentTimeMillis() - start < 10_000); + System.currentTimeMillis() - start < 1000); } @Test public void testInterruptedException() { RpcRetryOptions options = RpcRetryOptions.newBuilder() - .setInitialInterval(Duration.ofMillis(10)) - .setMaximumInterval(Duration.ofMillis(100)) + .setInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { DEFAULT_SYNC_RETRYER.retry( () -> { + if (attempts.incrementAndGet() > 1) fail("We should not retry on InterruptedException"); throw new InterruptedException(); }, new GrpcRetryer.GrpcRetryerOptions(options, null), @@ -129,20 +137,23 @@ public void testInterruptedException() { assertTrue(e instanceof CancellationException); } assertTrue( - "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 10_000); + "We should fail fast on InterruptedException", System.currentTimeMillis() - start < 1000); } @Test public void testNotStatusRuntimeException() { RpcRetryOptions options = RpcRetryOptions.newBuilder() - .setInitialInterval(Duration.ofMillis(10)) - .setMaximumInterval(Duration.ofMillis(100)) + .setInitialInterval(Duration.ofMillis(1000)) + .setMaximumInterval(Duration.ofMillis(1000)) .validateBuildWithDefaults(); long start = System.currentTimeMillis(); + final AtomicInteger attempts = new AtomicInteger(); try { DEFAULT_SYNC_RETRYER.retry( () -> { + if (attempts.incrementAndGet() > 1) + fail("We should not retry if the exception is not StatusRuntimeException"); throw new IllegalArgumentException("simulated"); }, new GrpcRetryer.GrpcRetryerOptions(options, null), @@ -153,8 +164,8 @@ public void testNotStatusRuntimeException() { assertEquals("simulated", e.getMessage()); } assertTrue( - "If the exception is not StatusRuntimeException - we shouldn't retry", - System.currentTimeMillis() - start < 10_000); + "Should fail fast if the exception is not StatusRuntimeException", + System.currentTimeMillis() - start < 1000); } @Test @@ -184,7 +195,7 @@ public void testRetryDeadlineExceededException() { assertEquals(Status.Code.DEADLINE_EXCEEDED, e.getStatus().getCode()); assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.", - System.currentTimeMillis() - start > 400); + System.currentTimeMillis() - start > 500); assertTrue( "We should retry DEADLINE_EXCEEDED if global Grpc Deadline, attempts, time are not exhausted.",