Expose job backoffLimit to users

apis/flinkcluster/v1beta1/assets/test/flinkcluster_type_expected.yaml

@@ -47,6 +47,7 @@ spec:
       afterJobFails: KeepCluster
       afterJobCancelled: DeleteCluster
     mode: Detached
+    backoffLimit: 0
   flinkProperties:
     taskmanager.numberOfTaskSlots: "1"
   recreateOnUpdate: true

apis/flinkcluster/v1beta1/flinkcluster_types.go

@@ -632,6 +632,13 @@ type JobSpec struct {
 	// +kubebuilder:validation:Enum=Never;FromSavepointOnFailure
 	RestartPolicy *JobRestartPolicy `json:"restartPolicy,omitempty"`
 
+	// BackoffLimit limits the number of job pod restarts. This can happen when when the pod is kicked out of a node
+	// or when the RestartPolicy is set to Never and a container errors out
+	// default: `0`,
+	// 0 means no retry.
+	// +kubebuilder:default:=0
+	BackoffLimit *int32 `json:"backoffLimit,omitempty"`
+
 	// The action to take after job finishes.
 	// +kubebuilder:default:={afterJobSucceeds:DeleteCluster, afterJobFails:KeepCluster, afterJobCancelled:DeleteCluster}
 	CleanupPolicy *CleanupPolicy `json:"cleanupPolicy,omitempty"`

config/crd/bases/flinkoperator.k8s.io_flinkclusters.yaml

@@ -587,6 +587,10 @@ spec:
                     autoSavepointSeconds:
                       format: int32
                       type: integer
+                    backoffLimit:
+                      default: 0
+                      format: int32
+                      type: integer
                     cancelRequested:
                       type: boolean
                     className:

config/webhook/manifests.yaml

@@ -1,4 +1,3 @@
-
 ---
 apiVersion: admissionregistration.k8s.io/v1
 kind: MutatingWebhookConfiguration
@@ -26,7 +25,6 @@ webhooks:
     resources:
     - flinkclusters
   sideEffects: None
-
 ---
 apiVersion: admissionregistration.k8s.io/v1
 kind: ValidatingWebhookConfiguration

controllers/flinkcluster/flinkcluster_converter.go

@@ -61,7 +61,6 @@ const (
 )
 
 var (
-	backoffLimit                  int32 = 0
 	terminationGracePeriodSeconds int64 = 60
 	flinkSysProps                       = map[string]struct{}{
 		"jobmanager.rpc.address": {},
@@ -927,7 +926,7 @@ func newJob(flinkCluster *v1beta1.FlinkCluster) *batchv1.Job {
 				},
 				Spec: *podSpec,
 			},
-			BackoffLimit: &backoffLimit,
+			BackoffLimit: jobSpec.BackoffLimit,
 		},
 	}
 }

controllers/flinkcluster/flinkcluster_converter_test.go

@@ -57,7 +57,7 @@ var (
 	memoryOffHeapMin                   = resource.MustParse("600M")
 	memoryProcessRatio int32           = 80
 	jobMode            v1beta1.JobMode = v1beta1.JobModeDetached
-	jobBackoffLimit    int32           = 0
+	jobBackoffLimit    int32           = 10
 	ingressPathType                    = networkingv1.PathTypePrefix
 	storageClassName                   = "default-class"
 	jmReadinessProbe                   = corev1.Probe{
@@ -192,6 +192,7 @@ func getDummyFlinkCluster() *v1beta1.FlinkCluster {
 				},
 				SecurityContext: &securityContext,
 				HostAliases:     hostAliases,
+				BackoffLimit:    &jobBackoffLimit,
 			},
 			JobManager: &v1beta1.JobManagerSpec{
 				AccessScope: v1beta1.AccessScopeVPC,
@@ -1634,4 +1635,6 @@ func TestClassPath(t *testing.T) {
 	args := desired.Job.Spec.Template.Spec.Containers[0].Args
 
 	assert.DeepEqual(t, args, expectedArgs)
+
+	assert.Equal(t, observed.cluster.Spec.Job.BackoffLimit, desired.Job.Spec.BackoffLimit)
 }