apiserver: allow zero value for the 'nominalConcurrencyShares' field

pkg/apis/flowcontrol/fuzzer/fuzzer.go

@@ -21,6 +21,7 @@ import (
 
 	runtimeserializer "k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/kubernetes/pkg/apis/flowcontrol"
+	"k8s.io/utils/ptr"
 )
 
 // Funcs returns the fuzzer functions for the flowcontrol api group.
@@ -28,20 +29,23 @@ var Funcs = func(codecs runtimeserializer.CodecFactory) []interface{} {
 	return []interface{}{
 		func(obj *flowcontrol.LimitedPriorityLevelConfiguration, c fuzz.Continue) {
 			c.FuzzNoCustom(obj) // fuzz self without calling this function again
+
+			// NOTE: setting a zero value here will cause the roundtrip
+			// test (from internal to v1beta2, v1beta1) to fail
+			if obj.NominalConcurrencyShares == 0 {
+				obj.NominalConcurrencyShares = int32(1)
+			}
 			if obj.LendablePercent == nil {
-				i := int32(0)
+				obj.LendablePercent = ptr.To(int32(0))
-				obj.LendablePercent = &i
 			}
 		},
 		func(obj *flowcontrol.ExemptPriorityLevelConfiguration, c fuzz.Continue) {
 			c.FuzzNoCustom(obj) // fuzz self without calling this function again
 			if obj.NominalConcurrencyShares == nil {
-				i := int32(0)
+				obj.NominalConcurrencyShares = ptr.To(int32(0))
-				obj.NominalConcurrencyShares = &i
 			}
 			if obj.LendablePercent == nil {
-				i := int32(0)
+				obj.LendablePercent = ptr.To(int32(0))
-				obj.LendablePercent = &i
 			}
 		},
 	}

pkg/apis/flowcontrol/types.go

@@ -419,6 +419,10 @@ type LimitedPriorityLevelConfiguration struct {
 	// Bigger numbers mean a larger nominal concurrency limit,
 	// at the expense of every other priority level.
 	// This field has a default value of 30.
+	//
+	// Setting this field to zero allows for the construction of a
+	// "jail" for this priority level that is used to hold some request(s)
+	//
 	// +optional
 	NominalConcurrencyShares int32
 

pkg/apis/flowcontrol/v1/defaults.go

@@ -18,6 +18,7 @@ package v1
 
 import (
 	v1 "k8s.io/api/flowcontrol/v1"
+	"k8s.io/utils/ptr"
 )
 
 // Default settings for flow-schema
@@ -52,8 +53,8 @@ func SetDefaults_ExemptPriorityLevelConfiguration(eplc *v1.ExemptPriorityLevelCo
 }
 
 func SetDefaults_LimitedPriorityLevelConfiguration(lplc *v1.LimitedPriorityLevelConfiguration) {
-	if lplc.NominalConcurrencyShares == 0 {
+	if lplc.NominalConcurrencyShares == nil {
-		lplc.NominalConcurrencyShares = PriorityLevelConfigurationDefaultNominalConcurrencyShares
+		lplc.NominalConcurrencyShares = ptr.To(PriorityLevelConfigurationDefaultNominalConcurrencyShares)
 	}
 	if lplc.LendablePercent == nil {
 		lplc.LendablePercent = new(int32)

pkg/apis/flowcontrol/v1/defaults_test.go

@@ -57,7 +57,7 @@ func TestDefaultWithPriorityLevelConfiguration(t *testing.T) {
 				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
 					Type: flowcontrolv1.PriorityLevelEnablementLimited,
 					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
-						NominalConcurrencyShares: 5,
+						NominalConcurrencyShares: ptr.To(int32(5)),
 						LimitResponse: flowcontrolv1.LimitResponse{
 							Type: flowcontrolv1.LimitResponseTypeReject,
 						},
@@ -68,7 +68,59 @@ func TestDefaultWithPriorityLevelConfiguration(t *testing.T) {
 				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
 					Type: flowcontrolv1.PriorityLevelEnablementLimited,
 					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
-						NominalConcurrencyShares: 5,
+						NominalConcurrencyShares: ptr.To(int32(5)),
+						LendablePercent:          ptr.To(int32(0)),
+						LimitResponse: flowcontrolv1.LimitResponse{
+							Type: flowcontrolv1.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "NominalConcurrencyShares is not specified in Limited, should default to 30",
+			original: &flowcontrolv1.PriorityLevelConfiguration{
+				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: nil,
+						LimitResponse: flowcontrolv1.LimitResponse{
+							Type: flowcontrolv1.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1.PriorityLevelConfiguration{
+				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: ptr.To(int32(PriorityLevelConfigurationDefaultNominalConcurrencyShares)),
+						LendablePercent:          ptr.To(int32(0)),
+						LimitResponse: flowcontrolv1.LimitResponse{
+							Type: flowcontrolv1.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "NominalConcurrencyShares is set to zero in Limited, no defaulting should be applied",
+			original: &flowcontrolv1.PriorityLevelConfiguration{
+				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: ptr.To(int32(0)),
+						LimitResponse: flowcontrolv1.LimitResponse{
+							Type: flowcontrolv1.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1.PriorityLevelConfiguration{
+				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: ptr.To(int32(0)),
 						LendablePercent:          ptr.To(int32(0)),
 						LimitResponse: flowcontrolv1.LimitResponse{
 							Type: flowcontrolv1.LimitResponseTypeReject,

pkg/apis/flowcontrol/v1beta3/conversion.go

@@ -0,0 +1,87 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1beta3
+
+import (
+	"k8s.io/api/flowcontrol/v1beta3"
+	"k8s.io/apimachinery/pkg/conversion"
+	"k8s.io/kubernetes/pkg/apis/flowcontrol"
+)
+
+func Convert_v1beta3_PriorityLevelConfiguration_To_flowcontrol_PriorityLevelConfiguration(in *v1beta3.PriorityLevelConfiguration, out *flowcontrol.PriorityLevelConfiguration, s conversion.Scope) error {
+	if err := autoConvert_v1beta3_PriorityLevelConfiguration_To_flowcontrol_PriorityLevelConfiguration(in, out, nil); err != nil {
+		return err
+	}
+
+	// during v1beta3 -> internal conversion:
+	//  - remove the roundtrip annotation for the 'NominalConcurrencyShares' field
+	//  - make sure we don't mutate the source (v1beta3) object's annotations
+	annotations, copied := dropPriorityLevelConcurrencyShareDefaultAnnotation(out.ObjectMeta.Annotations)
+	if copied {
+		out.ObjectMeta.Annotations = annotations
+	}
+	return nil
+}
+
+func Convert_flowcontrol_PriorityLevelConfiguration_To_v1beta3_PriorityLevelConfiguration(in *flowcontrol.PriorityLevelConfiguration, out *v1beta3.PriorityLevelConfiguration, s conversion.Scope) error {
+	if err := autoConvert_flowcontrol_PriorityLevelConfiguration_To_v1beta3_PriorityLevelConfiguration(in, out, nil); err != nil {
+		return err
+	}
+
+	// during internal -> v1beta3 conversion:
+	//  - add the roundtrip annotation for the 'NominalConcurrencyShares' field,
+	//      IIF the 'NominalConcurrencyShares' field has a value of zero.
+	//  - make sure we don't mutate the source (internal) object's annotations
+	if limited := in.Spec.Limited; limited != nil && limited.NominalConcurrencyShares == 0 {
+		annotations, copied := addPriorityLevelConcurrencyShareDefaultAnnotation(out.ObjectMeta.Annotations)
+		if copied {
+			out.ObjectMeta.Annotations = annotations
+		}
+	}
+
+	return nil
+}
+
+func dropPriorityLevelConcurrencyShareDefaultAnnotation(in map[string]string) (map[string]string, bool) {
+	if _, ok := in[v1beta3.PriorityLevelConcurrencyShareDefaultKey]; !ok {
+		return in, false
+	}
+
+	out := copyStringMap(in)
+	delete(out, v1beta3.PriorityLevelConcurrencyShareDefaultKey)
+	return out, true
+}
+
+func addPriorityLevelConcurrencyShareDefaultAnnotation(in map[string]string) (map[string]string, bool) {
+	if _, ok := in[v1beta3.PriorityLevelConcurrencyShareDefaultKey]; ok {
+		return in, false
+	}
+
+	out := copyStringMap(in)
+	out[v1beta3.PriorityLevelConcurrencyShareDefaultKey] = ""
+	return out, true
+}
+
+// copyStringMap returns a copy of the input map.
+// If input is nil, an empty map is returned.
+func copyStringMap(in map[string]string) map[string]string {
+	out := make(map[string]string, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}

pkg/apis/flowcontrol/v1beta3/conversion_test.go

@@ -0,0 +1,236 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1beta3
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"k8s.io/api/flowcontrol/v1beta3"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/kubernetes/pkg/apis/flowcontrol"
+)
+
+func TestConvert_v1beta3_PriorityLevelConfiguration_To_flowcontrol_PriorityLevelConfiguration(t *testing.T) {
+	inObjFn := func(shares int32, annotations map[string]string) *v1beta3.PriorityLevelConfiguration {
+		return &v1beta3.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: annotations,
+			},
+			Spec: v1beta3.PriorityLevelConfigurationSpec{
+				Type: v1beta3.PriorityLevelEnablementLimited,
+				Limited: &v1beta3.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: shares,
+					LimitResponse: v1beta3.LimitResponse{
+						Type: v1beta3.LimitResponseTypeReject,
+					},
+				},
+			},
+		}
+	}
+
+	outObjFn := func(shares int32, annotations map[string]string) *flowcontrol.PriorityLevelConfiguration {
+		return &flowcontrol.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: annotations,
+			},
+			Spec: flowcontrol.PriorityLevelConfigurationSpec{
+				Type: flowcontrol.PriorityLevelEnablementLimited,
+				Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: shares,
+					LimitResponse: flowcontrol.LimitResponse{
+						Type: flowcontrol.LimitResponseTypeReject,
+					},
+				},
+			},
+		}
+	}
+
+	tests := []struct {
+		name     string
+		in       *v1beta3.PriorityLevelConfiguration
+		expected *flowcontrol.PriorityLevelConfiguration
+	}{
+		{
+			name: "v1beta3 object, the roundtrip annotation is set, NominalConcurrencyShares is zero; the internal object should not have the roundtrip annotation set",
+			in: inObjFn(0, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+			}),
+			expected: outObjFn(0, map[string]string{
+				"foo": "bar",
+			}),
+		},
+		{
+			name: "v1beta3 object; the internal object should not have the roundtrip annotation set",
+			in: &v1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						"foo": "bar",
+						v1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+			},
+			expected: &flowcontrol.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						"foo": "bar",
+					},
+				},
+			},
+		},
+		{
+			name: "v1beta3 object, the roundtrip annotation is not set, NominalConcurrencyShares is zero; the internal object should not have the roundtrip annotation set",
+			in: inObjFn(0, map[string]string{
+				"foo": "bar",
+			}),
+			expected: outObjFn(0, map[string]string{
+				"foo": "bar",
+			}),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			copy := test.in.DeepCopy()
+
+			out := &flowcontrol.PriorityLevelConfiguration{}
+			if err := Convert_v1beta3_PriorityLevelConfiguration_To_flowcontrol_PriorityLevelConfiguration(test.in, out, nil); err != nil {
+				t.Errorf("Expected no error, but got: %v", err)
+			}
+			if !cmp.Equal(test.expected, out) {
+				t.Errorf("Expected a match, diff: %s", cmp.Diff(test.expected, out))
+			}
+			if want, got := copy.ObjectMeta.Annotations, test.in.ObjectMeta.Annotations; !cmp.Equal(want, got) {
+				t.Errorf("Did not expect the 'Annotations' field of the source to be mutated, diff: %s", cmp.Diff(want, got))
+			}
+		})
+	}
+}
+
+func TestConvert_flowcontrol_PriorityLevelConfiguration_To_v1beta3_PriorityLevelConfiguration(t *testing.T) {
+	inObjFn := func(shares int32, annotations map[string]string) *flowcontrol.PriorityLevelConfiguration {
+		return &flowcontrol.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: annotations,
+			},
+			Spec: flowcontrol.PriorityLevelConfigurationSpec{
+				Type: flowcontrol.PriorityLevelEnablementLimited,
+				Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: shares,
+					LimitResponse: flowcontrol.LimitResponse{
+						Type: flowcontrol.LimitResponseTypeReject,
+					},
+				},
+			},
+		}
+	}
+
+	outObjFn := func(shares int32, annotations map[string]string) *v1beta3.PriorityLevelConfiguration {
+		return &v1beta3.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: annotations,
+			},
+			Spec: v1beta3.PriorityLevelConfigurationSpec{
+				Type: v1beta3.PriorityLevelEnablementLimited,
+				Limited: &v1beta3.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: shares,
+					LimitResponse: v1beta3.LimitResponse{
+						Type: v1beta3.LimitResponseTypeReject,
+					},
+				},
+			},
+		}
+	}
+
+	tests := []struct {
+		name     string
+		in       *flowcontrol.PriorityLevelConfiguration
+		expected *v1beta3.PriorityLevelConfiguration
+	}{
+		{
+			name: "internal object, NominalConcurrencyShares is 0; v1beta3 object should have the roundtrip annotation set",
+			in: inObjFn(0, map[string]string{
+				"foo": "bar",
+			}),
+			expected: outObjFn(0, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+			}),
+		},
+		{
+			name: "internal object, NominalConcurrencyShares is not 0; v1beta3 object should not have the roundtrip annotation set",
+			in: inObjFn(1, map[string]string{
+				"foo": "bar",
+			}),
+			expected: outObjFn(1, map[string]string{
+				"foo": "bar",
+			}),
+		},
+		{
+			name: "internal object, the roundtrip annotation is set, NominalConcurrencyShares is 0, no error expected",
+			in: inObjFn(0, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+			}),
+			expected: outObjFn(0, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+			}),
+		},
+		{
+			name: "internal object, the roundtrip annotation is set with a non-empty value, NominalConcurrencyShares is 0, the annotation value should be preserved",
+			in: inObjFn(0, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "non-empty",
+			}),
+			expected: outObjFn(0, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "non-empty",
+			}),
+		},
+		{
+			name: "internal object, the roundtrip annotation is set with a non-empty value, NominalConcurrencyShares is not 0, the annotation value should be preserved",
+			in: inObjFn(1, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "non-empty",
+			}),
+			expected: outObjFn(1, map[string]string{
+				"foo": "bar",
+				v1beta3.PriorityLevelConcurrencyShareDefaultKey: "non-empty",
+			}),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			copy := test.in.DeepCopy()
+
+			out := &v1beta3.PriorityLevelConfiguration{}
+			if err := Convert_flowcontrol_PriorityLevelConfiguration_To_v1beta3_PriorityLevelConfiguration(test.in, out, nil); err != nil {
+				t.Errorf("Expected no error, but got: %v", err)
+			}
+
+			if !cmp.Equal(test.expected, out) {
+				t.Errorf("Expected a match, diff: %s", cmp.Diff(test.expected, out))
+			}
+			if want, got := copy.ObjectMeta.Annotations, test.in.ObjectMeta.Annotations; !cmp.Equal(want, got) {
+				t.Errorf("Did not expect the 'Annotations' field of the source to be mutated, diff: %s", cmp.Diff(want, got))
+			}
+		})
+	}
+}

