Repository: kestra-io/kestra
Issue: Provide a task to fail a flow

### Feature description

Provide a `Fail` task that will fail the flow unconditionally.

This can be convenient for switch to fail on un-wanted values.

We can also provide a condition for fine-grain failure.

Issue comments:
Task example:

```yaml
id: fail
type: io.kestra.core.taks.flow.Fail
#optional condition that is at true by default so fail un-conditionnaly
condition: {{outputs.previous.value == "you must fail"}}
```

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/tasks/executions/Fail.java b/core/src/main/java/io/kestra/core/tasks/executions/Fail.java
new file mode 100644
index 0000000000..d9182d3b5b
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/executions/Fail.java
@@ -0,0 +1,110 @@
+package io.kestra.core.tasks.executions;
+
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.tasks.RunnableTask;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.tasks.VoidOutput;
+import io.kestra.core.runners.RunContext;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "Fail the execution",
+    description = "Used to fail the execution for example on a switch branch or on some conditions based on the execution context."
+)
+@Plugin(
+    examples = {
+        @Example(
+            title = "Fail on a switch branch",
+            full = true,
+            code = {
+                "id: fail-on-switch\n" +
+                "namespace: io.kestra.tests\n" +
+                "\n" +
+                "inputs:\n" +
+                "  - name: param\n" +
+                "    type: STRING\n" +
+                "    required: true\n" +
+                "\n" +
+                "tasks:\n" +
+                "  - id: switch\n" +
+                "    type: io.kestra.core.tasks.flows.Switch\n" +
+                "    value: \"{{inputs.param}}\"\n" +
+                "    cases:\n" +
+                "      case1:\n" +
+                "        - id: case1\n" +
+                "          type: io.kestra.core.tasks.debugs.Echo\n" +
+                "          format: Case 1\n" +
+                "      case2:\n" +
+                "        - id: case2\n" +
+                "          type: io.kestra.core.tasks.debugs.Echo\n" +
+                "          format: Case 2\n" +
+                "      notexist:\n" +
+                "        - id: fail\n" +
+                "          type: io.kestra.core.tasks.executions.Fail",
+            }
+        ),
+        @Example(
+            title = "Fail on a condition",
+            full = true,
+            code = {
+                "id: fail-on-condition\n" +
+                "namespace: io.kestra.tests\n" +
+                "\n" +
+                "inputs:\n" +
+                "  - name: param\n" +
+                "    type: STRING\n" +
+                "    required: true\n" +
+                "\n" +
+                "tasks:\n" +
+                "  - id: before\n" +
+                "    type: io.kestra.core.tasks.debugs.Echo\n" +
+                "    format: I'm before the fail on condition \n" +
+                "  - id: fail\n" +
+                "    type: io.kestra.core.tasks.executions.Fail\n" +
+                "    condition: '{{inputs.param == \"fail\"}}'\n" +
+                "  - id: after\n" +
+                "    type: io.kestra.core.tasks.debugs.Echo\n" +
+                "    format: I'm after the fail on condition "
+            }
+        )
+    }
+)
+public class Fail extends Task implements RunnableTask<VoidOutput> {
+
+    @PluginProperty(dynamic = true)
+    @Schema(title = "Optional condition, must evaluate to a boolean.")
+    private String condition;
+
+    @PluginProperty(dynamic = true)
+    @Schema(title = "Optional error message.")
+    @Builder.Default
+    private String errorMessage = "Task failure";
+
+    @Override
+    public VoidOutput run(RunContext runContext) throws Exception {
+        if (condition != null) {
+            String rendered = runContext.render(condition);
+            if (Boolean.parseBoolean(rendered)) {
+                runContext.logger().error(runContext.render(errorMessage));
+                throw new RuntimeException("Fail on a condition");
+            }
+            return null;
+        }
+
+        runContext.logger().error(runContext.render(errorMessage));
+        throw new RuntimeException("Fail always");
+    }
+}
diff --git a/core/src/main/resources/icons/io.kestra.core.tasks.executions.Fail.svg b/core/src/main/resources/icons/io.kestra.core.tasks.executions.Fail.svg
new file mode 100644
index 0000000000..f4789f1b62
--- /dev/null
+++ b/core/src/main/resources/icons/io.kestra.core.tasks.executions.Fail.svg
@@ -0,0 +1,10 @@
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" version="1.1" width="256" height="256" viewBox="0 0 256 256" xml:space="preserve">
+
+<defs>
+</defs>
+<g style="stroke: none; stroke-width: 0; stroke-dasharray: none; stroke-linecap: butt; stroke-linejoin: miter; stroke-miterlimit: 10; fill: none; fill-rule: nonzero; opacity: 1;" transform="translate(1.4065934065934016 1.4065934065934016) scale(2.81 2.81)" >
+	<path d="M 45 90 C 20.187 90 0 69.813 0 45 C 0 20.187 20.187 0 45 0 c 24.813 0 45 20.187 45 45 C 90 69.813 69.813 90 45 90 z" style="stroke: none; stroke-width: 1; stroke-dasharray: none; stroke-linecap: butt; stroke-linejoin: miter; stroke-miterlimit: 10; fill: rgb(236,0,0); fill-rule: nonzero; opacity: 1;" transform=" matrix(1 0 0 1 0 0) " stroke-linecap="round" />
+	<path d="M 28.5 65.5 c -1.024 0 -2.047 -0.391 -2.829 -1.172 c -1.562 -1.562 -1.562 -4.095 0 -5.656 l 33 -33 c 1.561 -1.562 4.096 -1.562 5.656 0 c 1.563 1.563 1.563 4.095 0 5.657 l -33 33 C 30.547 65.109 29.524 65.5 28.5 65.5 z" style="stroke: none; stroke-width: 1; stroke-dasharray: none; stroke-linecap: butt; stroke-linejoin: miter; stroke-miterlimit: 10; fill: rgb(255,255,255); fill-rule: nonzero; opacity: 1;" transform=" matrix(1 0 0 1 0 0) " stroke-linecap="round" />
+	<path d="M 61.5 65.5 c -1.023 0 -2.048 -0.391 -2.828 -1.172 l -33 -33 c -1.562 -1.563 -1.562 -4.095 0 -5.657 c 1.563 -1.562 4.095 -1.562 5.657 0 l 33 33 c 1.563 1.562 1.563 4.095 0 5.656 C 63.548 65.109 62.523 65.5 61.5 65.5 z" style="stroke: none; stroke-width: 1; stroke-dasharray: none; stroke-linecap: butt; stroke-linejoin: miter; stroke-miterlimit: 10; fill: rgb(255,255,255); fill-rule: nonzero; opacity: 1;" transform=" matrix(1 0 0 1 0 0) " stroke-linecap="round" />
+</g>
+</svg>
\ No newline at end of file


