Merge pull request #169 from rayhaanbhikha/rayhaanb/skip-default-conf…

…ig-flag

Skip FlinkCluster Defaults flag added

pkg/flink/resources.go

@@ -339,12 +339,6 @@ func NewFlinkCluster(config *Config, taskCtx FlinkTaskContext) (*flinkOp.FlinkCl
 
 	// fill in defaults
 	resource := flinkOp.FlinkCluster(cluster)
-	resource.Default()
-
-	err := resource.ValidateCreate()
-	if err != nil {
-		return nil, err
-	}
 
 	return &resource, nil
 }

pkg/flink/resources_test.go

@@ -25,7 +25,9 @@ import (
 	flinkOp "github.com/spotify/flink-on-k8s-operator/apis/flinkcluster/v1beta1"
 	flinkIdl "github.com/spotify/flyte-flink-plugin/gen/pb-go/flyteidl-flink"
 	"gotest.tools/assert"
+
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var artifacts = []string{"gs://bucket/a.jar", "gs://bucket/b.jar", "gs://bucket/c.jar"}
@@ -127,29 +129,6 @@ func TestWithPersistentVolume(t *testing.T) {
 	assert.Assert(t, cluster.Spec.FlinkProperties[flinkIoTmpDirsProperty] == volumeClaimMountPath)
 }
 
-func TestBuildFlinkClusterSpecInvalid(t *testing.T) {
-	job := flinkIdl.FlinkJob{
-		JarFiles: artifacts,
-		FlinkProperties: map[string]string{
-			"taskmanager.numberOfTaskSlots": "1",
-		},
-	}
-
-	// Use empty config
-	config := &Config{}
-
-	flinkCtx := FlinkTaskContext{
-		ClusterName: ClusterName("generated-name"),
-		Namespace:   "test-namespace",
-		Annotations: make(map[string]string),
-		Labels:      make(map[string]string),
-		Job:         job,
-	}
-
-	_, err := NewFlinkCluster(config, flinkCtx)
-	assert.Error(t, err, "image name is unspecified")
-}
-
 func TestBuildFlinkClusterSpecServiceAccount(t *testing.T) {
 	job := flinkIdl.FlinkJob{
 		JarFiles: artifacts,
@@ -293,9 +272,15 @@ func TestBuildFlinkClusterSpecJobCommand(t *testing.T) {
 
 func TestBuildAnnotationPatch(t *testing.T) {
 	patch, err := NewAnnotationPatch("testKey", "testValue")
+	if err != nil {
+		t.Error(err)
+	}
 	assert.Equal(t, patch.Type(), types.MergePatchType)
 
 	bytes, err := patch.Data(nil)
+	if err != nil {
+		t.Error(err)
+	}
 
 	var jsonData map[string]interface{}
 	err = json.Unmarshal(bytes, &jsonData)
@@ -308,3 +293,59 @@ func TestBuildAnnotationPatch(t *testing.T) {
 
 	assert.NilError(t, err)
 }
+
+func TestPartialFlinkCluster(t *testing.T) {
+	parallelism := int32(10)
+	mainClass := "SomeClass"
+	jobIdl := flinkIdl.FlinkJob{
+		JarFiles:    artifacts,
+		MainClass:   mainClass,
+		Parallelism: parallelism,
+		FlinkProperties: map[string]string{
+			"taskmanager.numberOfTaskSlots":            "1",
+			"metrics.reporter.promgateway.groupingKey": `namespace={{.Namespace}};cluster={{.ClusterName}};execution_id={{index .Labels "execution-id"}}`,
+		},
+	}
+
+	config := &Config{}
+
+	flinkCtx := FlinkTaskContext{
+		ClusterName: ClusterName("generated-name"),
+		Namespace:   "test-namespace",
+		Annotations: make(map[string]string),
+		Labels:      map[string]string{"execution-id": "1"},
+		Job:         jobIdl,
+	}
+
+	cluster, err := NewFlinkCluster(config, flinkCtx)
+	if err != nil {
+		t.Error(err)
+	}
+
+	assert.DeepEqual(t, cluster.ObjectMeta, metav1.ObjectMeta{
+		Name:        "generated-name",
+		Namespace:   "test-namespace",
+		Labels:      map[string]string{"execution-id": "1"},
+		Annotations: make(map[string]string),
+	})
+
+	assert.DeepEqual(t, cluster.TypeMeta, metav1.TypeMeta{
+		Kind:       "FlinkCluster",
+		APIVersion: "flinkoperator.k8s.io/v1beta1",
+	})
+
+	assert.DeepEqual(t, cluster.Spec, flinkOp.FlinkClusterSpec{
+		Job: &flinkOp.JobSpec{
+			ClassName:      &mainClass,
+			Parallelism:    &parallelism,
+			InitContainers: []corev1.Container{},
+			PodAnnotations: map[string]string{},
+			PodLabels:      map[string]string{"execution-id": "1"},
+		},
+		EnvVars: []corev1.EnvVar{{Name: "STAGED_JARS", Value: "gs://bucket/a.jar gs://bucket/b.jar gs://bucket/c.jar"}},
+		FlinkProperties: map[string]string{
+			"metrics.reporter.promgateway.groupingKey": "namespace=test-namespace;cluster=generated-name;execution_id=1",
+			"taskmanager.numberOfTaskSlots":            "1",
+		},
+	})
+}