From e29ae62f5a6c3d3911bc44558ea4de970630cfc5 Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Fri, 3 Jul 2026 16:51:01 -0300 Subject: [PATCH 1/2] Fix NPE in TryExecutor when retry policy omits optional fields The spec defines backoff, limit, and delay as optional in a retry policy, but the runtime assumed all three were always present, causing NullPointerException when any was omitted. - Default to constant backoff when backoff is not specified - Default to unlimited retries when limit is not specified - Default to zero delay when delay is not specified Signed-off-by: Matheus Cruz --- .../fluent/test/FuncTryCatchTest.java | 130 ++++++++++++++++++ .../impl/executors/TryExecutor.java | 17 ++- .../retry/AbstractRetryIntervalFunction.java | 6 +- 3 files changed, 150 insertions(+), 3 deletions(-) diff --git a/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java b/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java index db33b5a53..2c0371458 100644 --- a/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java +++ b/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java @@ -32,6 +32,7 @@ */ import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function; +import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.tasks; import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.tryCatch; import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat; @@ -43,6 +44,8 @@ import io.serverlessworkflow.impl.WorkflowException; import io.serverlessworkflow.impl.WorkflowModel; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -59,6 +62,8 @@ public class FuncTryCatchTest { private static final String ORDER_002 = "ORDER#002"; private static final String ORDER_003 = "ORDER#003"; + private static final String TRANSIENT_ERROR = "ERR_TRANSIENT"; + @Test void booking_compensation_dsl() { @@ -447,6 +452,131 @@ void testCatchAll_WithAnyError() { } } + @Test + void testRetryWithoutBackoff() { + AtomicInteger attempts = new AtomicInteger(); + + Workflow workflow = + FuncWorkflowBuilder.workflow() + .tasks( + tryCatch( + "tryTask", + t -> + t.tryCatch( + tasks( + function( + "riskyTask", + (String input) -> { + if (attempts.incrementAndGet() <= 2) { + throw new WorkflowException( + WorkflowError.error(TRANSIENT_ERROR, 503).build()); + } + return "success"; + }, + String.class))) + .catchHandler( + handler -> + handler + .errorsWith(err -> err.type(TRANSIENT_ERROR)) + .retry( + retry -> + retry + .delay(d -> d.milliseconds(10)) + .limit( + limit -> limit.attempt(a -> a.count(3))))))) + .build(); + + try (WorkflowApplication application = WorkflowApplication.builder().build()) { + WorkflowDefinition definition = application.workflowDefinition(workflow); + WorkflowModel result = definition.instance("input").start().join(); + Assertions.assertThat(result.asText()).hasValue("success"); + Assertions.assertThat(attempts.get()).isEqualTo(3); + } + } + + @Test + void testRetryWithoutLimit() { + AtomicInteger attempts = new AtomicInteger(); + + Workflow workflow = + FuncWorkflowBuilder.workflow() + .tasks( + tryCatch( + "tryTask", + t -> + t.tryCatch( + tasks( + function( + "riskyTask", + (String input) -> { + if (attempts.incrementAndGet() <= 2) { + throw new WorkflowException( + WorkflowError.error(TRANSIENT_ERROR, 503).build()); + } + return "success"; + }, + String.class))) + .catchHandler( + handler -> + handler + .errorsWith(err -> err.type(TRANSIENT_ERROR)) + .retry( + retry -> + retry + .delay(d -> d.milliseconds(10)) + .backoff(b -> b.constant("c", "10")))))) + .build(); + + try (WorkflowApplication application = WorkflowApplication.builder().build()) { + WorkflowDefinition definition = application.workflowDefinition(workflow); + WorkflowModel result = definition.instance("input").start().join(); + Assertions.assertThat(result.asText()).hasValue("success"); + Assertions.assertThat(attempts.get()).isEqualTo(3); + } + } + + @Test + void testRetryWithoutDelay() { + AtomicInteger attempts = new AtomicInteger(); + + Workflow workflow = + FuncWorkflowBuilder.workflow() + .tasks( + tryCatch( + "tryTask", + t -> + t.tryCatch( + tasks( + function( + "riskyTask", + (String input) -> { + if (attempts.incrementAndGet() <= 2) { + throw new WorkflowException( + WorkflowError.error(TRANSIENT_ERROR, 503).build()); + } + return "success"; + }, + String.class))) + .catchHandler( + handler -> + handler + .errorsWith(err -> err.type(TRANSIENT_ERROR)) + .retry( + retry -> + retry + .backoff(b -> b.constant("c", "10")) + .limit( + limit -> limit.attempt(a -> a.count(3))))))) + .build(); + + try (WorkflowApplication application = WorkflowApplication.builder().build()) { + WorkflowDefinition definition = application.workflowDefinition(workflow); + WorkflowModel result = definition.instance("input").start().join(); + Assertions.assertThat(result.asText()).hasValue("success"); + Assertions.assertThat(attempts.get()).isEqualTo(3); + } + } + public String reserveStock(String order) { log.info("Reserving stock for order: {}", order); if (order.equals(ORDER_001)) { diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index 00e765d5e..5328d93e2 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -19,6 +19,8 @@ import io.serverlessworkflow.api.types.ErrorFilter; import io.serverlessworkflow.api.types.Retry; import io.serverlessworkflow.api.types.RetryBackoff; +import io.serverlessworkflow.api.types.RetryLimit; +import io.serverlessworkflow.api.types.RetryLimitAttempt; import io.serverlessworkflow.api.types.RetryPolicy; import io.serverlessworkflow.api.types.TaskItem; import io.serverlessworkflow.api.types.TryTask; @@ -107,15 +109,26 @@ private Optional buildRetryInterval(Retry retry) { protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { return new DefaultRetryExecutor( - retryPolicy.getLimit().getAttempt().getCount(), + resolveMaxAttempts(retryPolicy.getLimit()), buildIntervalFunction(retryPolicy), WorkflowUtils.optionalPredicate(application, retryPolicy.getWhen()), WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen())); } + private static int resolveMaxAttempts(RetryLimit limit) { + if (limit == null) { + return Integer.MAX_VALUE; + } + RetryLimitAttempt attempt = limit.getAttempt(); + if (attempt == null) { + return Integer.MAX_VALUE; + } + return attempt.getCount(); + } + private RetryIntervalFunction buildIntervalFunction(RetryPolicy retryPolicy) { RetryBackoff backoff = retryPolicy.getBackoff(); - if (backoff.getConstantBackoff() != null) { + if (backoff == null || backoff.getConstantBackoff() != null) { return new ConstantRetryIntervalFunction( application, retryPolicy.getDelay(), retryPolicy.getJitter()); } else if (backoff.getLinearBackoff() != null) { diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/AbstractRetryIntervalFunction.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/AbstractRetryIntervalFunction.java index 8779f444d..2e53d096b 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/AbstractRetryIntervalFunction.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/AbstractRetryIntervalFunction.java @@ -32,6 +32,9 @@ public abstract class AbstractRetryIntervalFunction implements RetryIntervalFunc private final Optional> maxJitteringResolver; private final WorkflowValueResolver delayResolver; + private static final WorkflowValueResolver ZERO_DURATION_RESOLVER = + (w, t, m) -> Duration.ZERO; + public AbstractRetryIntervalFunction( WorkflowApplication appl, TimeoutAfter delay, RetryPolicyJitter jitter) { if (jitter != null) { @@ -41,7 +44,8 @@ public AbstractRetryIntervalFunction( minJitteringResolver = Optional.empty(); maxJitteringResolver = Optional.empty(); } - delayResolver = WorkflowUtils.fromTimeoutAfter(appl, delay); + delayResolver = + delay != null ? WorkflowUtils.fromTimeoutAfter(appl, delay) : ZERO_DURATION_RESOLVER; } @Override From c16a1423f511168c707a80f0fc79d85d5cf4d0a0 Mon Sep 17 00:00:00 2001 From: Matheus Cruz Date: Fri, 3 Jul 2026 17:03:32 -0300 Subject: [PATCH 2/2] Default to 0 retries when limit is not specified The retry limit default was set to Integer.MAX_VALUE, but when no limit is configured the retry should not execute. Also update the test to verify the workflow fails when no retry limit is set. Signed-off-by: Matheus Cruz --- .../fluent/test/FuncTryCatchTest.java | 13 ++++--------- .../impl/executors/TryExecutor.java | 11 +++-------- 2 files changed, 7 insertions(+), 17 deletions(-) diff --git a/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java b/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java index 2c0371458..8d8fcc808 100644 --- a/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java +++ b/experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java @@ -496,7 +496,6 @@ void testRetryWithoutBackoff() { @Test void testRetryWithoutLimit() { - AtomicInteger attempts = new AtomicInteger(); Workflow workflow = FuncWorkflowBuilder.workflow() @@ -509,11 +508,8 @@ void testRetryWithoutLimit() { function( "riskyTask", (String input) -> { - if (attempts.incrementAndGet() <= 2) { - throw new WorkflowException( - WorkflowError.error(TRANSIENT_ERROR, 503).build()); - } - return "success"; + throw new WorkflowException( + WorkflowError.error(TRANSIENT_ERROR, 503).build()); }, String.class))) .catchHandler( @@ -529,9 +525,8 @@ void testRetryWithoutLimit() { try (WorkflowApplication application = WorkflowApplication.builder().build()) { WorkflowDefinition definition = application.workflowDefinition(workflow); - WorkflowModel result = definition.instance("input").start().join(); - Assertions.assertThat(result.asText()).hasValue("success"); - Assertions.assertThat(attempts.get()).isEqualTo(3); + Assertions.assertThatThrownBy(() -> definition.instance("input").start().join()) + .hasCauseInstanceOf(WorkflowException.class); } } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java index 5328d93e2..6a7d6c050 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java @@ -20,7 +20,6 @@ import io.serverlessworkflow.api.types.Retry; import io.serverlessworkflow.api.types.RetryBackoff; import io.serverlessworkflow.api.types.RetryLimit; -import io.serverlessworkflow.api.types.RetryLimitAttempt; import io.serverlessworkflow.api.types.RetryPolicy; import io.serverlessworkflow.api.types.TaskItem; import io.serverlessworkflow.api.types.TryTask; @@ -116,14 +115,10 @@ protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) { } private static int resolveMaxAttempts(RetryLimit limit) { - if (limit == null) { - return Integer.MAX_VALUE; + if (limit == null || limit.getAttempt() == null) { + return 0; } - RetryLimitAttempt attempt = limit.getAttempt(); - if (attempt == null) { - return Integer.MAX_VALUE; - } - return attempt.getCount(); + return limit.getAttempt().getCount(); } private RetryIntervalFunction buildIntervalFunction(RetryPolicy retryPolicy) {