removed skipFlinkClusterDefault flag and resource defaulting.

spotify · May 25, 2022 · 4667618 · 4667618
1 parent 39bb245
commit 4667618
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 12 deletions.
diff --git a/pkg/flink/config.go b/pkg/flink/config.go
@@ -21,12 +21,11 @@ import (
 
 // Config ... Flink-specific configs
 type Config struct {
-	DefaultFlinkCluster      flinkOp.FlinkCluster `json:"defaultFlinkCluster"`
-	SkipFlinkClusterDefaults bool                 `json:"skipFlinkClusterDefaults"`
-	FlinkPropertiesOverride  map[string]string    `json:"flinkPropertiesOverride" pflag:",Key value pairs of flink properties to be overridden in every FlinkJob"`
-	LogConfig                logs.LogConfig       `json:"logs"`
-	GeneratedNameMaxLength   *int                 `json:"generatedNameMaxLength" pflag:"Specifies the length of TaskExecutionID generated name. default: 50"`
-	RemoteClusterConfig      ClusterConfig        `json:"remoteClusterConfig" pflag:"Configuration of remote K8s cluster for array jobs"`
+	DefaultFlinkCluster     flinkOp.FlinkCluster `json:"defaultFlinkCluster"`
+	FlinkPropertiesOverride map[string]string    `json:"flinkPropertiesOverride" pflag:",Key value pairs of flink properties to be overridden in every FlinkJob"`
+	LogConfig               logs.LogConfig       `json:"logs"`
+	GeneratedNameMaxLength  *int                 `json:"generatedNameMaxLength" pflag:"Specifies the length of TaskExecutionID generated name. default: 50"`
+	RemoteClusterConfig     ClusterConfig        `json:"remoteClusterConfig" pflag:"Configuration of remote K8s cluster for array jobs"`
 }
 
 func GetFlinkConfig() *Config {

diff --git a/pkg/flink/resources.go b/pkg/flink/resources.go
@@ -339,9 +339,6 @@ func NewFlinkCluster(config *Config, taskCtx FlinkTaskContext) (*flinkOp.FlinkCl
 
 	// fill in defaults
 	resource := flinkOp.FlinkCluster(cluster)
-	if !config.SkipFlinkClusterDefaults {
-		resource.Default()
-	}
 
 	return &resource, nil
 }

diff --git a/pkg/flink/resources_test.go b/pkg/flink/resources_test.go
@@ -307,9 +307,7 @@ func TestPartialFlinkCluster(t *testing.T) {
 		},
 	}
 
-	config := &Config{
-		SkipFlinkClusterDefaults: true,
-	}
+	config := &Config{}
 
 	flinkCtx := FlinkTaskContext{
 		ClusterName: ClusterName("generated-name"),