Allow a step to be fatally failed via a RetryArtifact in its output data

maestro-common/src/main/java/com/netflix/maestro/models/artifact/Artifact.java

@@ -33,6 +33,7 @@
   @JsonSubTypes.Type(name = "DYNAMIC_OUTPUT", value = DynamicOutputArtifact.class),
   @JsonSubTypes.Type(name = "KUBERNETES", value = KubernetesArtifact.class),
   @JsonSubTypes.Type(name = "HTTP", value = HttpArtifact.class),
+  @JsonSubTypes.Type(name = "RETRY", value = RetryArtifact.class),
 })
 @SuppressWarnings("PMD.ImplicitFunctionalInterface")
 public interface Artifact {
@@ -58,7 +59,9 @@ enum Type {
     /** kubernetes artifact. */
     KUBERNETES(Constants.MAESTRO_PREFIX + "kubernetes"),
     /** http artifact. */
-    HTTP(Constants.MAESTRO_PREFIX + "http");
+    HTTP(Constants.MAESTRO_PREFIX + "http"),
+    /** retry artifact. */
+    RETRY(Constants.MAESTRO_PREFIX + "retry");
 
     private final String key;
 
@@ -152,4 +155,13 @@ default KubernetesArtifact asKubernetes() {
   default HttpArtifact asHttp() {
     throw new MaestroInternalError("Artifact type [%s] cannot be used as HTTP", getType());
   }
+
+  /**
+   * Get Retry type artifact.
+   *
+   * @return concrete artifact object.
+   */
+  default RetryArtifact asRetry() {
+    throw new MaestroInternalError("Artifact type [%s] cannot be used as RETRY", getType());
+  }
 }

maestro-common/src/main/java/com/netflix/maestro/models/artifact/RetryArtifact.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2024 Netflix, Inc.
+ *
+ * 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 com.netflix.maestro.models.artifact;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Data;
+
+/** Retry artifact for a step to influence whether the system retries it on failure. */
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder(
+    value = {"retryable"},
+    alphabetic = true)
+@Data
+public class RetryArtifact implements Artifact {
+  private boolean retryable = true; // whether the system should retry the step on failure
+
+  @JsonIgnore
+  @Override
+  public RetryArtifact asRetry() {
+    return this;
+  }
+
+  @Override
+  public Type getType() {
+    return Type.RETRY;
+  }
+}

maestro-common/src/test/java/com/netflix/maestro/models/artifact/RetryArtifactTest.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2024 Netflix, Inc.
+ *
+ * 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 com.netflix.maestro.models.artifact;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.netflix.maestro.MaestroBaseTest;
+import org.junit.Test;
+
+public class RetryArtifactTest extends MaestroBaseTest {
+  @Test
+  public void testRoundTripSerde() throws Exception {
+    RetryArtifact artifact =
+        loadObject("fixtures/artifact/sample-retry-artifact.json", RetryArtifact.class);
+    assertEquals(
+        artifact, MAPPER.readValue(MAPPER.writeValueAsString(artifact), RetryArtifact.class));
+  }
+
+  @Test
+  public void testDeserializeRetryable() throws Exception {
+    Artifact artifact =
+        MAPPER.readValue(
+            """
+            {"type": "RETRY", "retryable": false}
+            """, Artifact.class);
+    assertEquals(Artifact.Type.RETRY, artifact.getType());
+    assertFalse(artifact.asRetry().isRetryable());
+  }
+
+  @Test
+  public void testDeserializeDefaultsToRetryable() throws Exception {
+    Artifact artifact =
+        MAPPER.readValue("""
+            {"type": "RETRY"}
+            """, Artifact.class);
+    assertTrue(artifact.asRetry().isRetryable());
+  }
+}

maestro-common/src/testFixtures/resources/fixtures/artifact/sample-retry-artifact.json

@@ -0,0 +1,4 @@
+{
+  "retryable": false,
+  "type": "RETRY"
+}

maestro-engine/src/main/java/com/netflix/maestro/engine/params/OutputDataManager.java

@@ -91,6 +91,26 @@ public void validateAndMergeOutputParamsAndArtifacts(StepRuntimeSummary runtimeS
     }
   }
 
+  /**
+   * Checks whether the step's output data marks a failed step as non-retryable by the system.
+   *
+   * @param runtimeSummary step runtime summary used to locate the output data
+   * @return true if the output data marks the step non-retryable, false otherwise
+   */
+  public boolean isStepNonRetryable(StepRuntimeSummary runtimeSummary) {
+    Optional<String> externalJobId = extractExternalJobId(runtimeSummary);
+    if (externalJobId.isEmpty()) {
+      return false;
+    }
+    Optional<OutputData> outputDataOpt =
+        outputDataDao.getOutputDataForExternalJob(externalJobId.get(), runtimeSummary.getType());
+    return outputDataOpt
+        .map(OutputData::getArtifacts)
+        .map(artifacts -> artifacts.get(Artifact.Type.RETRY.key()))
+        .map(artifact -> !artifact.asRetry().isRetryable())
+        .orElse(false);
+  }
+
   private Optional<String> extractExternalJobId(StepRuntimeSummary runtimeSummary) {
     Map<String, Artifact> artifacts = runtimeSummary.getArtifacts();
     String jobId = null;

maestro-engine/src/main/java/com/netflix/maestro/engine/tasks/MaestroTask.java

@@ -750,7 +750,7 @@ private void handleTimeoutError(
   }
 
   /** Executes the step instance. It returns true, if the task is in dummy run mode. */
-  @SuppressWarnings("PMD.ExhaustiveSwitchHasDefault")
+  @SuppressWarnings({"PMD.ExhaustiveSwitchHasDefault", "checkstyle:MethodLength"})
   private boolean doExecute(
       Flow flow,
       Task task,
@@ -847,6 +847,21 @@ private boolean doExecute(
             evaluateNextConditionParams(flow, stepDefinition, runtimeSummary);
             doneWithExecute = true;
             break;
+          case USER_FAILED:
+          case PLATFORM_FAILED:
+            // A retryable failure is escalated to FATALLY_FAILED when the step's output data marks
+            // it as non-retryable, then falls through to the FATALLY_FAILED handling.
+            if (!outputDataManager.isStepNonRetryable(runtimeSummary)) {
+              doneWithExecute = true;
+              break;
+            }
+            StepInstance.Status failedStatus = runtimeSummary.getRuntimeState().getStatus();
+            runtimeSummary.markTerminated(StepInstance.Status.FATALLY_FAILED, tracingManager);
+            runtimeSummary.addTimeline(
+                TimelineLogEvent.info(
+                    "Step failed with [%s] and its output data classified it as non-retryable.",
+                    failedStatus));
+            // fall through, to apply failure mode handling
           case FATALLY_FAILED: // Failure mode only applies to FATALLY_FAILED
             if (!runtimeSummary.isIgnoreFailureMode()) {
               if (FailureMode.IGNORE_FAILURE == stepDefinition.getFailureMode()) {
@@ -864,8 +879,6 @@ private boolean doExecute(
             }
             // fall through, otherwise
           case INTERNALLY_FAILED: // Ignoring failure model as the error happens within Maestro
-          case USER_FAILED:
-          case PLATFORM_FAILED:
           case TIMEOUT_FAILED:
           case STOPPED:
           case TIMED_OUT:

maestro-engine/src/test/java/com/netflix/maestro/engine/params/OutputDataManagerTest.java

@@ -31,6 +31,7 @@
 import com.netflix.maestro.models.artifact.Artifact;
 import com.netflix.maestro.models.artifact.DynamicOutputArtifact;
 import com.netflix.maestro.models.artifact.KubernetesArtifact;
+import com.netflix.maestro.models.artifact.RetryArtifact;
 import com.netflix.maestro.models.artifact.TitusArtifact;
 import com.netflix.maestro.models.definition.StepType;
 import com.netflix.maestro.models.parameter.InternalParamMode;
@@ -110,6 +111,40 @@ public void testSaveOutputData() {
     Mockito.verify(outputDataDao, times(1)).insertOrUpdateOutputData(outputData);
   }
 
+  @Test
+  public void testIsStepNonRetryableWhenMarkedNonRetryable() {
+    RetryArtifact retryArtifact = new RetryArtifact();
+    retryArtifact.setRetryable(false);
+    OutputData output = new OutputData(null, Map.of(Artifact.Type.RETRY.key(), retryArtifact));
+    when(outputDataDao.getOutputDataForExternalJob(TASK_ID, StepType.TITUS))
+        .thenReturn(Optional.of(output));
+    runtimeSummary = runtimeSummaryBuilder().type(StepType.TITUS).artifacts(artifacts).build();
+    assertTrue(outputDataManager.isStepNonRetryable(runtimeSummary));
+  }
+
+  @Test
+  public void testIsStepNonRetryableWhenMarkedRetryable() {
+    OutputData output =
+        new OutputData(null, Map.of(Artifact.Type.RETRY.key(), new RetryArtifact()));
+    when(outputDataDao.getOutputDataForExternalJob(TASK_ID, StepType.TITUS))
+        .thenReturn(Optional.of(output));
+    runtimeSummary = runtimeSummaryBuilder().type(StepType.TITUS).artifacts(artifacts).build();
+    assertFalse(outputDataManager.isStepNonRetryable(runtimeSummary));
+  }
+
+  @Test
+  public void testIsStepNonRetryableWhenNoRetryArtifact() {
+    setupOutputDataDao();
+    runtimeSummary = runtimeSummaryBuilder().type(StepType.TITUS).artifacts(artifacts).build();
+    assertFalse(outputDataManager.isStepNonRetryable(runtimeSummary));
+  }
+
+  @Test
+  public void testIsStepNonRetryableWhenNoOutputData() {
+    runtimeSummary = runtimeSummaryBuilder().type(StepType.TITUS).artifacts(artifacts).build();
+    assertFalse(outputDataManager.isStepNonRetryable(runtimeSummary));
+  }
+
   @Test
   public void testMissingJobIdArtifact() {
     outputDataManager.validateAndMergeOutputParamsAndArtifacts(runtimeSummary);

maestro-server/src/test/resources/samples/sample-kubernetes-nonretryable-wf.json

@@ -0,0 +1,40 @@
+{
+  "properties": {
+    "owner": "tester",
+    "run_strategy": "sequential"
+  },
+  "workflow": {
+    "id": "sample-kubernetes-nonretryable-wf",
+    "name": "Test kubernetes workflow that fails and is classified non-retryable via output data",
+    "steps": [
+      {
+        "step": {
+          "id": "job1",
+          "type": "kubernetes",
+          "retry_policy": {
+            "error_retry_limit": 3
+          },
+          "params": {
+            "kubernetes": {
+              "value": {
+                "image": {
+                  "value": "busybox",
+                  "type": "STRING"
+                },
+                "command": {
+                  "value": ["/bin/sh"],
+                  "type": "STRING_ARRAY"
+                },
+                "args": {
+                  "value": ["-c", "sleep 5 && echo $$MAESTRO_OUTPUT_START$1$$MAESTRO_OUTPUT_END && exit 1", "sh", "{\"artifacts\":{\"maestro_retry\":{\"type\":\"RETRY\",\"retryable\":false}}}"],
+                  "type": "STRING_ARRAY"
+                }
+              },
+              "type": "MAP"
+            }
+          }
+        }
+      }
+    ]
+  }
+}