plumb feature gate value through job validation opts and modify validateCompletions function to only check completions == parallelism after the update, not before

pkg/apis/batch/validation/validation.go

@@ -25,17 +25,16 @@ import (
 	"github.com/robfig/cron/v3"
 
 	v1 "k8s.io/api/core/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	unversionedvalidation "k8s.io/apimachinery/pkg/apis/meta/v1/validation"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
 	apimachineryvalidation "k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
-	"k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/apis/batch"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	apivalidation "k8s.io/kubernetes/pkg/apis/core/validation"
-	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/utils/pointer"
 )
 
@@ -377,7 +376,7 @@ func ValidateJobUpdateStatus(job, oldJob *batch.Job) field.ErrorList {
 func ValidateJobSpecUpdate(spec, oldSpec batch.JobSpec, fldPath *field.Path, opts JobValidationOptions) field.ErrorList {
 	allErrs := field.ErrorList{}
 	allErrs = append(allErrs, ValidateJobSpec(&spec, fldPath, opts.PodValidationOptions)...)
-	allErrs = append(allErrs, validateCompletions(spec, oldSpec, fldPath.Child("completions"))...)
+	allErrs = append(allErrs, validateCompletions(spec, oldSpec, fldPath.Child("completions"), opts)...)
 	allErrs = append(allErrs, apivalidation.ValidateImmutableField(spec.Selector, oldSpec.Selector, fldPath.Child("selector"))...)
 	allErrs = append(allErrs, validatePodTemplateUpdate(spec, oldSpec, fldPath, opts)...)
 	allErrs = append(allErrs, apivalidation.ValidateImmutableField(spec.CompletionMode, oldSpec.CompletionMode, fldPath.Child("completionMode"))...)
@@ -567,31 +566,31 @@ func ValidateJobTemplateSpec(spec *batch.JobTemplateSpec, fldPath *field.Path, o
 	return allErrs
 }
 
-func validateCompletions(spec, oldSpec batch.JobSpec, fldPath *field.Path) field.ErrorList {
+func validateCompletions(spec, oldSpec batch.JobSpec, fldPath *field.Path, opts JobValidationOptions) field.ErrorList {
-	if !feature.DefaultFeatureGate.Enabled(features.ElasticIndexedJob) {
+	if !opts.AllowElasticIndexedJobs {
 		return apivalidation.ValidateImmutableField(spec.Completions, oldSpec.Completions, fldPath)
 	}
+
 	// Completions is immutable for non-indexed jobs.
 	// For Indexed Jobs, if ElasticIndexedJob feature gate is not enabled,
 	// fall back to validating that spec.Completions is always immutable.
-	var allErrs field.ErrorList
 	isIndexedJob := spec.CompletionMode != nil && *spec.CompletionMode == batch.IndexedCompletion
 	if !isIndexedJob {
 		return apivalidation.ValidateImmutableField(spec.Completions, oldSpec.Completions, fldPath)
 	}
 
+	var allErrs field.ErrorList
+	if apiequality.Semantic.DeepEqual(spec.Completions, oldSpec.Completions) {
+		return allErrs
+	}
 	// Indexed Jobs cannot set completions to nil. The nil check
 	// is already performed in validateJobSpec, no need to add another error.
-	if spec.Completions == nil || oldSpec.Completions == nil {
+	if spec.Completions == nil {
 		return allErrs
 	}
 
-	// For Indexed Jobs, spec.Completions can only be mutated in tandem
+	if *spec.Completions != *spec.Parallelism {
-	// with spec.Parallelism such that spec.Completions == spec.Parallelism
+		allErrs = append(allErrs, field.Invalid(fldPath, spec.Completions, fmt.Sprintf("can only be modified in tandem with %s", fldPath.Root().Child("parallelism").String())))
-	if *spec.Completions != *oldSpec.Completions {
-		if *spec.Completions != *spec.Parallelism || *oldSpec.Completions != *oldSpec.Parallelism {
-			allErrs = append(allErrs, field.Invalid(fldPath, spec.Completions, fmt.Sprintf("can only be modified in tandem with %s", fldPath.Root().Child("parallelism").String())))
-		}
 	}
 	return allErrs
 }
@@ -602,4 +601,6 @@ type JobValidationOptions struct {
 	AllowTrackingAnnotation bool
 	// Allow mutable node affinity, selector and tolerations of the template
 	AllowMutableSchedulingDirectives bool
+	// Allow elastic indexed jobs
+	AllowElasticIndexedJobs bool
 }

pkg/apis/batch/validation/validation_test.go

@@ -1298,6 +1298,9 @@ func TestValidateJobUpdate(t *testing.T) {
 				job.Spec.Completions = pointer.Int32Ptr(2)
 				job.Spec.Parallelism = pointer.Int32Ptr(2)
 			},
+			opts: JobValidationOptions{
+				AllowElasticIndexedJobs: true,
+			},
 		},
 		"previous parallelism != previous completions, new parallelism == new completions": {
 			old: batch.Job{
@@ -1314,9 +1317,8 @@ func TestValidateJobUpdate(t *testing.T) {
 				job.Spec.Completions = pointer.Int32Ptr(3)
 				job.Spec.Parallelism = pointer.Int32Ptr(3)
 			},
-			err: &field.Error{
+			opts: JobValidationOptions{
-				Type:  field.ErrorTypeInvalid,
+				AllowElasticIndexedJobs: true,
-				Field: "spec.completions",
 			},
 		},
 		"indexed job updating completions and parallelism to different values is invalid": {
@@ -1334,6 +1336,9 @@ func TestValidateJobUpdate(t *testing.T) {
 				job.Spec.Completions = pointer.Int32Ptr(2)
 				job.Spec.Parallelism = pointer.Int32Ptr(3)
 			},
+			opts: JobValidationOptions{
+				AllowElasticIndexedJobs: true,
+			},
 			err: &field.Error{
 				Type:  field.ErrorTypeInvalid,
 				Field: "spec.completions",
@@ -1354,6 +1359,9 @@ func TestValidateJobUpdate(t *testing.T) {
 				job.Spec.Completions = nil
 				job.Spec.Parallelism = pointer.Int32Ptr(3)
 			},
+			opts: JobValidationOptions{
+				AllowElasticIndexedJobs: true,
+			},
 			err: &field.Error{
 				Type:  field.ErrorTypeRequired,
 				Field: "spec.completions",
@@ -1374,6 +1382,9 @@ func TestValidateJobUpdate(t *testing.T) {
 				job.Spec.Completions = pointer.Int32Ptr(2)
 				job.Spec.Parallelism = pointer.Int32Ptr(1)
 			},
+			opts: JobValidationOptions{
+				AllowElasticIndexedJobs: true,
+			},
 		},
 		"indexed job with completions unchanged, parallelism increased higher than completions": {
 			old: batch.Job{
@@ -1390,6 +1401,9 @@ func TestValidateJobUpdate(t *testing.T) {
 				job.Spec.Completions = pointer.Int32Ptr(2)
 				job.Spec.Parallelism = pointer.Int32Ptr(3)
 			},
+			opts: JobValidationOptions{
+				AllowElasticIndexedJobs: true,
+			},
 		},
 	}
 	ignoreValueAndDetail := cmpopts.IgnoreFields(field.Error{}, "BadValue", "Detail")

pkg/registry/batch/job/strategy.go

@@ -185,6 +185,8 @@ func validationOptionsForJob(newJob, oldJob *batch.Job) batchvalidation.JobValid
 		opts.AllowMutableSchedulingDirectives = utilfeature.DefaultFeatureGate.Enabled(features.JobMutableNodeSchedulingDirectives) &&
 			suspended && notStarted
 
+		// Elastic indexed jobs (mutable completions iff updated parallelism == updated completions)
+		opts.AllowElasticIndexedJobs = utilfeature.DefaultFeatureGate.Enabled(features.ElasticIndexedJob)
 	}
 
 	return opts