pkg/apis/flowcontrol/v1beta3/defaults.go

@@ -18,6 +18,7 @@ package v1beta3
 
 import (
 	"k8s.io/api/flowcontrol/v1beta3"
+	"k8s.io/utils/ptr"
 )
 
 // Default settings for flow-schema
@@ -40,28 +41,38 @@ func SetDefaults_FlowSchemaSpec(spec *v1beta3.FlowSchemaSpec) {
 	}
 }
 
+// SetDefaults_PriorityLevelConfiguration sets the default values for a
+// PriorityLevelConfiguration object. Since we need to inspect the presence
+// of the roundtrip annotation in order to determine whether the user has
+// specified a zero value for the 'NominalConcurrencyShares' field,
+// the defaulting logic needs visibility to the annotations field.
+func SetDefaults_PriorityLevelConfiguration(in *v1beta3.PriorityLevelConfiguration) {
+	if limited := in.Spec.Limited; limited != nil {
+		// for v1beta3, we apply a default value to the NominalConcurrencyShares
+		// field only when:
+		//   a) NominalConcurrencyShares == 0, and
+		//   b) the roundtrip annotation is not set
+		if _, ok := in.Annotations[v1beta3.PriorityLevelConcurrencyShareDefaultKey]; !ok && limited.NominalConcurrencyShares == 0 {
+			limited.NominalConcurrencyShares = PriorityLevelConfigurationDefaultNominalConcurrencyShares
+		}
+	}
+}
+
 func SetDefaults_ExemptPriorityLevelConfiguration(eplc *v1beta3.ExemptPriorityLevelConfiguration) {
 	if eplc.NominalConcurrencyShares == nil {
-		eplc.NominalConcurrencyShares = new(int32)
+		eplc.NominalConcurrencyShares = ptr.To(int32(0))
-		*eplc.NominalConcurrencyShares = 0
 	}
 	if eplc.LendablePercent == nil {
-		eplc.LendablePercent = new(int32)
+		eplc.LendablePercent = ptr.To(int32(0))
-		*eplc.LendablePercent = 0
 	}
 }
 
