Carry over DefaultEnvVarsFromEnv

go/tasks/plugins/k8s/spark/spark.go

@@ -90,6 +90,11 @@ func (sparkResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsCo
 	for _, envVar := range envVars {
 		sparkEnvVars[envVar.Name] = envVar.Value
 	}
+
+	for k, v := range config.GetK8sPluginConfig().DefaultEnvVarsFromEnv {
+		sparkEnvVars[k] = v
+	}
+
 	sparkEnvVars["FLYTE_MAX_ATTEMPTS"] = strconv.Itoa(int(taskCtx.TaskExecutionMetadata().GetMaxAttempts()))
 
 	serviceAccountName := flytek8s.GetServiceAccountNameFromTaskExecutionMetadata(taskCtx.TaskExecutionMetadata())

go/tasks/plugins/k8s/spark/spark_test.go

@@ -370,6 +370,9 @@ func TestBuildResourceSpark(t *testing.T) {
 	defaultEnvVars := make(map[string]string)
 	defaultEnvVars["foo"] = "bar"
 
+	defaultEnvVarsFromEnv := make(map[string]string)
+	defaultEnvVarsFromEnv["fooEnv"] = "barEnv"
+
 	assert.NoError(t, config.SetK8sPluginConfig(&config.K8sPluginConfig{
 		DefaultPodSecurityContext: &corev1.PodSecurityContext{
 			RunAsUser: &runAsUser,
@@ -416,6 +419,7 @@ func TestBuildResourceSpark(t *testing.T) {
 		SchedulerName:           schedulerName,
 		EnableHostNetworkingPod: &defaultPodHostNetwork,
 		DefaultEnvVars:          defaultEnvVars,
+		DefaultEnvVarsFromEnv:   defaultEnvVarsFromEnv,
 	}),
 	)
 	resource, err := sparkResourceHandler.BuildResource(context.TODO(), dummySparkTaskContext(taskTemplate, true))
@@ -532,6 +536,8 @@ func TestBuildResourceSpark(t *testing.T) {
 	assert.Equal(t, len(sparkApp.Spec.Driver.EnvVars["FLYTE_MAX_ATTEMPTS"]), 1)
 	assert.Equal(t, sparkApp.Spec.Driver.EnvVars["foo"], defaultEnvVars["foo"])
 	assert.Equal(t, sparkApp.Spec.Executor.EnvVars["foo"], defaultEnvVars["foo"])
+	assert.Equal(t, sparkApp.Spec.Driver.EnvVars["fooEnv"], defaultEnvVarsFromEnv["fooEnv"])
+	assert.Equal(t, sparkApp.Spec.Executor.EnvVars["fooEnv"], defaultEnvVarsFromEnv["fooEnv"])
 
 	// Case 2: Driver/Executor request cores set.
 	dummyConfWithRequest := make(map[string]string)