diff --git a/apis/flinkcluster/v1beta1/flinkcluster_types.go b/apis/flinkcluster/v1beta1/flinkcluster_types.go
index 2109eb76..1d922576 100644
--- a/apis/flinkcluster/v1beta1/flinkcluster_types.go
+++ b/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}
diff --git a/config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml b/config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml
index 7c0ff875..437000cc 100644
--- a/config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml
+++ b/config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml
@@ -1679,6 +1679,8 @@ spec:
                           - name
                         type: object
                       type: array
+                  required:
+                    - restartPolicy
                   type: object
                 jobManager:
                   default:
diff --git a/controllers/flinkcluster/flinkcluster_util_test.go b/controllers/flinkcluster/flinkcluster_util_test.go
index 8ae363fa..902bbd2e 100644
--- a/controllers/flinkcluster/flinkcluster_util_test.go
+++ b/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{},