Fix parallelism inference (#34)

spotify · May 13, 2021 · cd796d4 · cd796d4
1 parent 82250c6
commit cd796d4
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 9 deletions.
diff --git a/controllers/flinkcluster_converter.go b/controllers/flinkcluster_converter.go
@@ -944,17 +944,22 @@ func calJobParallelism(cluster *v1beta1.FlinkCluster) *int32 {
 		return cluster.Spec.Job.Parallelism
 	}
 
-	tmReplicas := cluster.Spec.TaskManager.Replicas
-	ts, ok := cluster.Spec.FlinkProperties["taskmanager.numberOfTaskSlots"]
-	if !ok {
-		return nil
+	var value *int
+	if ts, ok := cluster.Spec.FlinkProperties["taskmanager.numberOfTaskSlots"]; ok {
+		parsed, err := strconv.Atoi(ts)
+		if err != nil {
+			return nil
+		}
+		value = &parsed
+	} else {
+		value = calTaskManagerTaskSlots(cluster)
 	}
-	value, err := strconv.Atoi(ts)
-	if err != nil {
+
+	if value == nil {
 		return nil
 	}
 
-	parallelism := tmReplicas * int32(value)
+	parallelism := cluster.Spec.TaskManager.Replicas * int32(*value)
 	return &parallelism
 }
 

diff --git a/controllers/flinkcluster_converter_test.go b/controllers/flinkcluster_converter_test.go
@@ -268,8 +268,7 @@ func TestGetDesiredClusterState(t *testing.T) {
 					},
 					SecurityContext: &securityContext,
 				},
-				FlinkProperties: map[string]string{"taskmanager.numberOfTaskSlots": "1"},
-				EnvVars:         []corev1.EnvVar{{Name: "FOO", Value: "abc"}},
+				EnvVars: []corev1.EnvVar{{Name: "FOO", Value: "abc"}},
 				EnvFrom: []corev1.EnvFromSource{{ConfigMapRef: &corev1.ConfigMapEnvSource{
 					LocalObjectReference: corev1.LocalObjectReference{
 						Name: "FOOMAP",