-func SetDefaults_LimitedPriorityLevelConfiguration(lplc *v1beta3.LimitedPriorityLevelConfiguration) {
+func SetDefaults_LimitedPriorityLevelConfiguration(in *v1beta3.LimitedPriorityLevelConfiguration) {
-	if lplc.NominalConcurrencyShares == 0 {
+	if in.LendablePercent == nil {
-		lplc.NominalConcurrencyShares = PriorityLevelConfigurationDefaultNominalConcurrencyShares
+		in.LendablePercent = ptr.To(int32(0))
-	}
-	if lplc.LendablePercent == nil {
-		lplc.LendablePercent = new(int32)
-		*lplc.LendablePercent = 0
 	}
 }
 
-// SetDefaults_FlowSchema sets default values for flow schema
 func SetDefaults_QueuingConfiguration(cfg *v1beta3.QueuingConfiguration) {
 	if cfg.HandSize == 0 {
 		cfg.HandSize = PriorityLevelConfigurationDefaultHandSize

pkg/apis/flowcontrol/v1beta3/defaults_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 
 	flowcontrolv1beta3 "k8s.io/api/flowcontrol/v1beta3"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/utils/pointer"
 )
@@ -77,6 +78,172 @@ func TestDefaultWithPriorityLevelConfiguration(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "Defaulting for queuing configuration",
+			original: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 5,
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type:    flowcontrolv1beta3.LimitResponseTypeQueue,
+							Queuing: &flowcontrolv1beta3.QueuingConfiguration{},
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 5,
+						LendablePercent:          pointer.Int32(0),
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeQueue,
+							Queuing: &flowcontrolv1beta3.QueuingConfiguration{
+								HandSize:         PriorityLevelConfigurationDefaultHandSize,
+								Queues:           PriorityLevelConfigurationDefaultQueues,
+								QueueLengthLimit: PriorityLevelConfigurationDefaultQueueLengthLimit,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "NominalConcurrencyShares is 0, roundtrip annotation is not set, default value should be applied",
+			original: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 0,
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: PriorityLevelConfigurationDefaultNominalConcurrencyShares,
+						LendablePercent:          pointer.Int32(0),
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "NominalConcurrencyShares is 0, roundtrip annotation is set, default value should not be applied",
+			original: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 0,
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 0,
+						LendablePercent:          pointer.Int32(0),
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "NominalConcurrencyShares is 0, roundtrip annotation is set with a non-empty string, default value should not be applied",
+			original: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 0,
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 0,
+						LendablePercent:          pointer.Int32(0),
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "NominalConcurrencyShares is positive, roundtrip annotation is set, annotation should be ignored",
+			original: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 1,
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+			expected: &flowcontrolv1beta3.PriorityLevelConfiguration{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+					},
+				},
+				Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+					Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+					Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+						NominalConcurrencyShares: 1,
+						LendablePercent:          pointer.Int32(0),
+						LimitResponse: flowcontrolv1beta3.LimitResponse{
+							Type: flowcontrolv1beta3.LimitResponseTypeReject,
+						},
+					},
+				},
+			},
+		},
 	}
 
 	scheme := runtime.NewScheme()

pkg/apis/flowcontrol/validation/validation.go

@@ -74,6 +74,14 @@ var supportedLimitResponseType = sets.NewString(
 	string(flowcontrol.LimitResponseTypeReject),
 )
 