Test patch:
diff --git a/core/src/test/java/io/kestra/core/Helpers.java b/core/src/test/java/io/kestra/core/Helpers.java
index 9cc950bc13..fc6e06bb4e 100644
--- a/core/src/test/java/io/kestra/core/Helpers.java
+++ b/core/src/test/java/io/kestra/core/Helpers.java
@@ -19,7 +19,7 @@
 import java.util.function.Consumer;
 
 public class Helpers {
-    public static long FLOWS_COUNT = 56;
+    public static long FLOWS_COUNT = 58;
 
     public static ApplicationContext applicationContext() throws URISyntaxException {
         return applicationContext(
diff --git a/core/src/test/java/io/kestra/core/tasks/executions/FailTest.java b/core/src/test/java/io/kestra/core/tasks/executions/FailTest.java
new file mode 100644
index 0000000000..67a8ecaa99
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/executions/FailTest.java
@@ -0,0 +1,46 @@
+package io.kestra.core.tasks.executions;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+public class FailTest extends AbstractMemoryRunnerTest {
+    @Test
+    void failOnSwitch() throws TimeoutException {
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "fail-on-switch", null,
+            (f, e) -> Map.of("param", "fail") , Duration.ofSeconds(120));
+
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.findTaskRunsByTaskId("switch").get(0).getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+    }
+
+    @Test
+    void failOnCondition() throws TimeoutException {
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "fail-on-condition", null,
+            (f, e) -> Map.of("param", "fail") , Duration.ofSeconds(120));
+
+        assertThat(execution.getTaskRunList(), hasSize(2));
+        assertThat(execution.findTaskRunsByTaskId("fail").get(0).getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+    }
+
+    @Test
+    void dontFailOnCondition() throws TimeoutException {
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "fail-on-condition", null,
+            (f, e) -> Map.of("param", "success") , Duration.ofSeconds(120));
+
+        assertThat(execution.getTaskRunList(), hasSize(3));
+        assertThat(execution.findTaskRunsByTaskId("fail").get(0).getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+}
diff --git a/core/src/test/resources/flows/valids/fail-on-condition.yaml b/core/src/test/resources/flows/valids/fail-on-condition.yaml
new file mode 100644
index 0000000000..62c729199d
--- /dev/null
+++ b/core/src/test/resources/flows/valids/fail-on-condition.yaml
@@ -0,0 +1,18 @@
+id: fail-on-condition
+namespace: io.kestra.tests
+
+inputs:
+  - name: param
+    type: STRING
+    required: true
+
+tasks:
+  - id: before
+    type: io.kestra.core.tasks.debugs.Echo
+    format: I'm before the fail on condition
+  - id: fail
+    type: io.kestra.core.tasks.executions.Fail
+    condition: '{{inputs.param == "fail"}}'
+  - id: after
+    type: io.kestra.core.tasks.debugs.Echo
+    format: I'm after the fail on condition
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/fail-on-switch.yaml b/core/src/test/resources/flows/valids/fail-on-switch.yaml
new file mode 100644
index 0000000000..71aaf2d4a6
--- /dev/null
+++ b/core/src/test/resources/flows/valids/fail-on-switch.yaml
@@ -0,0 +1,24 @@
+id: fail-on-switch
+namespace: io.kestra.tests
+
+inputs:
+  - name: param
+    type: STRING
+    required: true
+
+tasks:
+  - id: switch
+    type: io.kestra.core.tasks.flows.Switch
+    value: "{{inputs.param}}"
+    cases:
+      case1:
+        - id: case1
+          type: io.kestra.core.tasks.debugs.Echo
+          format: Case 1
+      case2:
+        - id: case2
+          type: io.kestra.core.tasks.debugs.Echo
+          format: Case 2
+      notexist:
+        - id: fail
+          type: io.kestra.core.tasks.executions.Fail
\ No newline at end of file