Fix: RestartPolicy serde results in different revision

The change to ommit the `RestartPolicy` when not set results in null not
be included in the json serialization resulting in different revision
number causing Stopped clusters to Update.

We should `ommitempty` for `RestartPolicy` but that should be deffered to `v1beta1`

apis/flinkcluster/v1beta1/flinkcluster_types.go

@@ -556,7 +556,8 @@ type JobSpec struct {
 	// the job will stay in failed state. This option is usually used together
 	// with `autoSavepointSeconds` and `savepointsDir`.
 	// +kubebuilder:default:=Never
-	RestartPolicy *JobRestartPolicy `json:"restartPolicy,omitempty"`
+	// TODO: this field should be omitempty but since it affects job revision we need to defer it to v1beta2
+	RestartPolicy *JobRestartPolicy `json:"restartPolicy"`
 
 	// The action to take after job finishes.
 	// +kubebuilder:default:={afterJobSucceeds:DeleteCluster, afterJobFails:KeepCluster, afterJobCancelled:DeleteCluster}

controllers/flinkcluster/flinkcluster_util_test.go

@@ -71,6 +71,7 @@ func TestNewRevision(t *testing.T) {
 	var jarFile = "gs://my-bucket/myjob.jar"
 	var parallelism int32 = 2
 	var savepointDir = "/savepoint_dir"
+	restartPolicy := v1beta1.JobRestartPolicyNever
 	var flinkCluster = v1beta1.FlinkCluster{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "mycluster",
@@ -107,20 +108,26 @@ func TestNewRevision(t *testing.T) {
 				JarFile:       &jarFile,
 				Parallelism:   &parallelism,
 				SavepointsDir: &savepointDir,
+				RestartPolicy: &restartPolicy,
 			},
 		},
 	}
+
+	var patched = flinkCluster.DeepCopy()
+	// RestartPolicy is not part of the revision.
+	patched.Spec.Job.RestartPolicy = nil
+
 	var collisionCount int32 = 0
 	var controller = true
 	var blockOwnerDeletion = true
-	var raw, _ = getPatch(&flinkCluster)
+	var raw, _ = getPatch(patched)
 	var revision, _ = newRevision(&flinkCluster, 1, &collisionCount)
 	var expectedRevision = appsv1.ControllerRevision{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      "mycluster-fb7687bf5",
+			Name:      "mycluster-7bc87c954f",
 			Namespace: "default",
 			Labels: map[string]string{
-				"flinkoperator.k8s.io/hash":       "fb7687bf5",
+				"flinkoperator.k8s.io/hash":       "7bc87c954f",
 				"flinkoperator.k8s.io/managed-by": "mycluster",
 			},
 			Annotations: map[string]string{},