+// PriorityLevelValidationOptions holds the validation options for a priority level object
+type PriorityLevelValidationOptions struct {
+	// AllowZeroLimitedNominalConcurrencyShares, if true, indicates that we allow
+	// a zero value for the 'nominalConcurrencyShares' field of the 'limited'
+	// section of a priority level.
+	AllowZeroLimitedNominalConcurrencyShares bool
+}
+
 // ValidateFlowSchema validates the content of flow-schema
 func ValidateFlowSchema(fs *flowcontrol.FlowSchema) field.ErrorList {
 	allErrs := apivalidation.ValidateObjectMeta(&fs.ObjectMeta, false, ValidateFlowSchemaName, field.NewPath("metadata"))
@@ -340,10 +348,11 @@ func ValidateFlowSchemaCondition(condition *flowcontrol.FlowSchemaCondition, fld
 }
 
 // ValidatePriorityLevelConfiguration validates priority-level-configuration.
-func ValidatePriorityLevelConfiguration(pl *flowcontrol.PriorityLevelConfiguration, requestGV schema.GroupVersion) field.ErrorList {
+func ValidatePriorityLevelConfiguration(pl *flowcontrol.PriorityLevelConfiguration, requestGV schema.GroupVersion, opts PriorityLevelValidationOptions) field.ErrorList {
 	allErrs := apivalidation.ValidateObjectMeta(&pl.ObjectMeta, false, ValidatePriorityLevelConfigurationName, field.NewPath("metadata"))
+
 	specPath := field.NewPath("spec")
-	allErrs = append(allErrs, ValidatePriorityLevelConfigurationSpec(&pl.Spec, requestGV, pl.Name, specPath)...)
+	allErrs = append(allErrs, ValidatePriorityLevelConfigurationSpec(&pl.Spec, requestGV, pl.Name, specPath, opts)...)
 	allErrs = append(allErrs, ValidateIfMandatoryPriorityLevelConfigurationObject(pl, specPath)...)
 	allErrs = append(allErrs, ValidatePriorityLevelConfigurationStatus(&pl.Status, field.NewPath("status"))...)
 	return allErrs
@@ -380,7 +389,7 @@ func ValidateIfMandatoryPriorityLevelConfigurationObject(pl *flowcontrol.Priorit
 }
 
 // ValidatePriorityLevelConfigurationSpec validates priority-level-configuration's spec.
-func ValidatePriorityLevelConfigurationSpec(spec *flowcontrol.PriorityLevelConfigurationSpec, requestGV schema.GroupVersion, name string, fldPath *field.Path) field.ErrorList {
+func ValidatePriorityLevelConfigurationSpec(spec *flowcontrol.PriorityLevelConfigurationSpec, requestGV schema.GroupVersion, name string, fldPath *field.Path, opts PriorityLevelValidationOptions) field.ErrorList {
 	var allErrs field.ErrorList
 	if (name == flowcontrol.PriorityLevelConfigurationNameExempt) != (spec.Type == flowcontrol.PriorityLevelEnablementExempt) {
 		allErrs = append(allErrs, field.Invalid(fldPath.Child("type"), spec.Type, "type must be 'Exempt' if and only if name is 'exempt'"))
@@ -401,7 +410,7 @@ func ValidatePriorityLevelConfigurationSpec(spec *flowcontrol.PriorityLevelConfi
 		if spec.Limited == nil {
 			allErrs = append(allErrs, field.Required(fldPath.Child("limited"), "must not be empty when type is Limited"))
 		} else {
-			allErrs = append(allErrs, ValidateLimitedPriorityLevelConfiguration(spec.Limited, requestGV, fldPath.Child("limited"))...)
+			allErrs = append(allErrs, ValidateLimitedPriorityLevelConfiguration(spec.Limited, requestGV, fldPath.Child("limited"), opts)...)
 		}
 	default:
 		allErrs = append(allErrs, field.NotSupported(fldPath.Child("type"), spec.Type, supportedPriorityLevelEnablement.List()))
@@ -410,10 +419,16 @@ func ValidatePriorityLevelConfigurationSpec(spec *flowcontrol.PriorityLevelConfi
 }
 
 // ValidateLimitedPriorityLevelConfiguration validates the configuration for an execution-limited priority level
-func ValidateLimitedPriorityLevelConfiguration(lplc *flowcontrol.LimitedPriorityLevelConfiguration, requestGV schema.GroupVersion, fldPath *field.Path) field.ErrorList {
+func ValidateLimitedPriorityLevelConfiguration(lplc *flowcontrol.LimitedPriorityLevelConfiguration, requestGV schema.GroupVersion, fldPath *field.Path, opts PriorityLevelValidationOptions) field.ErrorList {
 	var allErrs field.ErrorList
-	if lplc.NominalConcurrencyShares <= 0 {
+	if opts.AllowZeroLimitedNominalConcurrencyShares {
-		allErrs = append(allErrs, field.Invalid(fldPath.Child(getVersionedFieldNameForConcurrencyShares(requestGV)), lplc.NominalConcurrencyShares, "must be positive"))
+		if lplc.NominalConcurrencyShares < 0 {
+			allErrs = append(allErrs, field.Invalid(fldPath.Child(getVersionedFieldNameForConcurrencyShares(requestGV)), lplc.NominalConcurrencyShares, "must be a non-negative integer"))
+		}
+	} else {
+		if lplc.NominalConcurrencyShares <= 0 {
+			allErrs = append(allErrs, field.Invalid(fldPath.Child(getVersionedFieldNameForConcurrencyShares(requestGV)), lplc.NominalConcurrencyShares, "must be positive"))
+		}
 	}
 	allErrs = append(allErrs, ValidateLimitResponse(lplc.LimitResponse, fldPath.Child("limitResponse"))...)
 

pkg/apis/flowcontrol/validation/validation_test.go

@@ -800,6 +800,7 @@ func TestPriorityLevelConfigurationValidation(t *testing.T) {
 	testCases := []struct {
 		name                       string
 		priorityLevelConfiguration *flowcontrol.PriorityLevelConfiguration
+		requestGV                  *schema.GroupVersion
 		expectedErrors             field.ErrorList
 	}{{
 		name: "exempt should work",
@@ -1105,10 +1106,34 @@ func TestPriorityLevelConfigurationValidation(t *testing.T) {
 		expectedErrors: field.ErrorList{
 			field.Invalid(field.NewPath("spec").Child("limited").Child("limitResponse").Child("queuing").Child("handSize"), int32(8), "should not be greater than queues (7)"),
 		},
+	}, {
+		name: "the roundtrip annotation is not forbidden in v1beta3",
+		priorityLevelConfiguration: &flowcontrol.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "with-forbidden-annotation",
+				Annotations: map[string]string{
+					flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey: "",
+				},
+			},
+			Spec: flowcontrol.PriorityLevelConfigurationSpec{
+				Type: flowcontrol.PriorityLevelEnablementLimited,
+				Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: 42,
+					LimitResponse: flowcontrol.LimitResponse{
+						Type: flowcontrol.LimitResponseTypeReject},
+				},
+			},
+		},
+		requestGV:      &flowcontrolv1beta3.SchemeGroupVersion,
+		expectedErrors: field.ErrorList{},
 	}}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			errs := ValidatePriorityLevelConfiguration(testCase.priorityLevelConfiguration, flowcontrolv1beta3.SchemeGroupVersion)
+			gv := flowcontrolv1beta3.SchemeGroupVersion
+			if testCase.requestGV != nil {
+				gv = *testCase.requestGV
+			}
+			errs := ValidatePriorityLevelConfiguration(testCase.priorityLevelConfiguration, gv, PriorityLevelValidationOptions{})
 			if !assert.ElementsMatch(t, testCase.expectedErrors, errs) {
 				t.Logf("mismatch: %v", cmp.Diff(testCase.expectedErrors, errs))
 			}
@@ -1268,42 +1293,73 @@ func TestValidateNonResourceURLPath(t *testing.T) {
 }
 
 func TestValidateLimitedPriorityLevelConfiguration(t *testing.T) {
-	errExpectedFn := func(fieldName string) field.ErrorList {
+	errExpectedFn := func(fieldName string, v int32, msg string) field.ErrorList {
 		return field.ErrorList{
-			field.Invalid(field.NewPath("spec").Child("limited").Child(fieldName), int32(0), "must be positive"),
+			field.Invalid(field.NewPath("spec").Child("limited").Child(fieldName), int32(v), msg),
 		}
 	}
 
 	tests := []struct {
 		requestVersion    schema.GroupVersion
+		allowZero         bool
 		concurrencyShares int32
 		errExpected       field.ErrorList
 	}{{
 		requestVersion:    flowcontrolv1beta1.SchemeGroupVersion,
 		concurrencyShares: 0,
-		errExpected:       errExpectedFn("assuredConcurrencyShares"),
+		errExpected:       errExpectedFn("assuredConcurrencyShares", 0, "must be positive"),
 	}, {
 		requestVersion:    flowcontrolv1beta2.SchemeGroupVersion,
 		concurrencyShares: 0,
-		errExpected:       errExpectedFn("assuredConcurrencyShares"),
+		errExpected:       errExpectedFn("assuredConcurrencyShares", 0, "must be positive"),
 	}, {
 		requestVersion:    flowcontrolv1beta3.SchemeGroupVersion,
 		concurrencyShares: 0,
-		errExpected:       errExpectedFn("nominalConcurrencyShares"),
+		errExpected:       errExpectedFn("nominalConcurrencyShares", 0, "must be positive"),
 	}, {
 		requestVersion:    flowcontrolv1.SchemeGroupVersion,
 		concurrencyShares: 0,
-		errExpected:       errExpectedFn("nominalConcurrencyShares"),
+		errExpected:       errExpectedFn("nominalConcurrencyShares", 0, "must be positive"),
+	}, {
+		requestVersion:    flowcontrolv1beta3.SchemeGroupVersion,
+		concurrencyShares: 100,
+		errExpected:       nil,
+	}, {
+		requestVersion:    flowcontrolv1beta3.SchemeGroupVersion,
+		allowZero:         true,
+		concurrencyShares: 0,
+		errExpected:       nil,
+	}, {
+		requestVersion:    flowcontrolv1beta3.SchemeGroupVersion,
+		allowZero:         true,
+		concurrencyShares: -1,
+		errExpected:       errExpectedFn("nominalConcurrencyShares", -1, "must be a non-negative integer"),
+	}, {
+		requestVersion:    flowcontrolv1beta3.SchemeGroupVersion,
+		allowZero:         true,
+		concurrencyShares: 1,
+		errExpected:       nil,
+	}, {
+		requestVersion:    flowcontrolv1.SchemeGroupVersion,
+		allowZero:         true,
+		concurrencyShares: 0,
+		errExpected:       nil,
+	}, {
+		requestVersion:    flowcontrolv1.SchemeGroupVersion,
+		allowZero:         true,
+		concurrencyShares: -1,
+		errExpected:       errExpectedFn("nominalConcurrencyShares", -1, "must be a non-negative integer"),
+	}, {
+		requestVersion:    flowcontrolv1.SchemeGroupVersion,
+		allowZero:         true,
+		concurrencyShares: 1,
+		errExpected:       nil,
 	}, {
 		// this should never really happen in real life, the request
 		// context should always contain the request {group, version}
 		requestVersion:    schema.GroupVersion{},
 		concurrencyShares: 0,
-		errExpected:       errExpectedFn("nominalConcurrencyShares"),
+		errExpected:       errExpectedFn("nominalConcurrencyShares", 0, "must be positive"),
-	}, {
-		requestVersion:    flowcontrolv1beta3.SchemeGroupVersion,
-		concurrencyShares: 100,
-		errExpected:       nil,
 	}}
 
 	for _, test := range tests {
@@ -1316,7 +1372,7 @@ func TestValidateLimitedPriorityLevelConfiguration(t *testing.T) {
 			}
 			specPath := field.NewPath("spec").Child("limited")
 
-			errGot := ValidateLimitedPriorityLevelConfiguration(configuration, test.requestVersion, specPath)
+			errGot := ValidateLimitedPriorityLevelConfiguration(configuration, test.requestVersion, specPath, PriorityLevelValidationOptions{AllowZeroLimitedNominalConcurrencyShares: test.allowZero})
 			if !cmp.Equal(test.errExpected, errGot) {
 				t.Errorf("Expected error: %v, diff: %s", test.errExpected, cmp.Diff(test.errExpected, errGot))
 			}
@@ -1394,7 +1450,7 @@ func TestValidateLimitedPriorityLevelConfigurationWithBorrowing(t *testing.T) {
 			}
 			specPath := field.NewPath("spec").Child("limited")
 
-			errGot := ValidateLimitedPriorityLevelConfiguration(configuration, flowcontrolv1.SchemeGroupVersion, specPath)
+			errGot := ValidateLimitedPriorityLevelConfiguration(configuration, flowcontrolv1.SchemeGroupVersion, specPath, PriorityLevelValidationOptions{})
 			if !cmp.Equal(test.errExpected, errGot) {
 				t.Errorf("Expected error: %v, diff: %s", test.errExpected, cmp.Diff(test.errExpected, errGot))
 			}

pkg/features/kube_features.go

@@ -1166,6 +1166,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	genericfeatures.UnauthenticatedHTTP2DOSMitigation: {Default: true, PreRelease: featuregate.Beta},
 
+	genericfeatures.ZeroLimitedNominalConcurrencyShares: {Default: false, PreRelease: featuregate.Beta},
+
 	// inherited features from apiextensions-apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:
 

pkg/registry/flowcontrol/ensurer/prioritylevelconfiguration_test.go

@@ -30,6 +30,7 @@ import (
 	toolscache "k8s.io/client-go/tools/cache"
 	flowcontrolapisv1 "k8s.io/kubernetes/pkg/apis/flowcontrol/v1"
 	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 
 	"github.com/google/go-cmp/cmp"
 )
@@ -265,7 +266,7 @@ func TestPriorityLevelSpecChanged(t *testing.T) {
 		Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
 			Type: flowcontrolv1.PriorityLevelEnablementLimited,
 			Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 1,
+				NominalConcurrencyShares: ptr.To(int32(1)),
 			},
 		},
 	}
@@ -273,7 +274,7 @@ func TestPriorityLevelSpecChanged(t *testing.T) {
 		Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
 			Type: flowcontrolv1.PriorityLevelEnablementLimited,
 			Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: flowcontrolapisv1.PriorityLevelConfigurationDefaultNominalConcurrencyShares,
+				NominalConcurrencyShares: ptr.To(flowcontrolapisv1.PriorityLevelConfigurationDefaultNominalConcurrencyShares),
 				LendablePercent:          pointer.Int32(0),
 				LimitResponse: flowcontrolv1.LimitResponse{
 					Type: flowcontrolv1.LimitResponseTypeReject,
@@ -306,7 +307,7 @@ func TestPriorityLevelSpecChanged(t *testing.T) {
 		Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
 			Type: flowcontrolv1.PriorityLevelEnablementLimited,
 			Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 1,
+				NominalConcurrencyShares: ptr.To(int32(1)),
 			},
 		},
 	}
@@ -486,7 +487,7 @@ func (b *plBuilder) WithAutoUpdateAnnotation(value string) *plBuilder {
 func (b *plBuilder) WithLimited(nominalConcurrencyShares int32) *plBuilder {
 	b.object.Spec.Type = flowcontrolv1.PriorityLevelEnablementLimited
 	b.object.Spec.Limited = &flowcontrolv1.LimitedPriorityLevelConfiguration{
-		NominalConcurrencyShares: nominalConcurrencyShares,
+		NominalConcurrencyShares: ptr.To(nominalConcurrencyShares),
 		LendablePercent:          pointer.Int32(0),
 		LimitResponse: flowcontrolv1.LimitResponse{
 			Type: flowcontrolv1.LimitResponseTypeReject,

pkg/registry/flowcontrol/prioritylevelconfiguration/strategy.go

@@ -24,7 +24,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/features"
 	"k8s.io/apiserver/pkg/storage/names"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/apis/flowcontrol"
 	"k8s.io/kubernetes/pkg/apis/flowcontrol/validation"
@@ -87,7 +89,22 @@ func (priorityLevelConfigurationStrategy) PrepareForUpdate(ctx context.Context,
 
 // Validate validates a new priority-level.
 func (priorityLevelConfigurationStrategy) Validate(ctx context.Context, obj runtime.Object) field.ErrorList {
-	return validation.ValidatePriorityLevelConfiguration(obj.(*flowcontrol.PriorityLevelConfiguration), getRequestGroupVersion(ctx))
+	// 1.28 server is not aware of the roundtrip annotation, and will
+	// default any 0 value persisted (for the NominalConcurrencyShares
+	// field of a priority level configuration object) back to 30 when
+	// reading from etcd.
+	// That means we should not allow 0 values to be introduced, either
+	// via v1 or v1beta3(with the roundtrip annotation) until we know
+	// all servers are at 1.29+ and will honor the zero value correctly.
+	//
+	// TODO(121510): 1.29: don't allow a zero value, either via v1 or
+	//  v1beta3 (with the roundtrip annotation) for the
+	//  'nominalConcurrencyShares' field of 'limited' for CREATE operation.
+	//  1:30: lift this restriction, allow zero value via v1 or v1beta3
+	opts := validation.PriorityLevelValidationOptions{
+		AllowZeroLimitedNominalConcurrencyShares: utilfeature.DefaultFeatureGate.Enabled(features.ZeroLimitedNominalConcurrencyShares),
+	}
+	return validation.ValidatePriorityLevelConfiguration(obj.(*flowcontrol.PriorityLevelConfiguration), getRequestGroupVersion(ctx), opts)
 }
 
 // WarningsOnCreate returns warnings for the creation of the given object.
@@ -110,7 +127,27 @@ func (priorityLevelConfigurationStrategy) AllowCreateOnUpdate() bool {
 
 // ValidateUpdate is the default update validation for an end user.
 func (priorityLevelConfigurationStrategy) ValidateUpdate(ctx context.Context, obj, old runtime.Object) field.ErrorList {
-	return validation.ValidatePriorityLevelConfiguration(obj.(*flowcontrol.PriorityLevelConfiguration), getRequestGroupVersion(ctx))
+	newPL := obj.(*flowcontrol.PriorityLevelConfiguration)
+	oldPL := old.(*flowcontrol.PriorityLevelConfiguration)
+
+	// 1.28 server is not aware of the roundtrip annotation, and will
+	// default any 0 value persisted (for the NominalConcurrencyShares
+	// field of a priority level configuration object) back to 30 when
+	// reading from etcd.
+	// That means we should not allow 0 values to be introduced, either
+	// via v1 or v1beta3(with the roundtrip annotation) until we know
+	// all servers are at 1.29+ and will honor the zero value correctly.
+	//
+	// TODO(121510): 1.29: only allow a zero value, either via v1 or
+	//  v1beta3 (with the roundtrip annotation) for the
+	//  'nominalConcurrencyShares' field of 'limited' for UPDATE operation,
+	//  only if the existing object already contains a zero value.
+	//  1:30: lift this restriction, allow zero value via v1 or v1beta3
+	opts := validation.PriorityLevelValidationOptions{
+		AllowZeroLimitedNominalConcurrencyShares: utilfeature.DefaultFeatureGate.Enabled(features.ZeroLimitedNominalConcurrencyShares) ||
+			hasZeroLimitedNominalConcurrencyShares(oldPL),
+	}
+	return validation.ValidatePriorityLevelConfiguration(newPL, getRequestGroupVersion(ctx), opts)
 }
 
 // WarningsOnUpdate returns warnings for the given update.
@@ -177,3 +214,7 @@ func getRequestGroupVersion(ctx context.Context) schema.GroupVersion {
 	}
 	return schema.GroupVersion{}
 }
+
+func hasZeroLimitedNominalConcurrencyShares(obj *flowcontrol.PriorityLevelConfiguration) bool {
+	return obj != nil && obj.Spec.Limited != nil && obj.Spec.Limited.NominalConcurrencyShares == 0
+}

pkg/registry/flowcontrol/prioritylevelconfiguration/strategy_test.go

@@ -0,0 +1,325 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package prioritylevelconfiguration
+
+import (
+	"context"
+	"testing"
+
+	flowcontrolv1 "k8s.io/api/flowcontrol/v1"
+	flowcontrolv1beta3 "k8s.io/api/flowcontrol/v1beta3"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	"k8s.io/apiserver/pkg/features"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/apis/flowcontrol"
+	"k8s.io/utils/ptr"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+func TestPriorityLevelConfigurationValidation(t *testing.T) {
+	v1ObjFn := func(v *int32) *flowcontrolv1.PriorityLevelConfiguration {
+		return &flowcontrolv1.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "foo",
+			},
+			Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
+				Type: flowcontrolv1.PriorityLevelEnablementLimited,
+				Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: v,
+					LimitResponse: flowcontrolv1.LimitResponse{
+						Type: flowcontrolv1.LimitResponseTypeReject},
+				},
+			},
+		}
+	}
+	v1beta3ObjFn := func(v int32, isZero bool) *flowcontrolv1beta3.PriorityLevelConfiguration {
+		obj := &flowcontrolv1beta3.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "foo",
+			},
+			Spec: flowcontrolv1beta3.PriorityLevelConfigurationSpec{
+				Type: flowcontrolv1beta3.PriorityLevelEnablementLimited,
+				Limited: &flowcontrolv1beta3.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: v,
+					LimitResponse: flowcontrolv1beta3.LimitResponse{
+						Type: flowcontrolv1beta3.LimitResponseTypeReject},
+				},
+			},
+		}
+		if isZero && v == 0 {
+			obj.ObjectMeta.Annotations = map[string]string{}
+			obj.ObjectMeta.Annotations[flowcontrolv1beta3.PriorityLevelConcurrencyShareDefaultKey] = ""
+		}
+		return obj
+	}
+	internalObjFn := func(v int32) *flowcontrol.PriorityLevelConfiguration {
+		return &flowcontrol.PriorityLevelConfiguration{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "foo",
+			},
+			Spec: flowcontrol.PriorityLevelConfigurationSpec{
+				Type: flowcontrol.PriorityLevelEnablementLimited,
+				Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
+					NominalConcurrencyShares: v,
+					LimitResponse: flowcontrol.LimitResponse{
+						Type: flowcontrol.LimitResponseTypeReject},
+				},
+			},
+		}
+	}
+	v1SchemeFn := func(t *testing.T) *runtime.Scheme {
+		scheme := runtime.NewScheme()
+		if err := flowcontrolv1.AddToScheme(scheme); err != nil {
+			t.Fatalf("Failed to add to scheme: %v", err)
+		}
+		return scheme
+	}
+	v1beta3SchemeFn := func(t *testing.T) *runtime.Scheme {
+		scheme := runtime.NewScheme()
+		if err := flowcontrolv1beta3.AddToScheme(scheme); err != nil {
+			t.Fatalf("Failed to add to scheme: %v", err)
+		}
+		return scheme
+	}
+	errExpectedFn := func(v int32, msg string) field.ErrorList {
+		return field.ErrorList{
+			field.Invalid(field.NewPath("spec").Child("limited").Child("nominalConcurrencyShares"), int32(v), msg),
+		}
+	}
+
+	tests := []struct {
+		name               string
+		obj                runtime.Object
+		old                *flowcontrol.PriorityLevelConfiguration // for UPDATE only
+		zeroFeatureEnabled bool
+		scheme             *runtime.Scheme
+		errExpected        field.ErrorList
+	}{
+		{
+			name:               "v1, feature disabled, create, zero value, error expected",
+			obj:                v1ObjFn(ptr.To(int32(0))),
+			zeroFeatureEnabled: false,
+			scheme:             v1SchemeFn(t),
+			errExpected:        errExpectedFn(0, "must be positive"),
+		},
+		{
+			name:               "v1, feature disabled, create, unset, no error expected",
+			obj:                v1ObjFn(nil),
+			zeroFeatureEnabled: false,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature disabled, create, non-zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(1))),
+			zeroFeatureEnabled: false,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature enabled, create, zero value, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(0))),
+			zeroFeatureEnabled: true,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature enabled, create, unset, no error expected",
+			obj:                v1ObjFn(nil),
+			zeroFeatureEnabled: true,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature enabled, create, non-zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(1))),
+			zeroFeatureEnabled: true,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature disabled, create, zero value, error expected",
+			obj:                v1beta3ObjFn(0, true),
+			zeroFeatureEnabled: false,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        errExpectedFn(0, "must be positive"),
+		},
+		{
+			name:               "v1beta3, feature disabled, create, zero value without annotation, no error expected",
+			obj:                v1beta3ObjFn(0, false),
+			zeroFeatureEnabled: false,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature disabled, create, non-zero, no error expected",
+			obj:                v1beta3ObjFn(1, false),
+			zeroFeatureEnabled: false,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature enabled, create, zero value, no error expected",
+			obj:                v1beta3ObjFn(0, true),
+			zeroFeatureEnabled: true,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature enabled, create, zero value without annotation, no error expected",
+			obj:                v1beta3ObjFn(0, false),
+			zeroFeatureEnabled: true,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature enabled, create, non-zero, no error expected",
+			obj:                v1beta3ObjFn(1, false),
+			zeroFeatureEnabled: true,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+
+		// the following use cases cover UPDATE
+		{
+			name:               "v1, feature disabled, update, zero value, existing has non-zero, error expected",
+			obj:                v1ObjFn(ptr.To(int32(0))),
+			old:                internalObjFn(1),
+			zeroFeatureEnabled: false,
+			scheme:             v1SchemeFn(t),
+			errExpected:        errExpectedFn(0, "must be positive"),
+		},
+		{
+			name:               "v1, feature disabled, update, zero value, existing has zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(0))),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: false,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature disabled, update, non-zero value, existing has zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(1))),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: false,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature enabled, update, zero value, existing has non-zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(0))),
+			old:                internalObjFn(1),
+			zeroFeatureEnabled: true,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature enabled, update, zero value, existing has zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(0))),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: true,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1, feature enabled, update, non-zero value, existing has zero, no error expected",
+			obj:                v1ObjFn(ptr.To(int32(1))),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: true,
+			scheme:             v1SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature disabled, update, zero value, existing has non-zero, error expected",
+			obj:                v1beta3ObjFn(0, true),
+			old:                internalObjFn(1),
+			zeroFeatureEnabled: false,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        errExpectedFn(0, "must be positive"),
+		},
+		{
+			name:               "v1beta3, feature disabled, update, zero value, existing has zero, no error expected",
+			obj:                v1beta3ObjFn(0, true),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: false,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature disabled, update, non-zero value, existing has zero, no error expected",
+			obj:                v1beta3ObjFn(1, false),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: false,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature enabled, update, zero value, existing has non-zero, error expected",
+			obj:                v1beta3ObjFn(0, true),
+			old:                internalObjFn(1),
+			zeroFeatureEnabled: true,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature enabled, update, zero value, existing has zero, no error expected",
+			obj:                v1beta3ObjFn(0, true),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: true,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+		{
+			name:               "v1beta3, feature enabled, update, non-zero value, existing has zero, no error expected",
+			obj:                v1beta3ObjFn(1, false),
+			old:                internalObjFn(0),
+			zeroFeatureEnabled: true,
+			scheme:             v1beta3SchemeFn(t),
+			errExpected:        nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ZeroLimitedNominalConcurrencyShares, test.zeroFeatureEnabled)()
+
+			scheme := test.scheme
+			scheme.Default(test.obj)
+
+			ctx := context.TODO()
+			internal := &flowcontrol.PriorityLevelConfiguration{}
+			if err := scheme.Convert(test.obj, internal, ctx); err != nil {
+				t.Errorf("Expected no error while converting to internal type: %v", err)
+			}
+
+			err := func(obj, old *flowcontrol.PriorityLevelConfiguration) field.ErrorList {
+				if old == nil {
+					return Strategy.Validate(ctx, obj) // for create operation
+				}
+				return Strategy.ValidateUpdate(ctx, obj, old) // for update operation
+			}(internal, test.old)
+
+			if !cmp.Equal(test.errExpected, err) {
+				t.Errorf("Expected error: %v, diff: %s", test.errExpected, cmp.Diff(test.errExpected, err))
+			}
+		})
+	}
+}

