Fix: check if BatchScheduler is set (#175)

spotify · Dec 14, 2021 · 520f9e9 · 520f9e9
1 parent c23950a
commit 520f9e9
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 9 deletions.
diff --git a/api/v1beta1/flinkcluster_default.go b/api/v1beta1/flinkcluster_default.go
@@ -40,6 +40,13 @@ func _SetDefault(cluster *FlinkCluster) {
 		cluster.Spec.RecreateOnUpdate = new(bool)
 		*cluster.Spec.RecreateOnUpdate = true
 	}
+
+	if cluster.Spec.BatchSchedulerName != nil {
+		cluster.Spec.BatchScheduler = &BatchSchedulerSpec{
+			Name: *cluster.Spec.BatchSchedulerName,
+		}
+	}
+
 	_SetImageDefault(&cluster.Spec.Image)
 	flinkVersion, _ := version.NewVersion(cluster.Spec.FlinkVersion)
 	_SetJobManagerDefault(&cluster.Spec.JobManager, flinkVersion)

diff --git a/controllers/flinkcluster_reconciler.go b/controllers/flinkcluster_reconciler.go
@@ -117,18 +117,12 @@ func (reconciler *ClusterReconciler) reconcile() (ctrl.Result, error) {
 }
 
 func (reconciler *ClusterReconciler) reconcileBatchScheduler() error {
-	var batchSchedulerName string
-	if reconciler.observed.cluster.Spec.BatchSchedulerName != nil {
-		batchSchedulerName = *reconciler.observed.cluster.Spec.BatchSchedulerName
-	}
-	if reconciler.observed.cluster.Spec.BatchScheduler != nil {
-		batchSchedulerName = reconciler.observed.cluster.Spec.BatchScheduler.Name
-	}
-	if batchSchedulerName == "" {
+	if reconciler.observed.cluster.Spec.BatchScheduler == nil ||
+		reconciler.observed.cluster.Spec.BatchScheduler.Name == "" {
 		return nil
 	}
 
-	scheduler, err := batchscheduler.GetScheduler(batchSchedulerName)
+	scheduler, err := batchscheduler.GetScheduler(reconciler.observed.cluster.Spec.BatchScheduler.Name)
 	if err != nil {
 		return err
 	}