Bash Task : add a property warningOnStdErr
core/src/main/java/io/kestra/core/models/tasks/Output.java core/src/main/java/io/kestra/core/runners/Worker.java core/src/main/java/io/kestra/core/runners/WorkerTask.java core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java core/src/main/java/io/kestra/core/tasks/scripts/ScriptOutput.java
core/src/main/java/io/kestra/core/models/tasks/Output.java core/src/main/java/io/kestra/core/runners/Worker.java core/src/main/java/io/kestra/core/runners/WorkerTask.java core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java core/src/main/java/io/kestra/core/tasks/scripts/ScriptOutput.java
core/src/test/java/io/kestra/core/Helpers.java core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java core/src/test/java/io/kestra/core/tasks/BashTest.java core/src/test/resources/flows/valids/bash-warning.yaml webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
diff --git a/core/src/main/java/io/kestra/core/models/tasks/Output.java b/core/src/main/java/io/kestra/core/models/tasks/Output.java
index 62c3e3e5f7..24ca7e50f0 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/Output.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/Output.java
@@ -1,10 +1,16 @@
 package io.kestra.core.models.tasks;
 
+import io.kestra.core.models.flows.State;
 import io.kestra.core.serializers.JacksonMapper;
 
 import java.util.Map;
+import java.util.Optional;
 
 public interface Output {
+    default Optional<State.Type> finalState() {
+        return Optional.empty();
+    }
+
     default Map<String, Object> toMap() {
         return JacksonMapper.toMap(this);
     }
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 71d450f48e..7925c5f22b 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -442,6 +442,9 @@ public void run() {
                 }
 
                 taskState = io.kestra.core.models.flows.State.Type.SUCCESS;
+                if (taskOutput != null && taskOutput.finalState().isPresent()) {
+                    taskState = taskOutput.finalState().get();
+                }
             } catch (net.jodah.failsafe.TimeoutExceededException e) {
                 this.exceptionHandler(this, new TimeoutExceededException(workerTask.getTask().getTimeout(), e));
             } catch (Exception e) {
diff --git a/core/src/main/java/io/kestra/core/runners/WorkerTask.java b/core/src/main/java/io/kestra/core/runners/WorkerTask.java
index 775bad7ca1..1a664c32ba 100644
--- a/core/src/main/java/io/kestra/core/runners/WorkerTask.java
+++ b/core/src/main/java/io/kestra/core/runners/WorkerTask.java
@@ -1,11 +1,10 @@
 package io.kestra.core.runners;
 
-import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.tasks.Task;
 import lombok.Builder;
 import lombok.Data;
 import lombok.With;
-import io.kestra.core.models.executions.TaskRun;
-import io.kestra.core.models.tasks.Task;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -24,9 +23,6 @@ public class WorkerTask {
     @NotNull
     private RunContext runContext;
 
-    @NotNull
-    private Execution execution;
-
     public Logger logger() {
         return LoggerFactory.getLogger(
             "flow." + this.getTaskRun().getFlowId() + "." +
diff --git a/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java b/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java
index f5d1c0fd10..5c45c84568 100644
--- a/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java
+++ b/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java
@@ -124,6 +124,14 @@ abstract public class AbstractBash extends Task {
     )
     protected Map<String, String> env;
 
+    @Builder.Default
+    @Schema(
+        title = "Use `WARNING` state if any stdErr is sent"
+    )
+    @PluginProperty(dynamic = false)
+    @NotNull
+    protected Boolean warningOnStdErr = true;
+
     @Getter(AccessLevel.NONE)
     protected transient Path workingDirectory;
 
@@ -215,6 +223,7 @@ protected ScriptOutput run(RunContext runContext, Supplier<String> supplier) thr
             .exitCode(runResult.getExitCode())
             .stdOutLineCount(runResult.getStdOut().getLogsCount())
             .stdErrLineCount(runResult.getStdErr().getLogsCount())
+            .warningOnStdErr(this.warningOnStdErr)
             .vars(outputsVars)
             .files(uploaded)
             .outputFiles(uploaded)
diff --git a/core/src/main/java/io/kestra/core/tasks/scripts/ScriptOutput.java b/core/src/main/java/io/kestra/core/tasks/scripts/ScriptOutput.java
index 2ee10d3f37..34bdf18c66 100644
--- a/core/src/main/java/io/kestra/core/tasks/scripts/ScriptOutput.java
+++ b/core/src/main/java/io/kestra/core/tasks/scripts/ScriptOutput.java
@@ -1,12 +1,16 @@
 package io.kestra.core.tasks.scripts;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.models.tasks.Output;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Builder;
 import lombok.Getter;
 
 import java.net.URI;
 import java.util.Map;
+import java.util.Optional;
 import javax.validation.constraints.NotNull;
 
 @Builder
@@ -47,4 +51,12 @@ public class ScriptOutput implements io.kestra.core.models.tasks.Output {
     )
     @PluginProperty(additionalProperties = URI.class)
     private final Map<String, URI> outputFiles;
+
+    @JsonIgnore
+    private Boolean warningOnStdErr;
+
+    @Override
+    public Optional<State.Type> finalState() {
+        return this.warningOnStdErr != null && this.stdErrLineCount > 0 ? Optional.of(State.Type.WARNING) : Output.super.finalState();
+    }
 }

diff --git a/core/src/test/java/io/kestra/core/Helpers.java b/core/src/test/java/io/kestra/core/Helpers.java
index bdeb79c4a5..ae3d6c475b 100644
--- a/core/src/test/java/io/kestra/core/Helpers.java
+++ b/core/src/test/java/io/kestra/core/Helpers.java
@@ -20,7 +20,7 @@
 import java.util.function.Consumer;
 
 public class Helpers {
-    public static long FLOWS_COUNT = 47;
+    public static long FLOWS_COUNT = 48;
 
     public static ApplicationContext applicationContext() throws URISyntaxException {
         return applicationContext(
diff --git a/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java b/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
index 4ef7ed8565..bd67d7c936 100644
--- a/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
+++ b/core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java
@@ -53,7 +53,7 @@ void bash() {
         Map<String, Object> generate = jsonSchemaGenerator.properties(Task.class, Bash.class);
 
         assertThat(generate, is(not(nullValue())));
-        assertThat(((Map<String, Map<String, Object>>) generate.get("properties")).size(), is(11));
+        assertThat(((Map<String, Map<String, Object>>) generate.get("properties")).size(), is(12));
 
         generate = jsonSchemaGenerator.outputs(Task.class, Bash.class);
 
diff --git a/core/src/test/java/io/kestra/core/tasks/BashTest.java b/core/src/test/java/io/kestra/core/tasks/BashTest.java
new file mode 100644
index 0000000000..a8cf8d345e
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/BashTest.java
@@ -0,0 +1,24 @@
+package io.kestra.core.tasks;
+
+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.util.concurrent.TimeoutException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+class BashTest extends AbstractMemoryRunnerTest {
+    @Test
+    void error() throws TimeoutException {
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "bash-warning");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.WARNING));
+        assertThat(execution.getTaskRunList().get(0).getState().getCurrent(), is(State.Type.WARNING));
+        assertThat(execution.getTaskRunList().get(0).getOutputs().get("stdErrLineCount"), is(1));
+        assertThat(execution.getTaskRunList(), hasSize(1));
+    }
+}
diff --git a/core/src/test/resources/flows/valids/bash-warning.yaml b/core/src/test/resources/flows/valids/bash-warning.yaml
new file mode 100644
index 0000000000..ae9b872618
--- /dev/null
+++ b/core/src/test/resources/flows/valids/bash-warning.yaml
@@ -0,0 +1,13 @@
+id: bash-warning
+namespace: io.kestra.tests
+
+tasks:
+  - id: failed
+    type: io.kestra.core.tasks.scripts.Bash
+    commands:
+      - 'echo "error" >> /dev/stderr'
+
+errors:
+  - id: error
+    type: io.kestra.core.tasks.debugs.Echo
+    format: second {{task.id}}
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
index 25bceeecc9..910bf93303 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
@@ -75,7 +75,7 @@ void bash() throws URISyntaxException {
             assertThat(doc.getMarkdown(), containsString("io.kestra.core.tasks.scripts.Bash"));
             assertThat(doc.getMarkdown(), containsString("Exit if any non true return value"));
             assertThat(doc.getMarkdown(), containsString("The standard output line count"));
-            assertThat(((Map<String, Object>) doc.getSchema().getProperties().get("properties")).size(), is(11));
+            assertThat(((Map<String, Object>) doc.getSchema().getProperties().get("properties")).size(), is(12));
             assertThat(((Map<String, Object>) doc.getSchema().getOutputs().get("properties")).size(), is(6));
         });
     }