staging/src/k8s.io/api/flowcontrol/v1/types.go

@@ -476,9 +476,14 @@ type LimitedPriorityLevelConfiguration struct {
 	//
 	// Bigger numbers mean a larger nominal concurrency limit,
 	// at the expense of every other priority level.
-	// This field has a default value of 30.
+	//
+	// If not specified, this field defaults to a value of 30.
+	//
+	// Setting this field to zero allows for the construction of a
+	// "jail" for this priority level that is used to hold some request(s)
+	//
 	// +optional
-	NominalConcurrencyShares int32 `json:"nominalConcurrencyShares" protobuf:"varint,1,opt,name=nominalConcurrencyShares"`
+	NominalConcurrencyShares *int32 `json:"nominalConcurrencyShares" protobuf:"varint,1,opt,name=nominalConcurrencyShares"`
 
 	// `limitResponse` indicates what to do with requests that can not be executed right now
 	LimitResponse LimitResponse `json:"limitResponse,omitempty" protobuf:"bytes,2,opt,name=limitResponse"`

staging/src/k8s.io/api/flowcontrol/v1beta3/types.go

@@ -103,6 +103,20 @@ const (
 	AutoUpdateAnnotationKey = "apf.kubernetes.io/autoupdate-spec"
 )
 
+const (
+	// This annotation applies to v1beta3 only.
+	//
+	// The presence of this annotation in a v1beta3 object means that
+	// a zero value in the 'NominalConcurrencyShares' field means zero
+	// rather than the old default of 30.
+	//
+	// To set a zero value for the 'NominalConcurrencyShares' field in v1beta3,
+	// set the annotation to an empty string:
+	//   "flowcontrol.k8s.io/concurrency-shares-defaults-to-zero": ""
+	//
+	PriorityLevelConcurrencyShareDefaultKey = "flowcontrol.k8s.io/concurrency-shares-defaults-to-zero"
+)
+
 // +genclient
 // +genclient:nonNamespaced
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

staging/src/k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap/default.go

@@ -23,7 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apiserver/pkg/authentication/serviceaccount"
 	"k8s.io/apiserver/pkg/authentication/user"
-	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 // The objects that define an apiserver's initial behavior.  The
@@ -90,8 +90,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementExempt,
 			Exempt: &flowcontrol.ExemptPriorityLevelConfiguration{
-				NominalConcurrencyShares: pointer.Int32(0),
+				NominalConcurrencyShares: ptr.To(int32(0)),
-				LendablePercent:          pointer.Int32(0),
+				LendablePercent:          ptr.To(int32(0)),
 			},
 		},
 	)
