Skip to content

Commit 7c3afb3

Browse files
siri-varmaartur-ciocanusalaboy
committed
Add Failure Policy for Jobs SDK (#1448)
* Update CONTRIBUTING.md Signed-off-by: Siri Varma Vegiraju <[email protected]> * Add failrue policy Signed-off-by: sirivarma <[email protected]> * Add tests Signed-off-by: sirivarma <[email protected]> * Add Tests Signed-off-by: sirivarma <[email protected]> * Upgrading to 1.15.7 (#1458) * upgrading to 1.15.7 Signed-off-by: salaboy <[email protected]> * using DAPR VERSION Signed-off-by: salaboy <[email protected]> --------- Signed-off-by: salaboy <[email protected]> Signed-off-by: siri-varma <[email protected]> * Rename classes Signed-off-by: siri-varma <[email protected]> * add rc Signed-off-by: sirivarma <[email protected]> * fix checkstyle Signed-off-by: sirivarma <[email protected]> * Fix things Signed-off-by: sirivarma <[email protected]> * Test latest Signed-off-by: sirivarma <[email protected]> * fix checkstyle Signed-off-by: sirivarma <[email protected]> * Address comments Signed-off-by: sirivarma <[email protected]> * Address comments Signed-off-by: sirivarma <[email protected]> --------- Signed-off-by: Siri Varma Vegiraju <[email protected]> Signed-off-by: sirivarma <[email protected]> Signed-off-by: salaboy <[email protected]> Signed-off-by: siri-varma <[email protected]> Co-authored-by: artur-ciocanu <[email protected]> Co-authored-by: salaboy <[email protected]> Signed-off-by: sirivarma <[email protected]>
1 parent 48ad25e commit 7c3afb3

File tree

9 files changed

+609
-5
lines changed

9 files changed

+609
-5
lines changed

‎sdk-tests/src/test/java/io/dapr/it/testcontainers/jobs/DaprJobsIT.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,10 @@
1414
package io.dapr.it.testcontainers.jobs;
1515

1616
import io.dapr.client.DaprPreviewClient;
17+
import io.dapr.client.domain.ConstantFailurePolicy;
1718
import io.dapr.client.domain.DeleteJobRequest;
19+
import io.dapr.client.domain.DropFailurePolicy;
20+
import io.dapr.client.domain.FailurePolicyType;
1821
import io.dapr.client.domain.GetJobRequest;
1922
import io.dapr.client.domain.GetJobResponse;
2023
import io.dapr.client.domain.JobSchedule;
@@ -25,6 +28,7 @@
2528
import org.junit.jupiter.api.BeforeEach;
2629
import org.junit.jupiter.api.Tag;
2730
import org.junit.jupiter.api.Test;
31+
import org.junit.runner.notification.Failure;
2832
import org.springframework.beans.factory.annotation.Autowired;
2933
import org.springframework.boot.test.context.SpringBootTest;
3034
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
@@ -34,6 +38,7 @@
3438
import org.testcontainers.junit.jupiter.Container;
3539
import org.testcontainers.junit.jupiter.Testcontainers;
3640

41+
import java.time.Duration;
3742
import java.time.Instant;
3843
import java.time.ZoneOffset;
3944
import java.time.format.DateTimeFormatter;
@@ -97,6 +102,9 @@ public void testJobScheduleCreationWithDueTime() {
97102

98103
GetJobResponse getJobResponse =
99104
daprPreviewClient.getJob(new GetJobRequest("Job")).block();
105+
106+
daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
107+
100108
assertEquals(iso8601Formatter.format(currentTime), getJobResponse.getDueTime().toString());
101109
assertEquals("Job", getJobResponse.getName());
102110
}
@@ -112,6 +120,9 @@ public void testJobScheduleCreationWithSchedule() {
112120

113121
GetJobResponse getJobResponse =
114122
daprPreviewClient.getJob(new GetJobRequest("Job")).block();
123+
124+
daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
125+
115126
assertEquals(iso8601Formatter.format(currentTime), getJobResponse.getDueTime().toString());
116127
assertEquals(JobSchedule.hourly().getExpression(), getJobResponse.getSchedule().getExpression());
117128
assertEquals("Job", getJobResponse.getName());
@@ -134,6 +145,9 @@ public void testJobScheduleCreationWithAllParameters() {
134145

135146
GetJobResponse getJobResponse =
136147
daprPreviewClient.getJob(new GetJobRequest("Job")).block();
148+
149+
daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
150+
137151
assertEquals(iso8601Formatter.format(currentTime), getJobResponse.getDueTime().toString());
138152
assertEquals("2 * 3 * * FRI", getJobResponse.getSchedule().getExpression());
139153
assertEquals("Job", getJobResponse.getName());
@@ -143,6 +157,57 @@ public void testJobScheduleCreationWithAllParameters() {
143157
getJobResponse.getTtl().toString());
144158
}
145159

160+
@Test
161+
public void testJobScheduleCreationWithDropFailurePolicy() {
162+
Instant currentTime = Instant.now();
163+
DateTimeFormatter iso8601Formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
164+
.withZone(ZoneOffset.UTC);
165+
166+
String cronExpression = "2 * 3 * * FRI";
167+
168+
daprPreviewClient.scheduleJob(new ScheduleJobRequest("Job", currentTime)
169+
.setTtl(currentTime.plus(2, ChronoUnit.HOURS))
170+
.setData("Job data".getBytes())
171+
.setRepeat(3)
172+
.setFailurePolicy(new DropFailurePolicy())
173+
.setSchedule(JobSchedule.fromString(cronExpression))).block();
174+
175+
GetJobResponse getJobResponse =
176+
daprPreviewClient.getJob(new GetJobRequest("Job")).block();
177+
178+
daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
179+
180+
assertEquals(FailurePolicyType.DROP, getJobResponse.getFailurePolicy().getFailurePolicyType());
181+
}
182+
183+
@Test
184+
public void testJobScheduleCreationWithConstantFailurePolicy() {
185+
Instant currentTime = Instant.now();
186+
DateTimeFormatter iso8601Formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
187+
.withZone(ZoneOffset.UTC);
188+
189+
String cronExpression = "2 * 3 * * FRI";
190+
191+
daprPreviewClient.scheduleJob(new ScheduleJobRequest("Job", currentTime)
192+
.setTtl(currentTime.plus(2, ChronoUnit.HOURS))
193+
.setData("Job data".getBytes())
194+
.setRepeat(3)
195+
.setFailurePolicy(new ConstantFailurePolicy(3)
196+
.setDurationBetweenRetries(Duration.of(10, ChronoUnit.SECONDS)))
197+
.setSchedule(JobSchedule.fromString(cronExpression))).block();
198+
199+
GetJobResponse getJobResponse =
200+
daprPreviewClient.getJob(new GetJobRequest("Job")).block();
201+
202+
daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
203+
204+
ConstantFailurePolicy jobFailurePolicyConstant = (ConstantFailurePolicy) getJobResponse.getFailurePolicy();
205+
assertEquals(FailurePolicyType.CONSTANT, getJobResponse.getFailurePolicy().getFailurePolicyType());
206+
assertEquals(3, (int)jobFailurePolicyConstant.getMaxRetries());
207+
assertEquals(Duration.of(10, ChronoUnit.SECONDS).getNano(),
208+
jobFailurePolicyConstant.getDurationBetweenRetries().getNano());
209+
}
210+
146211
@Test
147212
public void testDeleteJobRequest() {
148213
Instant currentTime = Instant.now();

‎sdk/src/main/java/io/dapr/client/DaprClientImpl.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import io.dapr.client.domain.ComponentMetadata;
3232
import io.dapr.client.domain.ConfigurationItem;
3333
import io.dapr.client.domain.ConversationToolsFunction;
34+
import io.dapr.client.domain.ConstantFailurePolicy;
3435
import io.dapr.client.domain.ConversationInput;
3536
import io.dapr.client.domain.ConversationInputAlpha2;
3637
import io.dapr.client.domain.ConversationMessage;
@@ -49,7 +50,10 @@
4950
import io.dapr.client.domain.DaprMetadata;
5051
import io.dapr.client.domain.DeleteJobRequest;
5152
import io.dapr.client.domain.DeleteStateRequest;
53+
import io.dapr.client.domain.DropFailurePolicy;
5254
import io.dapr.client.domain.ExecuteStateTransactionRequest;
55+
import io.dapr.client.domain.FailurePolicy;
56+
import io.dapr.client.domain.FailurePolicyType;
5357
import io.dapr.client.domain.GetBulkSecretRequest;
5458
import io.dapr.client.domain.GetBulkStateRequest;
5559
import io.dapr.client.domain.GetConfigurationRequest;
@@ -122,6 +126,7 @@
122126
import java.time.Instant;
123127
import java.time.ZoneOffset;
124128
import java.time.format.DateTimeFormatter;
129+
import java.time.temporal.ChronoUnit;
125130
import java.util.ArrayList;
126131
import java.util.Arrays;
127132
import java.util.Collections;
@@ -1353,6 +1358,10 @@ public Mono<Void> scheduleJob(ScheduleJobRequest scheduleJobRequest) {
13531358
scheduleJobRequestBuilder.setDueTime(iso8601Formatter.format(scheduleJobRequest.getDueTime()));
13541359
}
13551360

1361+
if (scheduleJobRequest.getFailurePolicy() != null) {
1362+
scheduleJobRequestBuilder.setFailurePolicy(getJobFailurePolicy(scheduleJobRequest.getFailurePolicy()));
1363+
}
1364+
13561365
scheduleJobRequestBuilder.setOverwrite(scheduleJobRequest.getOverwrite());
13571366

13581367
Mono<DaprProtos.ScheduleJobResponse> scheduleJobResponseMono =
@@ -1397,6 +1406,10 @@ public Mono<GetJobResponse> getJob(GetJobRequest getJobRequest) {
13971406
getJobResponse = new GetJobResponse(job.getName(), Instant.parse(job.getDueTime()));
13981407
}
13991408

1409+
if (job.hasFailurePolicy()) {
1410+
getJobResponse.setFailurePolicy(getJobFailurePolicy(job.getFailurePolicy()));
1411+
}
1412+
14001413
return getJobResponse
14011414
.setTtl(job.hasTtl() ? Instant.parse(job.getTtl()) : null)
14021415
.setData(job.hasData() ? job.getData().getValue().toByteArray() : null)
@@ -1407,6 +1420,53 @@ public Mono<GetJobResponse> getJob(GetJobRequest getJobRequest) {
14071420
}
14081421
}
14091422

1423+
private FailurePolicy getJobFailurePolicy(CommonProtos.JobFailurePolicy jobFailurePolicy) {
1424+
if (jobFailurePolicy.hasDrop()) {
1425+
return new DropFailurePolicy();
1426+
}
1427+
1428+
CommonProtos.JobFailurePolicyConstant jobFailurePolicyConstant = jobFailurePolicy.getConstant();
1429+
if (jobFailurePolicyConstant.hasInterval() && jobFailurePolicyConstant.hasMaxRetries()) {
1430+
return new ConstantFailurePolicy(jobFailurePolicyConstant.getMaxRetries())
1431+
.setDurationBetweenRetries(Duration.of(jobFailurePolicyConstant.getInterval().getNanos(),
1432+
ChronoUnit.NANOS));
1433+
}
1434+
1435+
if (jobFailurePolicyConstant.hasMaxRetries()) {
1436+
return new ConstantFailurePolicy(jobFailurePolicyConstant.getMaxRetries());
1437+
}
1438+
1439+
return new ConstantFailurePolicy(
1440+
Duration.of(jobFailurePolicyConstant.getInterval().getNanos(),
1441+
ChronoUnit.NANOS));
1442+
}
1443+
1444+
private CommonProtos.JobFailurePolicy getJobFailurePolicy(FailurePolicy failurePolicy) {
1445+
CommonProtos.JobFailurePolicy.Builder jobFailurePolicyBuilder = CommonProtos.JobFailurePolicy.newBuilder();
1446+
1447+
if (failurePolicy.getFailurePolicyType() == FailurePolicyType.DROP) {
1448+
jobFailurePolicyBuilder.setDrop(CommonProtos.JobFailurePolicyDrop.newBuilder().build());
1449+
return jobFailurePolicyBuilder.build();
1450+
}
1451+
1452+
CommonProtos.JobFailurePolicyConstant.Builder constantPolicyBuilder =
1453+
CommonProtos.JobFailurePolicyConstant.newBuilder();
1454+
ConstantFailurePolicy jobConstantFailurePolicy = (ConstantFailurePolicy)failurePolicy;
1455+
1456+
if (jobConstantFailurePolicy.getMaxRetries() != null) {
1457+
constantPolicyBuilder.setMaxRetries(jobConstantFailurePolicy.getMaxRetries());
1458+
}
1459+
1460+
if (jobConstantFailurePolicy.getDurationBetweenRetries() != null) {
1461+
constantPolicyBuilder.setInterval(com.google.protobuf.Duration.newBuilder()
1462+
.setNanos(jobConstantFailurePolicy.getDurationBetweenRetries().getNano()).build());
1463+
}
1464+
1465+
jobFailurePolicyBuilder.setConstant(constantPolicyBuilder.build());
1466+
1467+
return jobFailurePolicyBuilder.build();
1468+
}
1469+
14101470
/**
14111471
* {@inheritDoc}
14121472
*/

‎sdk/src/main/java/io/dapr/client/domain/ConstantFailurePolicy.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
/*
2+
* Copyright 2021 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.client.domain;
15+
16+
import java.time.Duration;
17+
18+
/**
19+
* A failure policy that applies a constant retry interval for job retries.
20+
* This implementation of {@link FailurePolicy} retries a job a fixed number of times
21+
* with a constant delay between each retry attempt.
22+
*/
23+
public class ConstantFailurePolicy implements FailurePolicy {
24+
25+
private Integer maxRetries;
26+
private Duration durationBetweenRetries;
27+
28+
/**
29+
* Constructs a {@code JobConstantFailurePolicy} with the specified maximum number of retries.
30+
*
31+
* @param maxRetries the maximum number of retries
32+
*/
33+
public ConstantFailurePolicy(Integer maxRetries) {
34+
this.maxRetries = maxRetries;
35+
}
36+
37+
/**
38+
* Constructs a {@code JobConstantFailurePolicy} with the specified duration between retries.
39+
*
40+
* @param durationBetweenRetries the duration to wait between retries
41+
*/
42+
public ConstantFailurePolicy(Duration durationBetweenRetries) {
43+
this.durationBetweenRetries = durationBetweenRetries;
44+
}
45+
46+
/**
47+
* Sets the duration to wait between retry attempts.
48+
*
49+
* @param durationBetweenRetries the duration between retries
50+
* @return a {@code JobFailurePolicyConstant}.
51+
*/
52+
public ConstantFailurePolicy setDurationBetweenRetries(Duration durationBetweenRetries) {
53+
this.durationBetweenRetries = durationBetweenRetries;
54+
return this;
55+
}
56+
57+
/**
58+
* Sets the maximum number of retries allowed.
59+
*
60+
* @param maxRetries the number of retries
61+
* @return a {@code JobFailurePolicyConstant}.
62+
*/
63+
public ConstantFailurePolicy setMaxRetries(int maxRetries) {
64+
this.maxRetries = maxRetries;
65+
return this;
66+
}
67+
68+
/**
69+
* Returns the configured duration between retry attempts.
70+
*
71+
* @return the duration between retries
72+
*/
73+
public Duration getDurationBetweenRetries() {
74+
return this.durationBetweenRetries;
75+
}
76+
77+
/**
78+
* Returns the configured maximum number of retries.
79+
*
80+
* @return the maximum number of retries
81+
*/
82+
public Integer getMaxRetries() {
83+
return this.maxRetries;
84+
}
85+
86+
/**
87+
* Returns the type of failure policy.
88+
*
89+
* @return {@link FailurePolicyType#CONSTANT}
90+
*/
91+
@Override
92+
public FailurePolicyType getFailurePolicyType() {
93+
return FailurePolicyType.CONSTANT;
94+
}
95+
}

‎sdk/src/main/java/io/dapr/client/domain/DropFailurePolicy.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Copyright 2021 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.client.domain;
15+
16+
/**
17+
* A failure policy that drops the job upon failure without retrying.
18+
* This implementation of {@link FailurePolicy} immediately discards failed jobs
19+
* instead of retrying them.
20+
*/
21+
public class DropFailurePolicy implements FailurePolicy {
22+
23+
/**
24+
* Returns the type of failure policy.
25+
*
26+
* @return {@link FailurePolicyType#DROP}
27+
*/
28+
@Override
29+
public FailurePolicyType getFailurePolicyType() {
30+
return FailurePolicyType.DROP;
31+
}
32+
}

‎sdk/src/main/java/io/dapr/client/domain/FailurePolicy.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
/*
2+
* Copyright 2025 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.client.domain;
15+
16+
/**
17+
* Set a failure policy for the job.
18+
*/
19+
public interface FailurePolicy {
20+
FailurePolicyType getFailurePolicyType();
21+
}

‎sdk/src/main/java/io/dapr/client/domain/FailurePolicyType.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
/*
2+
* Copyright 2025 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
limitations under the License.
12+
*/
13+
14+
package io.dapr.client.domain;
15+
16+
public enum FailurePolicyType {
17+
DROP,
18+
19+
CONSTANT
20+
}

0 commit comments

Comments
 (0)