diff --git a/go/tasks/pluginmachinery/flytek8s/pod_helper.go b/go/tasks/pluginmachinery/flytek8s/pod_helper.go
index 1e771e9bb5..a0925de067 100755
--- a/go/tasks/pluginmachinery/flytek8s/pod_helper.go
+++ b/go/tasks/pluginmachinery/flytek8s/pod_helper.go
@@ -253,7 +253,7 @@ func DemystifyPending(status v1.PodStatus) (pluginsCore.PhaseInfo, error) {
 									&pluginsCore.TaskInfo{OccurredAt: &t},
 								), nil
 
-							case "CreateContainerConfigError":
+							case "CreateContainerConfigError", "InvalidImageName":
 								t := c.LastTransitionTime.Time
 								return pluginsCore.PhaseInfoFailure(finalReason, finalMessage, &pluginsCore.TaskInfo{
 									OccurredAt: &t,
diff --git a/go/tasks/pluginmachinery/flytek8s/pod_helper_test.go b/go/tasks/pluginmachinery/flytek8s/pod_helper_test.go
index 4cb68cd47e..2f516ab211 100755
--- a/go/tasks/pluginmachinery/flytek8s/pod_helper_test.go
+++ b/go/tasks/pluginmachinery/flytek8s/pod_helper_test.go
@@ -613,7 +613,7 @@ func TestDemystifyPending(t *testing.T) {
 		}
 		taskStatus, err := DemystifyPending(s)
 		assert.NoError(t, err)
-		assert.Equal(t, pluginsCore.PhaseRetryableFailure, taskStatus.Phase())
+		assert.Equal(t, pluginsCore.PhasePermanentFailure, taskStatus.Phase())
 	})
 
 	t.Run("RegistryUnavailable", func(t *testing.T) {
diff --git a/go/tasks/plugins/k8s/container/container_test.go b/go/tasks/plugins/k8s/container/container_test.go
index df32ccdc3c..20a01d8a6c 100755
--- a/go/tasks/plugins/k8s/container/container_test.go
+++ b/go/tasks/plugins/k8s/container/container_test.go
@@ -2,6 +2,7 @@ package container
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -197,3 +198,45 @@ func TestContainerTaskExecutor_GetProperties(t *testing.T) {
 	expected := k8s.PluginProperties{}
 	assert.Equal(t, expected, plugin.GetProperties())
 }
+
+func TestContainerTaskExecutor_GetTaskStatus_InvalidImageName(t *testing.T) {
+	ctx := context.TODO()
+	c := Plugin{}
+	reason := "InvalidImageName"
+	message := "Failed to apply default image tag \"TEST/flyteorg/myapp:latest\": couldn't parse image reference" +
+		" \"TEST/flyteorg/myapp:latest\": invalid reference format: repository name must be lowercase"
+
+	pendingPod := &v1.Pod{
+		Status: v1.PodStatus{
+			Phase: v1.PodPending,
+			Conditions: []v1.PodCondition{
+				{
+					Type:   v1.PodReady,
+					Status: v1.ConditionFalse,
+				},
+			},
+			ContainerStatuses: []v1.ContainerStatus{
+				{
+					ContainerID: "ContainerID",
+					Ready:       false,
+					State: v1.ContainerState{
+						Waiting: &v1.ContainerStateWaiting{
+							Reason:  reason,
+							Message: message,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	t.Run("failInvalidImageName", func(t *testing.T) {
+		pendingPod.Status.Phase = v1.PodPending
+		phaseInfo, err := c.GetTaskPhase(ctx, nil, pendingPod)
+		finalReason := fmt.Sprintf("|%s", reason)
+		finalMessage := fmt.Sprintf("|%s", message)
+		assert.NoError(t, err)
+		assert.Equal(t, pluginsCore.PhasePermanentFailure, phaseInfo.Phase())
+		assert.Equal(t, &core.ExecutionError{Code: finalReason, Message: finalMessage, Kind: core.ExecutionError_USER}, phaseInfo.Err())
+	})
+}