@@ -100,8 +100,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 5,
+				NominalConcurrencyShares: ptr.To(int32(5)),
-				LendablePercent:          pointer.Int32(0),
+				LendablePercent:          ptr.To(int32(0)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeReject,
 				},
@@ -173,8 +173,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 30,
+				NominalConcurrencyShares: ptr.To(int32(30)),
-				LendablePercent:          pointer.Int32(33),
+				LendablePercent:          ptr.To(int32(33)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,
 					Queuing: &flowcontrol.QueuingConfiguration{
@@ -190,8 +190,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 40,
+				NominalConcurrencyShares: ptr.To(int32(40)),
-				LendablePercent:          pointer.Int32(25),
+				LendablePercent:          ptr.To(int32(25)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,
 					Queuing: &flowcontrol.QueuingConfiguration{
@@ -208,8 +208,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 10,
+				NominalConcurrencyShares: ptr.To(int32(10)),
-				LendablePercent:          pointer.Int32(0),
+				LendablePercent:          ptr.To(int32(0)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,
 					Queuing: &flowcontrol.QueuingConfiguration{
@@ -226,8 +226,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 40,
+				NominalConcurrencyShares: ptr.To(int32(40)),
-				LendablePercent:          pointer.Int32(50),
+				LendablePercent:          ptr.To(int32(50)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,
 					Queuing: &flowcontrol.QueuingConfiguration{
@@ -244,8 +244,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 100,
+				NominalConcurrencyShares: ptr.To(int32(100)),
-				LendablePercent:          pointer.Int32(90),
+				LendablePercent:          ptr.To(int32(90)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,
 					Queuing: &flowcontrol.QueuingConfiguration{
@@ -262,8 +262,8 @@ var (
 		flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 20,
+				NominalConcurrencyShares: ptr.To(int32(20)),
-				LendablePercent:          pointer.Int32(50),
+				LendablePercent:          ptr.To(int32(50)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,
 					Queuing: &flowcontrol.QueuingConfiguration{

staging/src/k8s.io/apiserver/pkg/apis/flowcontrol/bootstrap/default_test.go

@@ -89,7 +89,7 @@ func TestBootstrapPriorityLevelConfigurationWithBorrowing(t *testing.T) {
 			t.Errorf("bootstrap PriorityLevelConfiguration %q is not %q", test.name, flowcontrol.PriorityLevelEnablementLimited)
 			continue
 		}
-		if test.nominalSharesExpected != bootstrapPL.Spec.Limited.NominalConcurrencyShares {
+		if test.nominalSharesExpected != *bootstrapPL.Spec.Limited.NominalConcurrencyShares {
 			t.Errorf("bootstrap PriorityLevelConfiguration %q: expected NominalConcurrencyShares: %d, but got: %d", test.name, test.nominalSharesExpected, bootstrapPL.Spec.Limited.NominalConcurrencyShares)
 		}
 		if test.lendablePercentexpected != *bootstrapPL.Spec.Limited.LendablePercent {

staging/src/k8s.io/apiserver/pkg/features/kube_features.go

@@ -250,6 +250,14 @@ const (
 	//
 	// Allow the API server to serve consistent lists from cache
 	ConsistentListFromCache featuregate.Feature = "ConsistentListFromCache"
+
+	// owner: @tkashem
+	// beta: v1.29
+	//
+	// Allow Priority & Fairness in the API server to use a zero value for
+	// the 'nominalConcurrencyShares' field of the 'limited' section of a
+	// priority level.
+	ZeroLimitedNominalConcurrencyShares featuregate.Feature = "ZeroLimitedNominalConcurrencyShares"
 )
 
 func init() {
@@ -314,4 +322,6 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	WatchList: {Default: false, PreRelease: featuregate.Alpha},
 
 	ConsistentListFromCache: {Default: false, PreRelease: featuregate.Alpha},
+
+	ZeroLimitedNominalConcurrencyShares: {Default: false, PreRelease: featuregate.Beta},
 }

staging/src/k8s.io/apiserver/pkg/server/filters/priority-and-fairness_test.go

@@ -52,6 +52,7 @@ import (
 	"k8s.io/component-base/metrics/testutil"
 	"k8s.io/klog/v2"
 	clocktesting "k8s.io/utils/clock/testing"
+	"k8s.io/utils/ptr"
 
 	"github.com/google/go-cmp/cmp"
 )
@@ -1307,7 +1308,7 @@ func newConfiguration(fsName, plName, user string, concurrency int32, queueLengt
 		Spec: flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: concurrency,
+				NominalConcurrencyShares: ptr.To(concurrency),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type:    responseType,
 					Queuing: qcfg,

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/apf_controller.go

@@ -702,7 +702,7 @@ func (meal *cfgMeal) digestNewPLsLocked(newPLs []*flowcontrol.PriorityLevelConfi
 			state.quiescing = false
 		}
 		nominalConcurrencyShares, _, _ := plSpecCommons(state.pl)
-		meal.shareSum += float64(nominalConcurrencyShares)
+		meal.shareSum += float64(*nominalConcurrencyShares)
 		meal.haveExemptPL = meal.haveExemptPL || pl.Name == flowcontrol.PriorityLevelConfigurationNameExempt
 		meal.haveCatchAllPL = meal.haveCatchAllPL || pl.Name == flowcontrol.PriorityLevelConfigurationNameCatchAll
 	}
@@ -807,7 +807,7 @@ func (meal *cfgMeal) processOldPLsLocked() {
 		// allocation determined by all the share values in the
 		// regular way.
 		nominalConcurrencyShares, _, _ := plSpecCommons(plState.pl)
-		meal.shareSum += float64(nominalConcurrencyShares)
+		meal.shareSum += float64(*nominalConcurrencyShares)
 		meal.haveExemptPL = meal.haveExemptPL || plName == flowcontrol.PriorityLevelConfigurationNameExempt
 		meal.haveCatchAllPL = meal.haveCatchAllPL || plName == flowcontrol.PriorityLevelConfigurationNameCatchAll
 		meal.newPLStates[plName] = plState
@@ -823,7 +823,7 @@ func (meal *cfgMeal) finishQueueSetReconfigsLocked() {
 		// The use of math.Ceil here means that the results might sum
 		// to a little more than serverConcurrencyLimit but the
 		// difference will be negligible.
-		concurrencyLimit := int(math.Ceil(float64(meal.cfgCtlr.serverConcurrencyLimit) * float64(nominalConcurrencyShares) / meal.shareSum))
+		concurrencyLimit := int(math.Ceil(float64(meal.cfgCtlr.serverConcurrencyLimit) * float64(*nominalConcurrencyShares) / meal.shareSum))
 		var lendableCL, borrowingCL int
 		if lendablePercent != nil {
 			lendableCL = int(math.Round(float64(concurrencyLimit) * float64(*lendablePercent) / 100))
@@ -974,7 +974,7 @@ func (meal *cfgMeal) imaginePL(proto *flowcontrol.PriorityLevelConfiguration) {
 		seatDemandRatioedGauge: seatDemandRatioedGauge,
 	}
 	nominalConcurrencyShares, _, _ := plSpecCommons(proto)
-	meal.shareSum += float64(nominalConcurrencyShares)
+	meal.shareSum += float64(*nominalConcurrencyShares)
 }
 
 // startRequest classifies and, if appropriate, enqueues the request.
@@ -1112,7 +1112,7 @@ func relDiff(x, y float64) float64 {
 }
 
 // plSpecCommons returns the (NominalConcurrencyShares, LendablePercent, BorrowingLimitPercent) of the given priority level config
-func plSpecCommons(pl *flowcontrol.PriorityLevelConfiguration) (int32, *int32, *int32) {
+func plSpecCommons(pl *flowcontrol.PriorityLevelConfiguration) (*int32, *int32, *int32) {
 	if limiter := pl.Spec.Limited; limiter != nil {
 		return limiter.NominalConcurrencyShares, limiter.LendablePercent, limiter.BorrowingLimitPercent
 	}
@@ -1121,5 +1121,5 @@ func plSpecCommons(pl *flowcontrol.PriorityLevelConfiguration) (int32, *int32, *
 	if limiter.NominalConcurrencyShares != nil {
 		nominalConcurrencyShares = *limiter.NominalConcurrencyShares
 	}
-	return nominalConcurrencyShares, limiter.LendablePercent, nil
+	return &nominalConcurrencyShares, limiter.LendablePercent, nil
 }

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/apf_filter_test.go

@@ -33,6 +33,7 @@ import (
 	fcrequest "k8s.io/apiserver/pkg/util/flowcontrol/request"
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/ptr"
 )
 
 // TestQueueWaitTimeLatencyTracker tests the queue wait times recorded by the P&F latency tracker
@@ -80,7 +81,7 @@ func TestQueueWaitTimeLatencyTracker(t *testing.T) {
 		Spec: flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 100,
+				NominalConcurrencyShares: ptr.To(int32(100)),
 				LendablePercent:          &lendable,
 				BorrowingLimitPercent:    &borrowingLimit,
 				LimitResponse: flowcontrol.LimitResponse{

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/borrowing_test.go

@@ -36,6 +36,7 @@ import (
 	fcrequest "k8s.io/apiserver/pkg/util/flowcontrol/request"
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/ptr"
 )
 
 type borrowingTestConstraints struct {
@@ -115,7 +116,7 @@ func TestBorrowing(t *testing.T) {
 					Spec: flowcontrol.PriorityLevelConfigurationSpec{
 						Type: flowcontrol.PriorityLevelEnablementLimited,
 						Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-							NominalConcurrencyShares: 100,
+							NominalConcurrencyShares: ptr.To(int32(100)),
 							LendablePercent:          &testCase.constraints[flow].lendable,
 							BorrowingLimitPercent:    &testCase.constraints[flow].borrowing,
 							LimitResponse: flowcontrol.LimitResponse{

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/controller_test.go

@@ -41,6 +41,7 @@ import (
 	fcclient "k8s.io/client-go/kubernetes/typed/flowcontrol/v1"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
 )
 
 // Some tests print a lot of debug logs which slows down tests considerably,
@@ -357,7 +358,7 @@ func TestAPFControllerWithGracefulShutdown(t *testing.T) {
 		Spec: flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: 10,
+				NominalConcurrencyShares: ptr.To(int32(10)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeReject,
 				},

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/gen_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"k8s.io/utils/clock"
+	"k8s.io/utils/ptr"
 
 	flowcontrol "k8s.io/api/flowcontrol/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -47,7 +48,7 @@ func genPL(rng *rand.Rand, name string) *flowcontrol.PriorityLevelConfiguration
 		Spec: flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: rng.Int31n(100) + 1,
+				NominalConcurrencyShares: ptr.To(int32(rng.Int31n(100) + 1)),
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeReject}}}}
 	if rng.Float32() < 0.95 {

staging/src/k8s.io/apiserver/pkg/util/flowcontrol/max_seats_test.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apiserver/pkg/util/flowcontrol/metrics"
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/ptr"
 )
 
 // Test_GetMaxSeats tests max seats retrieved from MaxSeatsTracker
@@ -120,7 +121,7 @@ func Test_GetMaxSeats(t *testing.T) {
 				Spec: flowcontrolv1.PriorityLevelConfigurationSpec{
 					Type: flowcontrolv1.PriorityLevelEnablementLimited,
 					Limited: &flowcontrolv1.LimitedPriorityLevelConfiguration{
-						NominalConcurrencyShares: 10000,
+						NominalConcurrencyShares: ptr.To(int32(10000)),
 						LimitResponse: flowcontrolv1.LimitResponse{
 							Queuing: &flowcontrolv1.QueuingConfiguration{
 								HandSize: testcase.handSize,

test/e2e/apimachinery/flowcontrol.go

@@ -40,6 +40,7 @@ import (
 	clientsideflowcontrol "k8s.io/client-go/util/flowcontrol"
 	"k8s.io/kubernetes/test/e2e/framework"
 	admissionapi "k8s.io/pod-security-admission/api"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -260,7 +261,7 @@ func createPriorityLevel(ctx context.Context, f *framework.Framework, priorityLe
 			Spec: flowcontrol.PriorityLevelConfigurationSpec{
 				Type: flowcontrol.PriorityLevelEnablementLimited,
 				Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-					NominalConcurrencyShares: nominalConcurrencyShares,
+					NominalConcurrencyShares: ptr.To(nominalConcurrencyShares),
 					LimitResponse: flowcontrol.LimitResponse{
 						Type: flowcontrol.LimitResponseTypeReject,
 					},

test/integration/apiserver/flowcontrol/concurrency_test.go

@@ -39,6 +39,7 @@ import (
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
 	"k8s.io/kubernetes/test/integration/framework"
 	"k8s.io/kubernetes/test/utils/ktesting"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -247,7 +248,7 @@ func createPriorityLevelAndBindingFlowSchemaForUser(c clientset.Interface, usern
 		Spec: flowcontrol.PriorityLevelConfigurationSpec{
 			Type: flowcontrol.PriorityLevelEnablementLimited,
 			Limited: &flowcontrol.LimitedPriorityLevelConfiguration{
-				NominalConcurrencyShares: int32(concurrencyShares),
+				NominalConcurrencyShares: ptr.To(int32(concurrencyShares)),
 				BorrowingLimitPercent:    &i0,
 				LimitResponse: flowcontrol.LimitResponse{
 					Type: flowcontrol.LimitResponseTypeQueue,