Skip to content

Commit 60baf21

Browse files
mcruzdevfjtirado
mcruzdev
and
fjtirado
authored
Fix NPE in TryExecutor when retry policy omits optional fields (#1519)
* 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 <matheuscruz.dev@gmail.com> * 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 <matheuscruz.dev@gmail.com> * [Fix #1517] Review comments. Signed-off-by: fjtirado <ftirados@ibm.com> --------- Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com> Signed-off-by: fjtirado <ftirados@ibm.com> Co-authored-by: fjtirado <ftirados@ibm.com>
1 parent 0c96a26 commit 60baf21

16 files changed

Lines changed: 216 additions & 66 deletions

File tree

‎experimental/test/src/test/java/io/serverlessworkflow/fluent/test/FuncTryCatchTest.java‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
*/
3333

3434
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function;
35+
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.tasks;
3536
import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.tryCatch;
3637
import static io.serverlessworkflow.fluent.test.TestSerializationUtils.writeAndReadInMemory;
3738
import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
@@ -45,6 +46,8 @@
4546
import io.serverlessworkflow.impl.WorkflowModel;
4647
import java.io.IOException;
4748
import java.util.List;
49+
import java.util.concurrent.atomic.AtomicInteger;
50+
import org.assertj.core.api.Assertions;
4851
import org.junit.jupiter.api.Test;
4952
import org.slf4j.Logger;
5053
import org.slf4j.LoggerFactory;
@@ -61,6 +64,8 @@ public class FuncTryCatchTest {
6164
private static final String ORDER_002 = "ORDER#002";
6265
private static final String ORDER_003 = "ORDER#003";
6366

67+
private static final String TRANSIENT_ERROR = "ERR_TRANSIENT";
68+
6469
@Test
6570
void booking_compensation_dsl() throws IOException {
6671

@@ -472,6 +477,129 @@ void testCatchAll_WithAnyError() throws IOException {
472477
}
473478
}
474479

480+
@Test
481+
void testRetryWithoutBackoff() {
482+
AtomicInteger attempts = new AtomicInteger();
483+
Workflow workflow =
484+
FuncWorkflowBuilder.workflow()
485+
.tasks(
486+
tryCatch(
487+
"tryTask",
488+
t ->
489+
t.tryCatch(
490+
tasks(
491+
function(
492+
"riskyTask",
493+
(String input) -> {
494+
if (attempts.incrementAndGet() <= 2) {
495+
throw new WorkflowException(
496+
WorkflowError.error(TRANSIENT_ERROR, 503).build());
497+
}
498+
return "success";
499+
},
500+
String.class)))
501+
.catchHandler(
502+
handler ->
503+
handler
504+
.errorsWith(err -> err.type(TRANSIENT_ERROR))
505+
.retry(
506+
retry ->
507+
retry
508+
.delay(d -> d.milliseconds(10))
509+
.limit(
510+
limit -> limit.attempt(a -> a.count(3)))))))
511+
.build();
512+
513+
try (WorkflowApplication application = WorkflowApplication.builder().build()) {
514+
WorkflowDefinition definition = application.workflowDefinition(workflow);
515+
WorkflowModel result = definition.instance("input").start().join();
516+
Assertions.assertThat(result.asText()).hasValue("success");
517+
Assertions.assertThat(attempts.get()).isEqualTo(3);
518+
}
519+
}
520+
521+
@Test
522+
void testRetryWithoutLimit() {
523+
AtomicInteger attempts = new AtomicInteger();
524+
Workflow workflow =
525+
FuncWorkflowBuilder.workflow()
526+
.tasks(
527+
tryCatch(
528+
"tryTask",
529+
t ->
530+
t.tryCatch(
531+
tasks(
532+
function(
533+
"riskyTask",
534+
(String input) -> {
535+
if (attempts.incrementAndGet() <= 2) {
536+
throw new WorkflowException(
537+
WorkflowError.error(TRANSIENT_ERROR, 503).build());
538+
}
539+
return "success";
540+
},
541+
String.class)))
542+
.catchHandler(
543+
handler ->
544+
handler
545+
.errorsWith(err -> err.type(TRANSIENT_ERROR))
546+
.retry(
547+
retry ->
548+
retry
549+
.delay(d -> d.milliseconds(10))
550+
.backoff(b -> b.constant("c", "10"))))))
551+
.build();
552+
553+
try (WorkflowApplication application = WorkflowApplication.builder().build()) {
554+
WorkflowDefinition definition = application.workflowDefinition(workflow);
555+
WorkflowModel result = definition.instance("input").start().join();
556+
Assertions.assertThat(result.asText()).hasValue("success");
557+
Assertions.assertThat(attempts.get()).isEqualTo(3);
558+
}
559+
}
560+
561+
@Test
562+
void testRetryWithoutDelay() {
563+
AtomicInteger attempts = new AtomicInteger();
564+
565+
Workflow workflow =
566+
FuncWorkflowBuilder.workflow()
567+
.tasks(
568+
tryCatch(
569+
"tryTask",
570+
t ->
571+
t.tryCatch(
572+
tasks(
573+
function(
574+
"riskyTask",
575+
(String input) -> {
576+
if (attempts.incrementAndGet() <= 2) {
577+
throw new WorkflowException(
578+
WorkflowError.error(TRANSIENT_ERROR, 503).build());
579+
}
580+
return "success";
581+
},
582+
String.class)))
583+
.catchHandler(
584+
handler ->
585+
handler
586+
.errorsWith(err -> err.type(TRANSIENT_ERROR))
587+
.retry(
588+
retry ->
589+
retry
590+
.backoff(b -> b.constant("c", "10"))
591+
.limit(
592+
limit -> limit.attempt(a -> a.count(3)))))))
593+
.build();
594+
595+
try (WorkflowApplication application = WorkflowApplication.builder().build()) {
596+
WorkflowDefinition definition = application.workflowDefinition(workflow);
597+
WorkflowModel result = definition.instance("input").start().join();
598+
Assertions.assertThat(result.asText()).hasValue("success");
599+
Assertions.assertThat(attempts.get()).isEqualTo(3);
600+
}
601+
}
602+
475603
public String reserveStock(String order) {
476604
log.info("Reserving stock for order: {}", order);
477605
if (order.equals(ORDER_001)) {

‎impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,10 @@ public class TaskContext implements TaskContextData {
3737
private WorkflowModel rawOutput;
3838
private Instant completedAt;
3939
private TransitionInfo transition;
40-
private short retryAttempt;
40+
private int retryAttempt;
4141
private int iteration;
4242
private AuthorizationDescriptor authorization;
43-
private Optional<Short> tryRetryCount = Optional.empty();
43+
private Optional<Integer> tryRetryCount = Optional.empty();
4444

4545
public TaskContext(
4646
WorkflowModel input,
@@ -71,7 +71,7 @@ private TaskContext(
7171
this.output = output;
7272
this.rawOutput = rawOutput;
7373
this.retryAttempt =
74-
parentContext.map(ctx -> ctx.tryRetryCount.orElse(ctx.retryAttempt())).orElse((short) 0);
74+
parentContext.map(ctx -> ctx.tryRetryCount.orElse(ctx.retryAttempt())).orElse(0);
7575
this.contextVariables =
7676
parentContext.map(p -> new HashMap<>(p.contextVariables)).orElseGet(HashMap::new);
7777
}
@@ -173,19 +173,19 @@ public boolean isCompleted() {
173173
}
174174

175175
@Override
176-
public short retryAttempt() {
176+
public int retryAttempt() {
177177
return retryAttempt;
178178
}
179179

180-
public void retryAttempt(short retryAttempt) {
180+
public void retryAttempt(int retryAttempt) {
181181
this.retryAttempt = retryAttempt;
182182
}
183183

184-
public void tryRetryCount(short tryRetryCount) {
184+
public void tryRetryCount(int tryRetryCount) {
185185
this.tryRetryCount = Optional.of(tryRetryCount);
186186
}
187187

188-
public Optional<Short> tryRetryCount() {
188+
public Optional<Integer> tryRetryCount() {
189189
return tryRetryCount;
190190
}
191191

‎impl/core/src/main/java/io/serverlessworkflow/impl/TaskContextData.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,5 +40,5 @@ public interface TaskContextData {
4040

4141
int iteration();
4242

43-
short retryAttempt();
43+
int retryAttempt();
4444
}

‎impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java‎

Lines changed: 27 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -187,30 +187,36 @@ public static boolean whenExceptTest(
187187
&& exceptFilter.map(w -> !w.test(workflow, taskContext, model)).orElse(true);
188188
}
189189

190+
private static final class ZeroDelayResolverHolder {
191+
private static final WorkflowValueResolver<Duration> ZERO_DELAY_RESOLVER =
192+
(w, t, f) -> Duration.ZERO;
193+
}
194+
190195
public static WorkflowValueResolver<Duration> fromTimeoutAfter(
191196
WorkflowApplication application, TimeoutAfter timeout) {
192-
if (timeout.getDurationExpression() != null) {
193-
return (w, f, t) ->
194-
Duration.parse(
195-
application
196-
.expressionFactory()
197-
.resolveString(ExpressionDescriptor.from(timeout.getDurationExpression()))
198-
.apply(w, f, t));
199-
} else if (timeout.getDurationLiteral() != null) {
200-
Duration duration = Duration.parse(timeout.getDurationLiteral());
201-
return (w, f, t) -> duration;
202-
} else if (timeout.getDurationInline() != null) {
203-
DurationInline inlineDuration = timeout.getDurationInline();
204-
return (w, t, f) ->
205-
Duration.ofDays(inlineDuration.getDays())
206-
.plus(
207-
Duration.ofHours(inlineDuration.getHours())
208-
.plus(Duration.ofMinutes(inlineDuration.getMinutes()))
209-
.plus(Duration.ofSeconds(inlineDuration.getSeconds()))
210-
.plus(Duration.ofMillis(inlineDuration.getMilliseconds())));
211-
} else {
212-
return (w, t, f) -> Duration.ZERO;
197+
if (timeout != null) {
198+
if (timeout.getDurationExpression() != null) {
199+
return (w, f, t) ->
200+
Duration.parse(
201+
application
202+
.expressionFactory()
203+
.resolveString(ExpressionDescriptor.from(timeout.getDurationExpression()))
204+
.apply(w, f, t));
205+
} else if (timeout.getDurationLiteral() != null) {
206+
Duration duration = Duration.parse(timeout.getDurationLiteral());
207+
return (w, f, t) -> duration;
208+
} else if (timeout.getDurationInline() != null) {
209+
DurationInline inlineDuration = timeout.getDurationInline();
210+
return (w, t, f) ->
211+
Duration.ofDays(inlineDuration.getDays())
212+
.plus(
213+
Duration.ofHours(inlineDuration.getHours())
214+
.plus(Duration.ofMinutes(inlineDuration.getMinutes()))
215+
.plus(Duration.ofSeconds(inlineDuration.getSeconds()))
216+
.plus(Duration.ofMillis(inlineDuration.getMilliseconds())));
217+
}
213218
}
219+
return ZeroDelayResolverHolder.ZERO_DELAY_RESOLVER;
214220
}
215221

216222
public static Optional<WorkflowValueResolver<Duration>> getTaskTimeout(

‎impl/core/src/main/java/io/serverlessworkflow/impl/executors/TryExecutor.java‎

Lines changed: 21 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import io.serverlessworkflow.api.types.ErrorFilter;
2020
import io.serverlessworkflow.api.types.Retry;
2121
import io.serverlessworkflow.api.types.RetryBackoff;
22+
import io.serverlessworkflow.api.types.RetryLimit;
2223
import io.serverlessworkflow.api.types.RetryPolicy;
2324
import io.serverlessworkflow.api.types.TaskItem;
2425
import io.serverlessworkflow.api.types.TryTask;
@@ -107,25 +108,34 @@ private Optional<RetryExecutor> buildRetryInterval(Retry retry) {
107108

108109
protected RetryExecutor buildRetryExecutor(RetryPolicy retryPolicy) {
109110
return new DefaultRetryExecutor(
110-
retryPolicy.getLimit().getAttempt().getCount(),
111+
resolveMaxAttempts(retryPolicy.getLimit()),
111112
buildIntervalFunction(retryPolicy),
112113
WorkflowUtils.optionalPredicate(application, retryPolicy.getWhen()),
113114
WorkflowUtils.optionalPredicate(application, retryPolicy.getExceptWhen()));
114115
}
115116

117+
private static int resolveMaxAttempts(RetryLimit limit) {
118+
return limit != null && limit.getAttempt() != null
119+
? limit.getAttempt().getCount()
120+
: Integer.MAX_VALUE - 1;
121+
}
122+
116123
private RetryIntervalFunction buildIntervalFunction(RetryPolicy retryPolicy) {
117124
RetryBackoff backoff = retryPolicy.getBackoff();
118-
if (backoff.getConstantBackoff() != null) {
119-
return new ConstantRetryIntervalFunction(
120-
application, retryPolicy.getDelay(), retryPolicy.getJitter());
121-
} else if (backoff.getLinearBackoff() != null) {
122-
return new LinearRetryIntervalFunction(
123-
application, retryPolicy.getDelay(), retryPolicy.getJitter());
124-
} else if (backoff.getExponentialBackOff() != null) {
125-
return new ExponentialRetryIntervalFunction(
126-
application, retryPolicy.getDelay(), retryPolicy.getJitter());
125+
if (backoff != null) {
126+
if (backoff.getConstantBackoff() != null) {
127+
return new ConstantRetryIntervalFunction(
128+
application, retryPolicy.getDelay(), retryPolicy.getJitter());
129+
} else if (backoff.getLinearBackoff() != null) {
130+
return new LinearRetryIntervalFunction(
131+
application, retryPolicy.getDelay(), retryPolicy.getJitter());
132+
} else if (backoff.getExponentialBackOff() != null) {
133+
return new ExponentialRetryIntervalFunction(
134+
application, retryPolicy.getDelay(), retryPolicy.getJitter());
135+
}
127136
}
128-
throw new IllegalStateException("A backoff strategy should be set");
137+
return new ConstantRetryIntervalFunction(
138+
application, retryPolicy.getDelay(), retryPolicy.getJitter());
129139
}
130140

131141
@Override

‎impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/AbstractRetryIntervalFunction.java‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -49,21 +49,24 @@ public Duration apply(
4949
WorkflowContext workflowContext,
5050
TaskContext taskContext,
5151
WorkflowModel model,
52-
short numAttempts) {
52+
int numAttempts) {
5353
Duration delay = delayResolver.apply(workflowContext, taskContext, model);
5454
Duration minJittering =
5555
minJitteringResolver
5656
.map(min -> min.apply(workflowContext, taskContext, model))
5757
.orElse(Duration.ZERO);
58-
Duration maxJittering =
58+
Duration result = calcDelay(delay, numAttempts).plus(minJittering);
59+
long maxJittering =
5960
maxJitteringResolver
6061
.map(max -> max.apply(workflowContext, taskContext, model))
61-
.orElse(Duration.ZERO);
62-
return calcDelay(delay, numAttempts)
63-
.plus(
64-
Duration.ofMillis(
65-
(long) (minJittering.toMillis() + Math.random() * maxJittering.toMillis())));
62+
.orElse(Duration.ZERO)
63+
.toMillis();
64+
long diff = maxJittering - minJittering.toMillis();
65+
if (diff > 0) {
66+
result = result.plus(Duration.ofMillis(Math.round(Math.random() * diff)));
67+
}
68+
return result;
6669
}
6770

68-
protected abstract Duration calcDelay(Duration delay, short numAttempts);
71+
protected abstract Duration calcDelay(Duration delay, int numAttempts);
6972
}

‎impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/ConstantRetryIntervalFunction.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ public ConstantRetryIntervalFunction(
2828
}
2929

3030
@Override
31-
protected Duration calcDelay(Duration delay, short numAttempts) {
31+
protected Duration calcDelay(Duration delay, int numAttempts) {
3232
return delay;
3333
}
3434
}

‎impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/DefaultRetryExecutor.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ public DefaultRetryExecutor(
4646
@Override
4747
public Optional<CompletableFuture<WorkflowModel>> retry(
4848
WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel model) {
49-
short numAttempts = taskContext.tryRetryCount().orElseThrow();
49+
int numAttempts = taskContext.tryRetryCount().orElseThrow();
5050
if (numAttempts++ < maxAttempts
5151
&& WorkflowUtils.whenExceptTest(
5252
whenFilter, exceptFilter, workflowContext, taskContext, model)) {
@@ -62,7 +62,7 @@ public Optional<CompletableFuture<WorkflowModel>> retry(
6262
@Override
6363
public void init(WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel model) {
6464
if (taskContext.tryRetryCount().isEmpty()) {
65-
taskContext.tryRetryCount((short) 0);
65+
taskContext.tryRetryCount(0);
6666
}
6767
}
6868
}

‎impl/core/src/main/java/io/serverlessworkflow/impl/executors/retry/ExponentialRetryIntervalFunction.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,16 @@
2222

2323
public class ExponentialRetryIntervalFunction extends AbstractRetryIntervalFunction {
2424

25+
// 2^33 will be several years, which I think is too high as delay
26+
private static final int MAX_EXPONENTIAL_RETRY = 32;
27+
2528
public ExponentialRetryIntervalFunction(
2629
WorkflowApplication appl, TimeoutAfter delay, RetryPolicyJitter jitter) {
2730
super(appl, delay, jitter);
2831
}
2932

3033
@Override
31-
protected Duration calcDelay(Duration delay, short numAttempts) {
32-
return delay.multipliedBy(1 << numAttempts);
34+
protected Duration calcDelay(Duration delay, int numAttempts) {
35+
return delay.multipliedBy(1L << Math.min(numAttempts, MAX_EXPONENTIAL_RETRY));
3336
}
3437
}

0 commit comments

Comments
 (0)