Add Failure Policy for Jobs SDK

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

@@ -14,7 +14,10 @@
 package io.dapr.it.testcontainers.jobs;
 
 import io.dapr.client.DaprPreviewClient;
+import io.dapr.client.domain.ConstantFailurePolicy;
 import io.dapr.client.domain.DeleteJobRequest;
+import io.dapr.client.domain.DropFailurePolicy;
+import io.dapr.client.domain.FailurePolicyType;
 import io.dapr.client.domain.GetJobRequest;
 import io.dapr.client.domain.GetJobResponse;
 import io.dapr.client.domain.JobSchedule;
@@ -25,6 +28,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
+import org.junit.runner.notification.Failure;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
@@ -34,6 +38,7 @@
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
@@ -97,6 +102,9 @@ public void testJobScheduleCreationWithDueTime() {
 
     GetJobResponse getJobResponse =
         daprPreviewClient.getJob(new GetJobRequest("Job")).block();
+
+    daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
+
     assertEquals(iso8601Formatter.format(currentTime), getJobResponse.getDueTime().toString());
     assertEquals("Job", getJobResponse.getName());
   }
@@ -112,6 +120,9 @@ public void testJobScheduleCreationWithSchedule() {
 
     GetJobResponse getJobResponse =
         daprPreviewClient.getJob(new GetJobRequest("Job")).block();
+
+    daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
+
     assertEquals(iso8601Formatter.format(currentTime), getJobResponse.getDueTime().toString());
     assertEquals(JobSchedule.hourly().getExpression(), getJobResponse.getSchedule().getExpression());
     assertEquals("Job", getJobResponse.getName());
@@ -134,6 +145,9 @@ public void testJobScheduleCreationWithAllParameters() {
 
     GetJobResponse getJobResponse =
         daprPreviewClient.getJob(new GetJobRequest("Job")).block();
+
+    daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
+
     assertEquals(iso8601Formatter.format(currentTime), getJobResponse.getDueTime().toString());
     assertEquals("2 * 3 * * FRI", getJobResponse.getSchedule().getExpression());
     assertEquals("Job", getJobResponse.getName());
@@ -143,6 +157,57 @@ public void testJobScheduleCreationWithAllParameters() {
             getJobResponse.getTtl().toString());
   }
 
+  @Test
+  public void testJobScheduleCreationWithDropFailurePolicy() {
+    Instant currentTime = Instant.now();
+    DateTimeFormatter iso8601Formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+        .withZone(ZoneOffset.UTC);
+
+    String cronExpression = "2 * 3 * * FRI";
+
+    daprPreviewClient.scheduleJob(new ScheduleJobRequest("Job", currentTime)
+        .setTtl(currentTime.plus(2, ChronoUnit.HOURS))
+        .setData("Job data".getBytes())
+        .setRepeat(3)
+            .setFailurePolicy(new DropFailurePolicy())
+        .setSchedule(JobSchedule.fromString(cronExpression))).block();
+
+    GetJobResponse getJobResponse =
+        daprPreviewClient.getJob(new GetJobRequest("Job")).block();
+
+    daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
+
+    assertEquals(FailurePolicyType.DROP, getJobResponse.getFailurePolicy().getFailurePolicyType());
+  }
+
+  @Test
+  public void testJobScheduleCreationWithConstantFailurePolicy() {
+    Instant currentTime = Instant.now();
+    DateTimeFormatter iso8601Formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
+        .withZone(ZoneOffset.UTC);
+
+    String cronExpression = "2 * 3 * * FRI";
+
+    daprPreviewClient.scheduleJob(new ScheduleJobRequest("Job", currentTime)
+        .setTtl(currentTime.plus(2, ChronoUnit.HOURS))
+        .setData("Job data".getBytes())
+        .setRepeat(3)
+        .setFailurePolicy(new ConstantFailurePolicy(3)
+            .setDurationBetweenRetries(Duration.of(10, ChronoUnit.SECONDS)))
+        .setSchedule(JobSchedule.fromString(cronExpression))).block();
+
+    GetJobResponse getJobResponse =
+        daprPreviewClient.getJob(new GetJobRequest("Job")).block();
+
+    daprPreviewClient.deleteJob(new DeleteJobRequest("Job")).block();
+
+    ConstantFailurePolicy jobFailurePolicyConstant = (ConstantFailurePolicy) getJobResponse.getFailurePolicy();
+    assertEquals(FailurePolicyType.CONSTANT, getJobResponse.getFailurePolicy().getFailurePolicyType());
+    assertEquals(3, (int)jobFailurePolicyConstant.getMaxRetries());
+    assertEquals(Duration.of(10, ChronoUnit.SECONDS).getNano(),
+        jobFailurePolicyConstant.getDurationBetweenRetries().getNano());
+  }
+
   @Test
   public void testDeleteJobRequest() {
     Instant currentTime = Instant.now();

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

@@ -27,14 +27,18 @@
 import io.dapr.client.domain.CloudEvent;
 import io.dapr.client.domain.ComponentMetadata;
 import io.dapr.client.domain.ConfigurationItem;
+import io.dapr.client.domain.ConstantFailurePolicy;
 import io.dapr.client.domain.ConversationInput;
 import io.dapr.client.domain.ConversationOutput;
 import io.dapr.client.domain.ConversationRequest;
 import io.dapr.client.domain.ConversationResponse;
 import io.dapr.client.domain.DaprMetadata;
 import io.dapr.client.domain.DeleteJobRequest;
 import io.dapr.client.domain.DeleteStateRequest;
+import io.dapr.client.domain.DropFailurePolicy;
 import io.dapr.client.domain.ExecuteStateTransactionRequest;
+import io.dapr.client.domain.FailurePolicy;
+import io.dapr.client.domain.FailurePolicyType;
 import io.dapr.client.domain.GetBulkSecretRequest;
 import io.dapr.client.domain.GetBulkStateRequest;
 import io.dapr.client.domain.GetConfigurationRequest;
@@ -105,6 +109,7 @@
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -1336,6 +1341,10 @@ public Mono<Void> scheduleJob(ScheduleJobRequest scheduleJobRequest) {
         scheduleJobRequestBuilder.setDueTime(iso8601Formatter.format(scheduleJobRequest.getDueTime()));
       }
 
+      if (scheduleJobRequest.getFailurePolicy() != null) {
+        scheduleJobRequestBuilder.setFailurePolicy(getJobFailurePolicy(scheduleJobRequest.getFailurePolicy()));
+      }
+
       scheduleJobRequestBuilder.setOverwrite(scheduleJobRequest.getOverwrite());
 
       Mono<DaprProtos.ScheduleJobResponse> scheduleJobResponseMono =
@@ -1380,6 +1389,10 @@ public Mono<GetJobResponse> getJob(GetJobRequest getJobRequest) {
           getJobResponse = new GetJobResponse(job.getName(), Instant.parse(job.getDueTime()));
         }
 
+        if (job.hasFailurePolicy()) {
+          getJobResponse.setFailurePolicy(getJobFailurePolicy(job.getFailurePolicy()));
+        }
+
         return getJobResponse
             .setTtl(job.hasTtl() ? Instant.parse(job.getTtl()) : null)
             .setData(job.hasData() ? job.getData().getValue().toByteArray() : null)
@@ -1390,6 +1403,53 @@ public Mono<GetJobResponse> getJob(GetJobRequest getJobRequest) {
     }
   }
 
+  private FailurePolicy getJobFailurePolicy(CommonProtos.JobFailurePolicy jobFailurePolicy) {
+    if (jobFailurePolicy.hasDrop()) {
+      return new DropFailurePolicy();
+    }
+
+    CommonProtos.JobFailurePolicyConstant jobFailurePolicyConstant = jobFailurePolicy.getConstant();
+    if (jobFailurePolicyConstant.hasInterval() && jobFailurePolicyConstant.hasMaxRetries()) {
+      return new ConstantFailurePolicy(jobFailurePolicyConstant.getMaxRetries())
+              .setDurationBetweenRetries(Duration.of(jobFailurePolicyConstant.getInterval().getNanos(),
+                  ChronoUnit.NANOS));
+    }
+
+    if (jobFailurePolicyConstant.hasMaxRetries()) {
+      return new ConstantFailurePolicy(jobFailurePolicyConstant.getMaxRetries());
+    }
+
+    return new ConstantFailurePolicy(
+        Duration.of(jobFailurePolicyConstant.getInterval().getNanos(),
+            ChronoUnit.NANOS));
+  }
+
+  private CommonProtos.JobFailurePolicy getJobFailurePolicy(FailurePolicy failurePolicy) {
+    CommonProtos.JobFailurePolicy.Builder jobFailurePolicyBuilder = CommonProtos.JobFailurePolicy.newBuilder();
+
+    if (failurePolicy.getFailurePolicyType() == FailurePolicyType.DROP) {
+      jobFailurePolicyBuilder.setDrop(CommonProtos.JobFailurePolicyDrop.newBuilder().build());
+      return jobFailurePolicyBuilder.build();
+    }
+
+    CommonProtos.JobFailurePolicyConstant.Builder constantPolicyBuilder =
+        CommonProtos.JobFailurePolicyConstant.newBuilder();
+    ConstantFailurePolicy jobConstantFailurePolicy = (ConstantFailurePolicy)failurePolicy;
+
+    if (jobConstantFailurePolicy.getMaxRetries() != null) {
+      constantPolicyBuilder.setMaxRetries(jobConstantFailurePolicy.getMaxRetries());
+    }
+
+    if (jobConstantFailurePolicy.getDurationBetweenRetries() != null) {
+      constantPolicyBuilder.setInterval(com.google.protobuf.Duration.newBuilder()
+          .setNanos(jobConstantFailurePolicy.getDurationBetweenRetries().getNano()).build());
+    }
+
+    jobFailurePolicyBuilder.setConstant(constantPolicyBuilder.build());
+
+    return jobFailurePolicyBuilder.build();
+  }
+
   /**
    * {@inheritDoc}
    */

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2021 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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.dapr.client.domain;
+
+import java.time.Duration;
+
+/**
+ * A failure policy that applies a constant retry interval for job retries.
+ * This implementation of {@link FailurePolicy} retries a job a fixed number of times
+ * with a constant delay between each retry attempt.
+ */
+public class ConstantFailurePolicy implements FailurePolicy {
+
+  private Integer maxRetries;
+  private Duration durationBetweenRetries;
+
+  /**
+   * Constructs a {@code JobConstantFailurePolicy} with the specified maximum number of retries.
+   *
+   * @param maxRetries the maximum number of retries
+   */
+  public ConstantFailurePolicy(Integer maxRetries) {
+    this.maxRetries = maxRetries;
+  }
+
+  /**
+   * Constructs a {@code JobConstantFailurePolicy} with the specified duration between retries.
+   *
+   * @param durationBetweenRetries the duration to wait between retries
+   */
+  public ConstantFailurePolicy(Duration durationBetweenRetries) {
+    this.durationBetweenRetries = durationBetweenRetries;
+  }
+
+  /**
+   * Sets the duration to wait between retry attempts.
+   *
+   * @param durationBetweenRetries the duration between retries
+   * @return a {@code JobFailurePolicyConstant}.
+   */
+  public ConstantFailurePolicy setDurationBetweenRetries(Duration durationBetweenRetries) {
+    this.durationBetweenRetries = durationBetweenRetries;
+    return this;
+  }
+
+  /**
+   * Sets the maximum number of retries allowed.
+   *
+   * @param maxRetries the number of retries
+   * @return a {@code JobFailurePolicyConstant}.
+   */
+  public ConstantFailurePolicy setMaxRetries(int maxRetries) {
+    this.maxRetries = maxRetries;
+    return this;
+  }
+
+  /**
+   * Returns the configured duration between retry attempts.
+   *
+   * @return the duration between retries
+   */
+  public Duration getDurationBetweenRetries() {
+    return this.durationBetweenRetries;
+  }
+
+  /**
+   * Returns the configured maximum number of retries.
+   *
+   * @return the maximum number of retries
+   */
+  public Integer getMaxRetries() {
+    return this.maxRetries;
+  }
+
+  /**
+   * Returns the type of failure policy.
+   *
+   * @return {@link FailurePolicyType#CONSTANT}
+   */
+  @Override
+  public FailurePolicyType getFailurePolicyType() {
+    return FailurePolicyType.CONSTANT;
+  }
+}

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2021 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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.dapr.client.domain;
+
+/**
+ * A failure policy that drops the job upon failure without retrying.
+ * This implementation of {@link FailurePolicy} immediately discards failed jobs
+ * instead of retrying them.
+ */
+public class DropFailurePolicy implements FailurePolicy {
+
+  /**
+   * Returns the type of failure policy.
+   *
+   * @return {@link FailurePolicyType#DROP}
+   */
+  @Override
+  public FailurePolicyType getFailurePolicyType() {
+    return FailurePolicyType.DROP;
+  }
+}

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

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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.dapr.client.domain;
+
+/**
+ * Set a failure policy for the job.
+ */
+public interface FailurePolicy {
+  FailurePolicyType getFailurePolicyType();
+}

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

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2025 The Dapr Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file 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.dapr.client.domain;
+
+public enum FailurePolicyType {
+  DROP,
+
+  CONSTANT
+}