diff --git a/flyteplugins/go/tasks/pluginmachinery/flytek8s/container_helper_test.go b/flyteplugins/go/tasks/pluginmachinery/flytek8s/container_helper_test.go
index 4e609c72b2e..3b7aa88aeb0 100644
--- a/flyteplugins/go/tasks/pluginmachinery/flytek8s/container_helper_test.go
+++ b/flyteplugins/go/tasks/pluginmachinery/flytek8s/container_helper_test.go
@@ -525,7 +525,7 @@ func TestAddFlyteCustomizationsToContainer(t *testing.T) {
 	assert.EqualValues(t, container.Command, []string{"s3://input/path"})
 	assert.Len(t, container.Resources.Limits, 3)
 	assert.Len(t, container.Resources.Requests, 3)
-	assert.Len(t, container.Env, 12)
+	assert.Len(t, container.Env, 13)
 }
 
 func TestAddFlyteCustomizationsToContainer_Resources(t *testing.T) {
diff --git a/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds.go b/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds.go
index 34e13adfa83..b77615120ab 100644
--- a/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds.go
+++ b/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds.go
@@ -60,6 +60,15 @@ func GetExecutionEnvVars(id pluginsCore.TaskExecutionID, consoleURL string) []v1
 			Name:  "FLYTE_INTERNAL_EXECUTION_DOMAIN",
 			Value: nodeExecutionID.Domain,
 		},
+		{
+			// FLYTE_INTERNAL_POD_NAME
+			Name: "_F_PN",
+			ValueFrom: &v1.EnvVarSource{
+				FieldRef: &v1.ObjectFieldSelector{
+					FieldPath: "metadata.name",
+				},
+			},
+		},
 		{
 			Name:  "FLYTE_ATTEMPT_NUMBER",
 			Value: attemptNumber,
diff --git a/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds_test.go b/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds_test.go
index 4015a8d9b8d..fd4828fbbd2 100644
--- a/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds_test.go
+++ b/flyteplugins/go/tasks/pluginmachinery/flytek8s/k8s_resource_adds_test.go
@@ -27,13 +27,13 @@ func TestGetExecutionEnvVars(t *testing.T) {
 	}{
 		{
 			"no-console-url",
-			12,
+			13,
 			"",
 			nil,
 		},
 		{
 			"with-console-url",
-			13,
+			14,
 			"scheme://host/path",
 			&v12.EnvVar{
 				Name:  "FLYTE_EXECUTION_URL",
@@ -42,7 +42,7 @@ func TestGetExecutionEnvVars(t *testing.T) {
 		},
 		{
 			"with-console-url-ending-in-single-slash",
-			13,
+			14,
 			"scheme://host/path/",
 			&v12.EnvVar{
 				Name:  "FLYTE_EXECUTION_URL",
@@ -51,7 +51,7 @@ func TestGetExecutionEnvVars(t *testing.T) {
 		},
 		{
 			"with-console-url-ending-in-multiple-slashes",
-			13,
+			14,
 			"scheme://host/path////",
 			&v12.EnvVar{
 				Name:  "FLYTE_EXECUTION_URL",
@@ -63,7 +63,7 @@ func TestGetExecutionEnvVars(t *testing.T) {
 		envVars := GetExecutionEnvVars(mock, tt.consoleURL)
 		assert.Len(t, envVars, tt.expectedEnvVars)
 		if tt.expectedEnvVar != nil {
-			assert.True(t, proto.Equal(&envVars[4], tt.expectedEnvVar))
+			assert.True(t, proto.Equal(&envVars[5], tt.expectedEnvVar))
 		}
 	}
 }
diff --git a/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go b/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go
index 7ea6c42be2a..d657d4c273c 100644
--- a/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go
+++ b/flyteplugins/go/tasks/plugins/k8s/spark/spark_test.go
@@ -853,7 +853,7 @@ func TestBuildResourcePodTemplate(t *testing.T) {
 	assert.Equal(t, defaultConfig.DefaultEnvVars["foo"], findEnvVarByName(sparkApp.Spec.Driver.Env, "foo").Value)
 	assert.Equal(t, defaultConfig.DefaultEnvVars["fooEnv"], findEnvVarByName(sparkApp.Spec.Driver.Env, "fooEnv").Value)
 	assert.Equal(t, findEnvVarByName(dummyEnvVarsWithSecretRef, "SECRET"), findEnvVarByName(sparkApp.Spec.Driver.Env, "SECRET"))
-	assert.Equal(t, 9, len(sparkApp.Spec.Driver.Env))
+	assert.Equal(t, 10, len(sparkApp.Spec.Driver.Env))
 	assert.Equal(t, testImage, *sparkApp.Spec.Driver.Image)
 	assert.Equal(t, flytek8s.GetServiceAccountNameFromTaskExecutionMetadata(taskCtx.TaskExecutionMetadata()), *sparkApp.Spec.Driver.ServiceAccount)
 	assert.Equal(t, defaultConfig.DefaultPodSecurityContext, sparkApp.Spec.Driver.SecurityContenxt)
@@ -890,7 +890,7 @@ func TestBuildResourcePodTemplate(t *testing.T) {
 	assert.Equal(t, defaultConfig.DefaultEnvVars["foo"], findEnvVarByName(sparkApp.Spec.Executor.Env, "foo").Value)
 	assert.Equal(t, defaultConfig.DefaultEnvVars["fooEnv"], findEnvVarByName(sparkApp.Spec.Executor.Env, "fooEnv").Value)
 	assert.Equal(t, findEnvVarByName(dummyEnvVarsWithSecretRef, "SECRET"), findEnvVarByName(sparkApp.Spec.Executor.Env, "SECRET"))
-	assert.Equal(t, 9, len(sparkApp.Spec.Executor.Env))
+	assert.Equal(t, 10, len(sparkApp.Spec.Executor.Env))
 	assert.Equal(t, testImage, *sparkApp.Spec.Executor.Image)
 	assert.Equal(t, defaultConfig.DefaultPodSecurityContext, sparkApp.Spec.Executor.SecurityContenxt)
 	assert.Equal(t, defaultConfig.DefaultPodDNSConfig, sparkApp.Spec.Executor.DNSConfig)