Add option to explicitly enable future gv or gvr in runtime-config.

Signed-off-by: Siyuan Zhang <sizhang@google.com>

cmd/kube-apiserver/app/options/options_test.go

@@ -129,6 +129,7 @@ func TestAddFlags(t *testing.T) {
 		"--lease-reuse-duration-seconds=100",
 		"--emulated-version=test=1.31",
 		"--emulation-forward-compatible=true",
+		"--runtime-config-emulation-forward-compatible=true",
 	}
 	fs.Parse(args)
 	utilruntime.Must(componentGlobalsRegistry.Set())
@@ -149,6 +150,7 @@ func TestAddFlags(t *testing.T) {
 				ComponentGlobalsRegistry:                componentGlobalsRegistry,
 				ComponentName:                           basecompatibility.DefaultKubeComponent,
 				EmulationForwardCompatible:              true,
+				RuntimeConfigEmulationForwardCompatible: true,
 			},
 			Admission: &kubeoptions.AdmissionOptions{
 				GenericAdmission: &apiserveroptions.AdmissionOptions{

pkg/controlplane/apiserver/apis.go

@@ -89,7 +89,12 @@ func (s *Server) InstallAPIs(restStorageProviders ...RESTStorageProvider) error
 	nonLegacy := []*genericapiserver.APIGroupInfo{}
 
 	// used later in the loop to filter the served resource by those that have expired.
-	resourceExpirationEvaluator, err := genericapiserver.NewResourceExpirationEvaluator(s.GenericAPIServer.EffectiveVersion.EmulationVersion(), s.GenericAPIServer.EmulationForwardCompatible)
+	resourceExpirationEvaluatorOpts := genericapiserver.ResourceExpirationEvaluatorOptions{
+		CurrentVersion:                          s.GenericAPIServer.EffectiveVersion.EmulationVersion(),
+		EmulationForwardCompatible:              s.GenericAPIServer.EmulationForwardCompatible,
+		RuntimeConfigEmulationForwardCompatible: s.GenericAPIServer.RuntimeConfigEmulationForwardCompatible,
+	}
+	resourceExpirationEvaluator, err := genericapiserver.NewResourceExpirationEvaluatorFromOptions(resourceExpirationEvaluatorOpts)
 	if err != nil {
 		return err
 	}
@@ -107,11 +112,13 @@ func (s *Server) InstallAPIs(restStorageProviders ...RESTStorageProvider) error
 			continue
 		}
 
-		// Remove resources that serving kinds that are removed.
+		// Remove resources that serving kinds that are removed or not introduced yet at the current version.
 		// We do this here so that we don't accidentally serve versions without resources or openapi information that for kinds we don't serve.
 		// This is a spot above the construction of individual storage handlers so that no sig accidentally forgets to check.
-		resourceExpirationEvaluator.RemoveDeletedKinds(groupName, apiGroupInfo.Scheme, apiGroupInfo.VersionedResourcesStorageMap)
+		err = resourceExpirationEvaluator.RemoveUnavailableKinds(groupName, apiGroupInfo.Scheme, apiGroupInfo.VersionedResourcesStorageMap, s.APIResourceConfigSource)
-		resourceExpirationEvaluator.RemoveUnIntroducedKinds(groupName, apiGroupInfo.Scheme, apiGroupInfo.VersionedResourcesStorageMap)
+		if err != nil {
+			return err
+		}
 		if len(apiGroupInfo.VersionedResourcesStorageMap) == 0 {
 			klog.V(1).Infof("Removing API group %v because it is time to stop serving it because it has no versions per APILifecycle.", groupName)
 			continue

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -154,10 +154,15 @@ type Config struct {
 	// EffectiveVersion determines which apis and features are available
 	// based on when the api/feature lifecyle.
 	EffectiveVersion basecompatibility.EffectiveVersion
-	// EmulationForwardCompatible indicates APIs introduced after the emulation version are installed.
+	// EmulationForwardCompatible is an option to implicitly enable all APIs which are introduced after the emulation version and
-	// If true, APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// have higher priority than APIs of the same group resource enabled at the emulation version.
-	// This is useful if a controller has switched to use newer APIs in the binary version, and we want it still functional in an older emulation version.
+	// If true, all APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// This is needed when a controller implementation migrates to newer API versions, for the binary version, and also uses the newer API versions even when emulation version is set.
 	EmulationForwardCompatible bool
+	// RuntimeConfigEmulationForwardCompatible is an option to explicitly enable specific APIs introduced after the emulation version through the runtime-config.
+	// If true, APIs identified by group/version that are enabled in the --runtime-config flag will be installed even if it is introduced after the emulation version. --runtime-config flag values that identify multiple APIs, such as api/all,api/ga,api/beta, are not influenced by this flag and will only enable APIs available at the current emulation version.
+	// If false, error would be thrown if any GroupVersion or GroupVersionResource explicitly enabled in the --runtime-config flag is introduced after the emulation version.
+	RuntimeConfigEmulationForwardCompatible bool
 	// FeatureGate is a way to plumb feature gate through if you have them.
 	FeatureGate featuregate.FeatureGate
 	// AuditBackend is where audit events are sent to.
@@ -845,6 +850,7 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 
 		EffectiveVersion:                        c.EffectiveVersion,
 		EmulationForwardCompatible:              c.EmulationForwardCompatible,
+		RuntimeConfigEmulationForwardCompatible: c.RuntimeConfigEmulationForwardCompatible,
 		FeatureGate:                             c.FeatureGate,
 
 		muxAndDiscoveryCompleteSignals: map[string]<-chan struct{}{},

staging/src/k8s.io/apiserver/pkg/server/deleted_kinds.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	apimachineryversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apiserver/pkg/registry/rest"
+	serverstorage "k8s.io/apiserver/pkg/server/storage"
 	"k8s.io/klog/v2"
 )
 
@@ -38,6 +39,7 @@ var alphaPattern = regexp.MustCompile(`^v\d+alpha\d+$`)
 type resourceExpirationEvaluator struct {
 	currentVersion                          *apimachineryversion.Version
 	emulationForwardCompatible              bool
+	runtimeConfigEmulationForwardCompatible bool
 	isAlpha                                 bool
 	// Special flag checking for the existence of alpha.0
 	// alpha.0 is a special case where everything merged to master is auto propagated to the release-1.n branch
@@ -54,24 +56,41 @@ type resourceExpirationEvaluator struct {
 
 // ResourceExpirationEvaluator indicates whether or not a resource should be served.
 type ResourceExpirationEvaluator interface {
-	// RemoveDeletedKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted.
+	// RemoveUnavailableKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted or are introduced after the current version.
 	// versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
-	RemoveDeletedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage)
+	RemoveUnavailableKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage, apiResourceConfigSource serverstorage.APIResourceConfigSource) error
-	// RemoveUnIntroducedKinds inspects the storage map and modifies it in place by removing storage for kinds that are introduced after the current version.
-	// versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
-	RemoveUnIntroducedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage)
 	// ShouldServeForVersion returns true if a particular version cut off is after the current version
 	ShouldServeForVersion(majorRemoved, minorRemoved int) bool
 }
 
-func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version, emulationForwardCompatible bool) (ResourceExpirationEvaluator, error) {
+type ResourceExpirationEvaluatorOptions struct {
+	// CurrentVersion is the current version of the apiserver.
+	CurrentVersion *apimachineryversion.Version
+	// EmulationForwardCompatible indicates whether the apiserver should serve resources that are introduced after the current version,
+	// when resources of the same group and resource name but with lower priority are served.
+	EmulationForwardCompatible bool
+	// RuntimeConfigEmulationForwardCompatible indicates whether the apiserver should serve resources that are introduced after the current version,
+	// when the resource is explicitly enabled in runtime-config.
+	RuntimeConfigEmulationForwardCompatible bool
+}
+
+func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version) (ResourceExpirationEvaluator, error) {
+	opts := ResourceExpirationEvaluatorOptions{
+		CurrentVersion: currentVersion,
+	}
+	return NewResourceExpirationEvaluatorFromOptions(opts)
+}
+
+func NewResourceExpirationEvaluatorFromOptions(opts ResourceExpirationEvaluatorOptions) (ResourceExpirationEvaluator, error) {
+	currentVersion := opts.CurrentVersion
 	if currentVersion == nil {
 		return nil, fmt.Errorf("empty NewResourceExpirationEvaluator currentVersion")
 	}
 	klog.V(1).Infof("NewResourceExpirationEvaluator with currentVersion: %s.", currentVersion)
 	ret := &resourceExpirationEvaluator{
 		strictRemovedHandlingInAlpha:            false,
-		emulationForwardCompatible:   emulationForwardCompatible,
+		emulationForwardCompatible:              opts.EmulationForwardCompatible,
+		runtimeConfigEmulationForwardCompatible: opts.RuntimeConfigEmulationForwardCompatible,
 	}
 	// Only keeps the major and minor versions from input version.
 	ret.currentVersion = apimachineryversion.MajorMinor(currentVersion.Major(), currentVersion.Minor())
@@ -97,6 +116,7 @@ func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version,
 	return ret, nil
 }
 
+// isNotRemoved checks if a resource is removed due to the APILifecycleRemoved information.
 func (e *resourceExpirationEvaluator) isNotRemoved(gv schema.GroupVersion, versioner runtime.ObjectVersioner, resourceServingInfo rest.Storage) bool {
 	internalPtr := resourceServingInfo.New()
 
@@ -152,9 +172,9 @@ type introducedInterface interface {
 	APILifecycleIntroduced() (major, minor int)
 }
 
-// RemoveDeletedKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted.
+// removeDeletedKinds inspects the storage map and modifies it in place by removing storage for kinds that have been deleted.
 // versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
-func (e *resourceExpirationEvaluator) RemoveDeletedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage) {
+func (e *resourceExpirationEvaluator) removeDeletedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage) {
 	versionsToRemove := sets.NewString()
 	for apiVersion := range sets.StringKeySet(versionedResourcesStorageMap) {
 		versionToResource := versionedResourcesStorageMap[apiVersion]
@@ -188,32 +208,42 @@ func (e *resourceExpirationEvaluator) RemoveDeletedKinds(groupName string, versi
 	}
 }
 
-// RemoveUnIntroducedKinds inspects the storage map and modifies it in place by removing storage for kinds that are introduced after the current version.
+func (e *resourceExpirationEvaluator) RemoveUnavailableKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage, apiResourceConfigSource serverstorage.APIResourceConfigSource) error {
+	e.removeDeletedKinds(groupName, versioner, versionedResourcesStorageMap)
+	return e.removeUnintroducedKinds(groupName, versioner, versionedResourcesStorageMap, apiResourceConfigSource)
+}
+
+// removeUnintroducedKinds inspects the storage map and modifies it in place by removing storage for kinds that are introduced after the current version.
 // versionedResourcesStorageMap mirrors the field on APIGroupInfo, it's a map from version to resource to the storage.
-func (e *resourceExpirationEvaluator) RemoveUnIntroducedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage) {
+func (e *resourceExpirationEvaluator) removeUnintroducedKinds(groupName string, versioner runtime.ObjectVersioner, versionedResourcesStorageMap map[string]map[string]rest.Storage, apiResourceConfigSource serverstorage.APIResourceConfigSource) error {
 	versionsToRemove := sets.NewString()
 	prioritizedVersions := versioner.PrioritizedVersionsForGroup(groupName)
 	enabledResources := sets.NewString()
 
-	// iterate from the end to the front, so that we remove the older versions first.
+	// iterate from the end to the front, so that we remove the lower priority versions first.
 	for i := len(prioritizedVersions) - 1; i >= 0; i-- {
 		apiVersion := prioritizedVersions[i].Version
 		versionToResource := versionedResourcesStorageMap[apiVersion]
-		resourcesToRemove := sets.NewString()
+		if len(versionToResource) == 0 {
-		for resourceName, resourceServingInfo := range versionToResource {
-			// if an earlier version of the resource has been enabled, the same resource with higher priority
-			// should also be enabled if emulationForwardCompatible.
-			if e.emulationForwardCompatible && enabledResources.Has(resourceName) {
 			continue
 		}
-			verIntroduced := versionIntroduced(schema.GroupVersion{Group: groupName, Version: apiVersion}, versioner, resourceServingInfo)
+		resourcesToRemove := sets.NewString()
-			if e.currentVersion.LessThan(verIntroduced) {
+		for resourceName, resourceServingInfo := range versionToResource {
-				resourcesToRemove.Insert(resourceName)
+			// we check the resource enablement from low priority to high priority.
-			} else {
+			// If the same resource with a different version that we have checked so far is already enabled, that means some resource with the same resourceName and a lower priority version has been enabled.
-				// emulation forward compatibility is not applicable to alpha apis.
+			// Then emulation forward compatibility for the version being checked now is made based on this information.
-				if !alphaPattern.MatchString(apiVersion) {
+			lowerPriorityEnabled := enabledResources.Has(resourceName)
-					enabledResources.Insert(resourceName)
+			shouldKeep, err := e.shouldServeBasedOnVersionIntroduced(schema.GroupVersionResource{Group: groupName, Version: apiVersion, Resource: resourceName},
+				versioner, resourceServingInfo, apiResourceConfigSource, lowerPriorityEnabled)
+			if err != nil {
+				return err
 			}
+			if !shouldKeep {
+				resourcesToRemove.Insert(resourceName)
+			} else if !alphaPattern.MatchString(apiVersion) {
+				// enabledResources is passed onto the next iteration to check the enablement of higher priority resources for emulation forward compatibility.
+				// But enablement alpha apis do not affect the enablement of other versions because emulation forward compatibility is not applicable to alpha apis.
+				enabledResources.Insert(resourceName)
 			}
 		}
 
@@ -235,16 +265,24 @@ func (e *resourceExpirationEvaluator) RemoveUnIntroducedKinds(groupName string,
 	}
 
 	for _, apiVersion := range versionsToRemove.List() {
+		gv := schema.GroupVersion{Group: groupName, Version: apiVersion}
+		if apiResourceConfigSource != nil && apiResourceConfigSource.VersionExplicitlyEnabled(gv) {
+			return fmt.Errorf(
+				"cannot enable version %s in runtime-config because all the resources have been introduced after the current version %s. Consider setting --runtime-config-emulation-forward-compatible=true",
+				gv, e.currentVersion)
+		}
 		klog.V(1).Infof("Removing version %v.%v because it is introduced after the current version %s and because it has no resources per APILifecycle.", apiVersion, groupName, e.currentVersion.String())
 		delete(versionedResourcesStorageMap, apiVersion)
 	}
+	return nil
 }
 
-func versionIntroduced(gv schema.GroupVersion, versioner runtime.ObjectVersioner, resourceServingInfo rest.Storage) *apimachineryversion.Version {
+func (e *resourceExpirationEvaluator) shouldServeBasedOnVersionIntroduced(gvr schema.GroupVersionResource, versioner runtime.ObjectVersioner, resourceServingInfo rest.Storage,
-	defaultVer := apimachineryversion.MajorMinor(0, 0)
+	apiResourceConfigSource serverstorage.APIResourceConfigSource, lowerPriorityEnabled bool) (bool, error) {
+	verIntroduced := apimachineryversion.MajorMinor(0, 0)
 	internalPtr := resourceServingInfo.New()
 
-	target := gv
+	target := gvr.GroupVersion()
 	// honor storage that overrides group version (used for things like scale subresources)
 	if versionProvider, ok := resourceServingInfo.(rest.GroupVersionKindProvider); ok {
 		target = versionProvider.GroupVersionKind(target).GroupVersion()
@@ -253,15 +291,38 @@ func versionIntroduced(gv schema.GroupVersion, versioner runtime.ObjectVersioner
 	versionedPtr, err := versioner.ConvertToVersion(internalPtr, target)
 	if err != nil {
 		utilruntime.HandleError(err)
-		return defaultVer
+		return false, err
 	}
 
 	introduced, ok := versionedPtr.(introducedInterface)
 	if ok {
 		majorIntroduced, minorIntroduced := introduced.APILifecycleIntroduced()
-		return apimachineryversion.MajorMinor(uint(majorIntroduced), uint(minorIntroduced))
+		verIntroduced = apimachineryversion.MajorMinor(uint(majorIntroduced), uint(minorIntroduced))
 	}
-	return defaultVer
+	// should serve resource introduced at or before the current version.
+	if e.currentVersion.AtLeast(verIntroduced) {
+		return true, nil
+	}
+	// the rest of the function is to determine if a resource introduced after current version should be served. (only applicable in emulation mode.)
+
+	// if a lower priority version of the resource has been enabled, the same resource with higher priority
+	// should also be enabled if emulationForwardCompatible = true.
+	if e.emulationForwardCompatible && lowerPriorityEnabled {
+		return true, nil
+	}
+	if apiResourceConfigSource == nil {
+		return false, nil
+	}
+	// could explicitly enable future resources in runtime-config forward compatible mode.
+	if e.runtimeConfigEmulationForwardCompatible && (apiResourceConfigSource.ResourceExplicitlyEnabled(gvr) || apiResourceConfigSource.VersionExplicitlyEnabled(gvr.GroupVersion())) {
+		return true, nil
+	}
+	// return error if a future resource is explicit enabled in runtime-config but runtimeConfigEmulationForwardCompatible is false.
+	if apiResourceConfigSource.ResourceExplicitlyEnabled(gvr) {
+		return false, fmt.Errorf("cannot enable resource %s in runtime-config because it is introduced at %s after the current version %s. Consider setting --runtime-config-emulation-forward-compatible=true",
+			gvr, verIntroduced, e.currentVersion)
+	}
+	return false, nil
 }
 
 func shouldRemoveResourceAndSubresources(resourcesToRemove sets.String, resourceName string) bool {

staging/src/k8s.io/apiserver/pkg/server/deleted_kinds_test.go

@@ -27,6 +27,10 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	apimachineryversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apiserver/pkg/registry/rest"
+	"k8s.io/apiserver/pkg/server/resourceconfig"
+	serverstorage "k8s.io/apiserver/pkg/server/storage"
+
+	"github.com/stretchr/testify/require"
 )
 
 func Test_newResourceExpirationEvaluator(t *testing.T) {
@@ -65,7 +69,7 @@ func Test_newResourceExpirationEvaluator(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			actual, actualErr := NewResourceExpirationEvaluator(apimachineryversion.MustParse(tt.currentVersion), false)
+			actual, actualErr := NewResourceExpirationEvaluator(apimachineryversion.MustParse(tt.currentVersion))
 
 			checkErr(t, actualErr, tt.expectedErr)
 			if actualErr != nil {
@@ -289,6 +293,18 @@ func (d *dummyConvertor) PrioritizedVersionsForGroup(group string) []schema.Grou
 	return d.prioritizedVersions
 }
 
+func (d *dummyConvertor) IsGroupRegistered(group string) bool {
+	return true
+}
+
+func (d *dummyConvertor) IsVersionRegistered(v schema.GroupVersion) bool {
+	return true
+}
+
+func (d *dummyConvertor) PrioritizedVersionsAllGroups() []schema.GroupVersion {
+	return d.prioritizedVersions
+}
+
 func checkErr(t *testing.T, actual error, expected string) {
 	t.Helper()
 	switch {
@@ -370,7 +386,7 @@ func Test_removeDeletedKinds(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			convertor := &dummyConvertor{prioritizedVersions: []schema.GroupVersion{
 				{Group: groupName, Version: "v2"}, {Group: groupName, Version: "v1"}}}
-			tt.resourceExpirationEvaluator.RemoveDeletedKinds(groupName, convertor, tt.versionedResourcesStorageMap)
+			tt.resourceExpirationEvaluator.removeDeletedKinds(groupName, convertor, tt.versionedResourcesStorageMap)
 			if !reflect.DeepEqual(tt.expectedStorage, tt.versionedResourcesStorageMap) {
 				t.Fatal(dump.Pretty(tt.versionedResourcesStorageMap))
 			}
@@ -385,13 +401,19 @@ func Test_removeUnIntroducedKinds(t *testing.T) {
 	tests := []struct {
 		name                         string
 		resourceExpirationEvaluator  resourceExpirationEvaluator
+		runtimeConfig                map[string]string
+		expectErr                    bool
 		versionedResourcesStorageMap map[string]map[string]rest.Storage
 		expectedStorage              map[string]map[string]rest.Storage
 	}{
 		{
 			name: "remove-future-version",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentVersion: apimachineryversion.MajorMinor(1, 20),
+				currentVersion: apimachineryversion.MajorMinor(1, 21),
+			},
+			runtimeConfig: map[string]string{
+				"api/beta":             "true",
+				groupName + "/v2beta1": "true",
 			},
 			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
 				"v1": {
@@ -419,6 +441,9 @@ func Test_removeUnIntroducedKinds(t *testing.T) {
 				"v2alpha1": {
 					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
 				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
 			},
 		},
 		{
@@ -588,14 +613,676 @@ func Test_removeUnIntroducedKinds(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "runtime-config-enable-future-version-err",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			expectErr: true,
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-enable-future-resource-err",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2/resource2": "true",
+			},
+			expectErr: true,
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-emulation-forward-compatible-beta2-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:                          apimachineryversion.MajorMinor(1, 20),
+				runtimeConfigEmulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-emulation-forward-compatible-beta2-resource",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:                          apimachineryversion.MajorMinor(1, 20),
+				runtimeConfigEmulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2/resource2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-runtime-config-beta2-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 20),
+				emulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "both-runtime-config-and-emulation-forward-compatible-runtime-config-beta2-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:                          apimachineryversion.MajorMinor(1, 20),
+				emulationForwardCompatible:              true,
+				runtimeConfigEmulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			resourceConfig := serverstorage.NewResourceConfig()
 			convertor := &dummyConvertor{prioritizedVersions: []schema.GroupVersion{
 				{Group: groupName, Version: "v2"}, {Group: groupName, Version: "v1"},
 				{Group: groupName, Version: "v2beta2"}, {Group: groupName, Version: "v2beta1"},
 				{Group: groupName, Version: "v2alpha1"}}}
-			tt.resourceExpirationEvaluator.RemoveUnIntroducedKinds(groupName, convertor, tt.versionedResourcesStorageMap)
+			resourceConfig.EnableVersions(convertor.PrioritizedVersionsForGroup(groupName)...)
+			resourceConfig, err := resourceconfig.MergeAPIResourceConfigs(resourceConfig, tt.runtimeConfig, convertor)
+			require.NoError(t, err)
+			err = tt.resourceExpirationEvaluator.removeUnintroducedKinds(groupName, convertor, tt.versionedResourcesStorageMap, resourceConfig)
+			if tt.expectErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			if !reflect.DeepEqual(tt.expectedStorage, tt.versionedResourcesStorageMap) {
+				t.Fatal(dump.Pretty(tt.versionedResourcesStorageMap))
+			}
+		})
+	}
+}
+
+func Test_RemoveUnavailableKinds(t *testing.T) {
+	groupName := "group.name"
+	resource1 := "resource1"
+	resource2 := "resource2"
+	tests := []struct {
+		name                         string
+		resourceExpirationEvaluator  resourceExpirationEvaluator
+		runtimeConfig                map[string]string
+		expectErr                    bool
+		versionedResourcesStorageMap map[string]map[string]rest.Storage
+		expectedStorage              map[string]map[string]rest.Storage
+	}{
+		{
+			name: "remove-future-version",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 21),
+			},
+			runtimeConfig: map[string]string{
+				"api/beta":             "true",
+				groupName + "/v2beta1": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+			},
+		},
+		{
+			name: "missing-introduced-version",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageRemovedIn(1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageRemovedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-ga-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 19),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-alpha-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 20),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 18),
+				},
+				"v2alpha1": {
+					resource1: storageIntroducedAndRemovedIn(1, 20, 1, 21),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-beta1-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 21),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-new-resource",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 22),
+				emulationForwardCompatible: true,
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-enable-future-version-err",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			expectErr: true,
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-enable-future-resource-err",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2/resource2": "true",
+			},
+			expectErr: true,
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-emulation-forward-compatible-beta2-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:                          apimachineryversion.MajorMinor(1, 20),
+				runtimeConfigEmulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+			},
+		},
+		{
+			name: "runtime-config-emulation-forward-compatible-beta2-resource",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:                          apimachineryversion.MajorMinor(1, 20),
+				runtimeConfigEmulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2/resource2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-runtime-config-beta2-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 20),
+				emulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "both-runtime-config-and-emulation-forward-compatible-runtime-config-beta2-api",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:                          apimachineryversion.MajorMinor(1, 20),
+				emulationForwardCompatible:              true,
+				runtimeConfigEmulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+		{
+			name: "emulation-forward-compatible-runtime-config-beta2-api-resource1-ok",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion:             apimachineryversion.MajorMinor(1, 21),
+				emulationForwardCompatible: true,
+			},
+			runtimeConfig: map[string]string{
+				groupName + "/v2beta2": "true",
+			},
+			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+					resource2: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+					resource2: storageIntroducedIn(1, 23),
+				},
+			},
+			expectedStorage: map[string]map[string]rest.Storage{
+				"v1": {
+					resource1: storageIntroducedIn(1, 20),
+				},
+				"v2beta1": {
+					resource1: storageIntroducedAndRemovedIn(1, 21, 1, 22),
+				},
+				"v2beta2": {
+					resource1: storageIntroducedAndRemovedIn(1, 22, 1, 23),
+				},
+				"v2": {
+					resource1: storageIntroducedIn(1, 23),
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resourceConfig := serverstorage.NewResourceConfig()
+			convertor := &dummyConvertor{prioritizedVersions: []schema.GroupVersion{
+				{Group: groupName, Version: "v2"}, {Group: groupName, Version: "v1"},
+				{Group: groupName, Version: "v2beta2"}, {Group: groupName, Version: "v2beta1"},
+				{Group: groupName, Version: "v2alpha1"}}}
+			resourceConfig.EnableVersions(convertor.PrioritizedVersionsForGroup(groupName)...)
+			resourceConfig, err := resourceconfig.MergeAPIResourceConfigs(resourceConfig, tt.runtimeConfig, convertor)
+			require.NoError(t, err)
+			err = tt.resourceExpirationEvaluator.RemoveUnavailableKinds(groupName, convertor, tt.versionedResourcesStorageMap, resourceConfig)
+			if tt.expectErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
 			if !reflect.DeepEqual(tt.expectedStorage, tt.versionedResourcesStorageMap) {
 				t.Fatal(dump.Pretty(tt.versionedResourcesStorageMap))
 			}

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

@@ -245,10 +245,15 @@ type GenericAPIServer struct {
 	// EffectiveVersion determines which apis and features are available
 	// based on when the api/feature lifecyle.
 	EffectiveVersion basecompatibility.EffectiveVersion
-	// EmulationForwardCompatible indicates APIs introduced after the emulation version are installed.
+	// EmulationForwardCompatible is an option to implicitly enable all APIs which are introduced after the emulation version and
-	// If true, APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// have higher priority than APIs of the same group resource enabled at the emulation version.
-	// This is useful if a controller has switched to use newer APIs in the binary version, and we want it still functional in an older emulation version.
+	// If true, all APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// This is needed when a controller implementation migrates to newer API versions, for the binary version, and also uses the newer API versions even when emulation version is set.
 	EmulationForwardCompatible bool
+	// RuntimeConfigEmulationForwardCompatible is an option to explicitly enable specific APIs introduced after the emulation version through the runtime-config.
+	// If true, APIs identified by group/version that are enabled in the --runtime-config flag will be installed even if it is introduced after the emulation version. --runtime-config flag values that identify multiple APIs, such as api/all,api/ga,api/beta, are not influenced by this flag and will only enable APIs available at the current emulation version.
+	// If false, error would be thrown if any GroupVersion or GroupVersionResource explicitly enabled in the --runtime-config flag is introduced after the emulation version.
+	RuntimeConfigEmulationForwardCompatible bool
 
 	// FeatureGate is a way to plumb feature gate through if you have them.
 	FeatureGate featuregate.FeatureGate

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options.go

@@ -98,10 +98,15 @@ type ServerRunOptions struct {
 	ComponentGlobalsRegistry basecompatibility.ComponentGlobalsRegistry
 	// ComponentName is name under which the server's global variabled are registered in the ComponentGlobalsRegistry.
 	ComponentName string
-	// EmulationForwardCompatible indicates APIs introduced after the emulation version are installed.
+	// EmulationForwardCompatible is an option to implicitly enable all APIs which are introduced after the emulation version and
-	// If true, APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// have higher priority than APIs of the same group resource enabled at the emulation version.
-	// This is useful if a controller has switched to use newer APIs in the binary version, and we want it still functional in an older emulation version.
+	// If true, all APIs that have higher priority than the APIs of the same group resource enabled at the emulation version will be installed.
+	// This is needed when a controller implementation migrates to newer API versions, for the binary version, and also uses the newer API versions even when emulation version is set.
 	EmulationForwardCompatible bool
+	// RuntimeConfigEmulationForwardCompatible is an option to explicitly enable specific APIs introduced after the emulation version through the runtime-config.
+	// If true, APIs identified by group/version that are enabled in the --runtime-config flag will be installed even if it is introduced after the emulation version. --runtime-config flag values that identify multiple APIs, such as api/all,api/ga,api/beta, are not influenced by this flag and will only enable APIs available at the current emulation version.
+	// If false, error would be thrown if any GroupVersion or GroupVersionResource explicitly enabled in the --runtime-config flag is introduced after the emulation version.
+	RuntimeConfigEmulationForwardCompatible bool
 }
 
 func NewServerRunOptions() *ServerRunOptions {
@@ -157,6 +162,7 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.EffectiveVersion = s.ComponentGlobalsRegistry.EffectiveVersionFor(s.ComponentName)
 	c.FeatureGate = s.ComponentGlobalsRegistry.FeatureGateFor(s.ComponentName)
 	c.EmulationForwardCompatible = s.EmulationForwardCompatible
+	c.RuntimeConfigEmulationForwardCompatible = s.RuntimeConfigEmulationForwardCompatible
 
 	return nil
 }
@@ -236,11 +242,16 @@ func (s *ServerRunOptions) Validate() []error {
 	if errs := s.ComponentGlobalsRegistry.Validate(); len(errs) != 0 {
 		errors = append(errors, errs...)
 	}
-	if s.EmulationForwardCompatible {
 	effectiveVersion := s.ComponentGlobalsRegistry.EffectiveVersionFor(s.ComponentName)
-		if effectiveVersion.BinaryVersion().WithPatch(0).EqualTo(effectiveVersion.EmulationVersion()) {
+	if effectiveVersion == nil {
+		return errors
+	}
+	notEmulationMode := effectiveVersion.BinaryVersion().WithPatch(0).EqualTo(effectiveVersion.EmulationVersion())
+	if notEmulationMode && s.EmulationForwardCompatible {
 		errors = append(errors, fmt.Errorf("ServerRunOptions.EmulationForwardCompatible cannot be set to true if the emulation version is the same as the binary version"))
 	}
+	if notEmulationMode && s.RuntimeConfigEmulationForwardCompatible {
+		errors = append(errors, fmt.Errorf("ServerRunOptions.RuntimeConfigEmulationForwardCompatible cannot be set to true if the emulation version is the same as the binary version"))
 	}
 	return errors
 }
@@ -388,7 +399,10 @@ func (s *ServerRunOptions) AddUniversalFlags(fs *pflag.FlagSet) {
 
 	s.ComponentGlobalsRegistry.AddFlags(fs)
 	fs.BoolVar(&s.EmulationForwardCompatible, "emulation-forward-compatible", s.EmulationForwardCompatible, ""+
-		"If true APIs that have higher priority than the APIs enabled at the emulation version of the same group resource will be installed. "+
+		"If true all APIs that have higher priority than the APIs enabled at the emulation version of the same group resource will be installed. "+
+		"Can only be set to true if the emulation version is lower than the binary version.")
+	fs.BoolVar(&s.RuntimeConfigEmulationForwardCompatible, "runtime-config-emulation-forward-compatible", s.RuntimeConfigEmulationForwardCompatible, ""+
+		"If true, APIs identified by group/version that are enabled in the --runtime-config flag will be installed even if it is introduced after the emulation version. "+
 		"Can only be set to true if the emulation version is lower than the binary version.")
 }
 

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options_test.go

@@ -221,6 +221,24 @@ func TestServerRunOptionsValidate(t *testing.T) {
 			},
 			expectErr: "ServerRunOptions.EmulationForwardCompatible cannot be set to true if the emulation version is the same as the binary version",
 		},
+		{
+			name: "Test RuntimeConfigEmulationForwardCompatible cannot be true if not in emulation mode",
+			testOptions: &ServerRunOptions{
+				AdvertiseAddress:                        netutils.ParseIPSloppy("192.168.10.10"),
+				CorsAllowedOriginList:                   []string{"^10.10.10.100$", "^10.10.10.200$"},
+				HSTSDirectives:                          []string{"max-age=31536000", "includeSubDomains", "preload"},
+				MaxRequestsInFlight:                     400,
+				MaxMutatingRequestsInFlight:             200,
+				RequestTimeout:                          time.Duration(2) * time.Minute,
+				MinRequestTimeout:                       1800,
+				JSONPatchMaxCopyBytes:                   10 * 1024 * 1024,
+				MaxRequestBodyBytes:                     10 * 1024 * 1024,
+				ComponentGlobalsRegistry:                newTestRegistry(testComponent),
+				ComponentName:                           testComponent,
+				RuntimeConfigEmulationForwardCompatible: true,
+			},
+			expectErr: "ServerRunOptions.RuntimeConfigEmulationForwardCompatible cannot be set to true if the emulation version is the same as the binary version",
+		},
 		{
 			name: "Test EmulationForwardCompatible can be true if in emulation mode",
 			testOptions: &ServerRunOptions{
@@ -236,6 +254,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				ComponentGlobalsRegistry:                newTestRegistry(testComponent),
 				ComponentName:                           testComponent,
 				EmulationForwardCompatible:              true,
+				RuntimeConfigEmulationForwardCompatible: true,
 			},
 			emulationVersion: "1.34",
 		},

staging/src/k8s.io/apiserver/pkg/server/resourceconfig/helpers.go

@@ -182,11 +182,11 @@ func applyVersionAndResourcePreferences(
 	for _, versionPreference := range versionPreferences {
 		if versionPreference.enabled {
 			// enable the groupVersion for "group/version=true"
-			resourceConfig.EnableVersions(versionPreference.groupVersion)
+			resourceConfig.ExplicitlyEnableVersions(versionPreference.groupVersion)
 
 		} else {
 			// disable the groupVersion only for "group/version=false"
-			resourceConfig.DisableVersions(versionPreference.groupVersion)
+			resourceConfig.ExplicitlyDisableVersions(versionPreference.groupVersion)
 		}
 	}
 
@@ -194,9 +194,9 @@ func applyVersionAndResourcePreferences(
 	for _, resourcePreference := range resourcePreferences {
 		if resourcePreference.enabled {
 			// enable the resource for "group/version/resource=true"
-			resourceConfig.EnableResources(resourcePreference.groupVersionResource)
+			resourceConfig.ExplicitlyEnableResources(resourcePreference.groupVersionResource)
 		} else {
-			resourceConfig.DisableResources(resourcePreference.groupVersionResource)
+			resourceConfig.ExplicitlyDisableResources(resourcePreference.groupVersionResource)
 		}
 	}
 	return nil
@@ -253,10 +253,11 @@ func EmulationForwardCompatibleResourceConfig(
 		if !enabled {
 			continue
 		}
-		// EmulationForwardCompatibility is not applicable to alpha apis.
+		// emulation forward compatibility is not applicable to alpha apis.
 		if alphaPattern.MatchString(gv.Version) {
 			continue
 		}
+		// if a gv is enabled, all the versions with higher priority (all the versions before gv in PrioritizedVersionsForGroup) are also implicitly enabled for emulation forward compatibility.
 		for _, pgv := range registry.PrioritizedVersionsForGroup(gv.Group) {
 			if pgv.Version == gv.Version {
 				break
@@ -269,10 +270,11 @@ func EmulationForwardCompatibleResourceConfig(
 		if !enabled {
 			continue
 		}
-		// EmulationForwardCompatibility is not applicable to alpha apis.
+		// emulation forward compatibility is not applicable to alpha apis.
 		if alphaPattern.MatchString(gvr.Version) {
 			continue
 		}
+		// if a gvr is enabled, all the versions with the same resource name and higher priority (all the versions before gv in PrioritizedVersionsForGroup) are also implicitly enabled for emulation forward compatibility.
 		for _, pgv := range registry.PrioritizedVersionsForGroup(gvr.Group) {
 			if pgv.Version == gvr.Version {
 				break

staging/src/k8s.io/apiserver/pkg/server/resourceconfig/helpers_test.go

@@ -51,7 +51,9 @@ func TestParseRuntimeConfig(t *testing.T) {
 				return newFakeAPIResourceConfigSource()
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
-				return newFakeAPIResourceConfigSource()
+				config := newFakeAPIResourceConfigSource()
+				config.ExplicitlyEnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				return config
 			},
 			expectedEnabledAPIs: defaultFakeEnabledResources(),
 			err:                 true,
@@ -102,6 +104,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
+				config.ExplicitlyEnableVersions(appsv1.SchemeGroupVersion)
 				return config
 			},
 			expectedEnabledAPIs: defaultFakeEnabledResources(),
@@ -117,7 +120,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableVersions(apiv1GroupVersion)
+				config.ExplicitlyDisableVersions(apiv1GroupVersion)
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -178,6 +181,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
+				config.ExplicitlyEnableVersions(apiv1GroupVersion)
 				config.DisableVersions(appsv1.SchemeGroupVersion)
 				config.DisableVersions(extensionsapiv1beta1.SchemeGroupVersion)
 				return config
@@ -202,7 +206,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.EnableResources(extensionsapiv1beta1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyEnableResources(extensionsapiv1beta1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -224,7 +228,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableResources(extensionsapiv1beta1.SchemeGroupVersion.WithResource("ingresses"))
+				config.ExplicitlyDisableResources(extensionsapiv1beta1.SchemeGroupVersion.WithResource("ingresses"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -246,7 +250,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableVersions(extensionsapiv1beta1.SchemeGroupVersion)
+				config.ExplicitlyDisableVersions(extensionsapiv1beta1.SchemeGroupVersion)
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -268,7 +272,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyDisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -315,7 +319,8 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyEnableVersions(appsv1.SchemeGroupVersion)
+				config.ExplicitlyDisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -339,8 +344,8 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableVersions(appsv1.SchemeGroupVersion)
+				config.ExplicitlyDisableVersions(appsv1.SchemeGroupVersion)
-				config.EnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyEnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -365,7 +370,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyDisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -391,7 +396,7 @@ func TestParseRuntimeConfig(t *testing.T) {
 				config := newFakeAPIResourceConfigSource()
 				config.DisableVersions(apiv1.SchemeGroupVersion)
 				config.DisableVersions(appsv1.SchemeGroupVersion)
-				config.EnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyEnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -417,8 +422,8 @@ func TestParseRuntimeConfig(t *testing.T) {
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
 				config.DisableVersions(apiv1.SchemeGroupVersion)
-				config.EnableVersions(appsv1.SchemeGroupVersion)
+				config.ExplicitlyEnableVersions(appsv1.SchemeGroupVersion)
-				config.DisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyDisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -445,8 +450,8 @@ func TestParseRuntimeConfig(t *testing.T) {
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
 				config.DisableVersions(apiv1.SchemeGroupVersion)
-				config.DisableVersions(appsv1.SchemeGroupVersion)
+				config.ExplicitlyDisableVersions(appsv1.SchemeGroupVersion)
-				config.EnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyEnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -471,8 +476,8 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.EnableVersions(appsv1.SchemeGroupVersion)
+				config.ExplicitlyEnableVersions(appsv1.SchemeGroupVersion)
-				config.DisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyDisableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -497,8 +502,8 @@ func TestParseRuntimeConfig(t *testing.T) {
 			},
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := newFakeAPIResourceConfigSource()
-				config.DisableVersions(appsv1.SchemeGroupVersion)
+				config.ExplicitlyDisableVersions(appsv1.SchemeGroupVersion)
-				config.EnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
+				config.ExplicitlyEnableResources(appsv1.SchemeGroupVersion.WithResource("deployments"))
 				return config
 			},
 			expectedEnabledAPIs: map[schema.GroupVersionResource]bool{
@@ -638,7 +643,7 @@ func TestEmulationForwardCompatibleResourceConfig(t *testing.T) {
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := serverstore.NewResourceConfig()
 				config.EnableVersions(v2beta1, v1, v2)
-				config.DisableVersions(v2beta2)
+				config.ExplicitlyDisableVersions(v2beta2)
 				return config
 			},
 			err: false,
@@ -701,7 +706,7 @@ func TestEmulationForwardCompatibleResourceConfig(t *testing.T) {
 			expectedAPIConfig: func() *serverstore.ResourceConfig {
 				config := serverstore.NewResourceConfig()
 				config.EnableResources(v2beta1.WithResource("testtype1"), v1.WithResource("testtype1"), v2.WithResource("testtype1"))
-				config.DisableResources(v2beta2.WithResource("testtype1"))
+				config.ExplicitlyDisableResources(v2beta2.WithResource("testtype1"))
 				return config
 			},
 			err: false,

staging/src/k8s.io/apiserver/pkg/server/storage/resource_config.go

@@ -24,6 +24,8 @@ import (
 type APIResourceConfigSource interface {
 	ResourceEnabled(resource schema.GroupVersionResource) bool
 	AnyResourceForGroupEnabled(group string) bool
+	ResourceExplicitlyEnabled(resource schema.GroupVersionResource) bool
+	VersionExplicitlyEnabled(version schema.GroupVersion) bool
 }
 
 var _ APIResourceConfigSource = &ResourceConfig{}
@@ -31,10 +33,17 @@ var _ APIResourceConfigSource = &ResourceConfig{}
 type ResourceConfig struct {
 	GroupVersionConfigs         map[schema.GroupVersion]bool
 	ResourceConfigs             map[schema.GroupVersionResource]bool
+	ExplicitGroupVersionConfigs map[schema.GroupVersion]bool
+	ExplicitResourceConfigs     map[schema.GroupVersionResource]bool
 }
 
 func NewResourceConfig() *ResourceConfig {
-	return &ResourceConfig{GroupVersionConfigs: map[schema.GroupVersion]bool{}, ResourceConfigs: map[schema.GroupVersionResource]bool{}}
+	return &ResourceConfig{
+		GroupVersionConfigs:         map[schema.GroupVersion]bool{},
+		ResourceConfigs:             map[schema.GroupVersionResource]bool{},
+		ExplicitGroupVersionConfigs: map[schema.GroupVersion]bool{},
+		ExplicitResourceConfigs:     map[schema.GroupVersionResource]bool{},
+	}
 }
 
 // DisableMatchingVersions disables all group/versions for which the matcher function returns true.
@@ -77,6 +86,7 @@ func (o *ResourceConfig) removeMatchingResourcePreferences(matcher func(gvr sche
 	}
 	for _, k := range keysToRemove {
 		delete(o.ResourceConfigs, k)
+		delete(o.ExplicitResourceConfigs, k)
 	}
 }
 
@@ -91,6 +101,13 @@ func (o *ResourceConfig) DisableVersions(versions ...schema.GroupVersion) {
 	}
 }
 
+func (o *ResourceConfig) ExplicitlyDisableVersions(versions ...schema.GroupVersion) {
+	for _, version := range versions {
+		o.ExplicitGroupVersionConfigs[version] = false
+	}
+	o.DisableVersions(versions...)
+}
+
 // EnableVersions enables all resources in a given groupVersion.
 // This will remove any preferences previously set on individual resources.
 func (o *ResourceConfig) EnableVersions(versions ...schema.GroupVersion) {
@@ -103,10 +120,16 @@ func (o *ResourceConfig) EnableVersions(versions ...schema.GroupVersion) {
 
 }
 
+func (o *ResourceConfig) ExplicitlyEnableVersions(versions ...schema.GroupVersion) {
+	for _, version := range versions {
+		o.ExplicitGroupVersionConfigs[version] = true
+	}
+	o.EnableVersions(versions...)
+}
+
 // TODO this must be removed and we enable/disable individual resources.
 func (o *ResourceConfig) versionEnabled(version schema.GroupVersion) bool {
-	enabled, _ := o.GroupVersionConfigs[version]
+	return o.GroupVersionConfigs[version]
-	return enabled
 }
 
 func (o *ResourceConfig) DisableResources(resources ...schema.GroupVersionResource) {
@@ -115,12 +138,26 @@ func (o *ResourceConfig) DisableResources(resources ...schema.GroupVersionResour
 	}
 }
 
+func (o *ResourceConfig) ExplicitlyDisableResources(resources ...schema.GroupVersionResource) {
+	for _, resource := range resources {
+		o.ExplicitResourceConfigs[resource] = false
+	}
+	o.DisableResources(resources...)
+}
+
 func (o *ResourceConfig) EnableResources(resources ...schema.GroupVersionResource) {
 	for _, resource := range resources {
 		o.ResourceConfigs[resource] = true
 	}
 }
 
+func (o *ResourceConfig) ExplicitlyEnableResources(resources ...schema.GroupVersionResource) {
+	for _, resource := range resources {
+		o.ExplicitResourceConfigs[resource] = true
+	}
+	o.EnableResources(resources...)
+}
+
 func (o *ResourceConfig) ResourceEnabled(resource schema.GroupVersionResource) bool {
 	// if a resource is explicitly set, that takes priority over the preference of the version.
 	resourceEnabled, explicitlySet := o.ResourceConfigs[resource]
@@ -151,3 +188,19 @@ func (o *ResourceConfig) AnyResourceForGroupEnabled(group string) bool {
 
 	return false
 }
+
+func (o *ResourceConfig) ResourceExplicitlyEnabled(resource schema.GroupVersionResource) bool {
+	resourceEnabled, explicitlySet := o.ExplicitResourceConfigs[resource]
+	if explicitlySet {
+		return resourceEnabled
+	}
+	return false
+}
+
+func (o *ResourceConfig) VersionExplicitlyEnabled(version schema.GroupVersion) bool {
+	versionEnabled, explicitlySet := o.ExplicitGroupVersionConfigs[version]
+	if explicitlySet {
+		return versionEnabled
+	}
+	return false
+}

test/integration/apiserver/apiserver_test.go

@@ -260,7 +260,7 @@ func TestCacheControl(t *testing.T) {
 	}
 	for _, path := range paths {
 		t.Run(path, func(t *testing.T) {
-			req, err := http.NewRequest("GET", server.ClientConfig.Host+path, nil)
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+path, nil)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -306,7 +306,7 @@ func TestHSTS(t *testing.T) {
 	}
 	for _, path := range paths {
 		t.Run(path, func(t *testing.T) {
-			req, err := http.NewRequest("GET", server.ClientConfig.Host+path, nil)
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+path, nil)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -3320,7 +3320,7 @@ func TestAllowedEmulationVersions(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			req, err := http.NewRequest("GET", server.ClientConfig.Host+"/", nil)
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+"/", nil)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -3383,7 +3383,7 @@ func TestEnableEmulationVersion(t *testing.T) {
 
 	for _, tc := range tcs {
 		t.Run(tc.path, func(t *testing.T) {
-			req, err := http.NewRequest("GET", server.ClientConfig.Host+tc.path, nil)
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+tc.path, nil)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -3405,7 +3405,7 @@ func TestEnableEmulationVersionForwardCompatible(t *testing.T) {
 	featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.33"))
 	server := kubeapiservertesting.StartTestServerOrDie(t,
 		&kubeapiservertesting.TestServerInstanceOptions{BinaryVersion: "1.33"},
-		[]string{"--emulated-version=kube=1.31", "--emulation-forward-compatible=true"}, framework.SharedEtcd())
+		[]string{"--emulated-version=kube=1.31", "--runtime-config=api/beta=true", "--emulation-forward-compatible=true"}, framework.SharedEtcd())
 	defer server.TearDownFn()
 
 	rt, err := restclient.TransportFor(server.ClientConfig)
@@ -3445,7 +3445,69 @@ func TestEnableEmulationVersionForwardCompatible(t *testing.T) {
 
 	for _, tc := range tcs {
 		t.Run(tc.path, func(t *testing.T) {
-			req, err := http.NewRequest("GET", server.ClientConfig.Host+tc.path, nil)
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+tc.path, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			resp, err := rt.RoundTrip(req)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if resp.StatusCode != tc.expectedStatusCode {
+				t.Errorf("expect status code: %d, got : %d\n", tc.expectedStatusCode, resp.StatusCode)
+			}
+			defer func() {
+				_ = resp.Body.Close()
+			}()
+		})
+	}
+}
+
+func TestEnableRuntimeConfigEmulationVersionForwardCompatible(t *testing.T) {
+	featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, utilfeature.DefaultFeatureGate, version.MustParse("1.33"))
+	server := kubeapiservertesting.StartTestServerOrDie(t,
+		&kubeapiservertesting.TestServerInstanceOptions{BinaryVersion: "1.33"},
+		[]string{"--emulated-version=kube=1.31", "--runtime-config-emulation-forward-compatible=true", "--runtime-config=api/beta=true,networking.k8s.io/v1=true"}, framework.SharedEtcd())
+	defer server.TearDownFn()
+
+	rt, err := restclient.TransportFor(server.ClientConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tcs := []struct {
+		path               string
+		expectedStatusCode int
+	}{
+		{
+			path:               "/",
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/apps/v1/deployments",
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/flowcontrol.apiserver.k8s.io/v1/flowschemas",
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/flowcontrol.apiserver.k8s.io/v1beta3/flowschemas", // introduced at 1.26, removed at 1.32
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/networking.k8s.io/v1beta1/servicecidrs", // introduced at 1.31, removed at 1.34
+			expectedStatusCode: 200,
+		},
+		{
+			path:               "/apis/networking.k8s.io/v1/servicecidrs", // introduced at 1.33
+			expectedStatusCode: 200,
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.path, func(t *testing.T) {
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+tc.path, nil)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -3499,7 +3561,7 @@ func TestDisableEmulationVersion(t *testing.T) {
 
 	for _, tc := range tcs {
 		t.Run(tc.path, func(t *testing.T) {
-			req, err := http.NewRequest("GET", server.ClientConfig.Host+tc.path, nil)
+			req, err := http.NewRequest(http.MethodGet, server.ClientConfig.Host+tc.path, nil)
 			if err != nil {
 				t.Fatal(err)
 			}

test/integration/etcd/data.go

@@ -278,12 +278,14 @@ func GetEtcdStorageDataForNamespaceServedAt(namespace string, v string, removeAl
 			ExpectedEtcdPath:                       "/registry/ipaddresses/192.168.2.3",
 			ExpectedGVK:                            gvkP("networking.k8s.io", "v1beta1", "IPAddress"),
 			IntroducedVersion:                      "1.33",
+			EmulationForwardCompatibleSinceVersion: "1.31",
 		},
 		gvr("networking.k8s.io", "v1", "servicecidrs"): {
 			Stub:                                   `{"metadata": {"name": "range-b2"}, "spec": {"cidrs": ["192.168.0.0/16","fd00:1::/120"]}}`,
 			ExpectedEtcdPath:                       "/registry/servicecidrs/range-b2",
 			ExpectedGVK:                            gvkP("networking.k8s.io", "v1beta1", "ServiceCIDR"),
 			IntroducedVersion:                      "1.33",
+			EmulationForwardCompatibleSinceVersion: "1.31",
 		},
 		// --
 
@@ -623,13 +625,16 @@ func GetEtcdStorageDataForNamespaceServedAt(namespace string, v string, removeAl
 
 	// Delete types no longer served or not yet added at a particular emulated version.
 	for key, data := range etcdStorageData {
-		if data.IntroducedVersion != "" && version.MustParse(data.IntroducedVersion).GreaterThan(version.MustParse(v)) {
-			delete(etcdStorageData, key)
-		}
-
 		if data.RemovedVersion != "" && version.MustParse(v).AtLeast(version.MustParse(data.RemovedVersion)) {
 			delete(etcdStorageData, key)
 		}
+		if data.IntroducedVersion == "" || version.MustParse(v).AtLeast(version.MustParse(data.IntroducedVersion)) {
+			continue
+		}
+		if data.EmulationForwardCompatibleSinceVersion != "" && version.MustParse(v).AtLeast(version.MustParse(data.EmulationForwardCompatibleSinceVersion)) {
+			continue
+		}
+		delete(etcdStorageData, key)
 	}
 
 	if removeAlphas {
@@ -673,6 +678,10 @@ type StorageData struct {
 	ExpectedGVK       *schema.GroupVersionKind // The GVK that we expect this object to be stored as - leave this nil to use the default
 	IntroducedVersion string                   // The version that this type is introduced
 	RemovedVersion    string                   // The version that this type is removed. May be empty for stable resources
+	// EmulationForwardCompatibleSinceVersion indicates the api should be kept if the emulation version is >= this version, even when the api is introduced after the emulation version.
+	// Only needed for some Beta (Beta2+) and GA APIs, and is the version when the lowest Beta api is introduced.
+	// This is needed to enable some Beta features where the api used in the corresponding controller has GAed.
+	EmulationForwardCompatibleSinceVersion string
 }
 
 // Prerequisite contains information required to create a resource (but not verify it)

test/integration/etcd/etcd_storage_path_test.go

@@ -45,7 +45,6 @@ import (
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	componentbaseversion "k8s.io/component-base/version"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
-	"k8s.io/kubernetes/pkg/features"
 )
 
 // Only add kinds to this list when this a virtual resource with get and create verbs that doesn't actually
@@ -95,19 +94,12 @@ func testEtcdStoragePathWithVersion(t *testing.T, v string) {
 		// Only test for beta and GA APIs with emulated version.
 		featuregatetesting.SetFeatureGateEmulationVersionDuringTest(t, feature.DefaultFeatureGate, version.MustParse(v))
 		featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, "AllBeta", true)
-		// Feature Gates that are GA and depend directly on the API version to work can not be emulated in previous versions.
-		// Example feature:
-		// v1.x-2 : FeatureGate alpha , API v1alpha1/feature
-		// v1.x-1 : FeatureGate beta  , API v1beta1/feature
-		// v1.x   : FeatureGate GA    , API v1/feature
-		// The code in v1.x uses the clients with the v1 API, if we emulate v1.x-1 it will not work against apiserver that
-		// only understand v1beta1.
-		featuregatetesting.SetFeatureGateDuringTest(t, feature.DefaultFeatureGate, features.MultiCIDRServiceAllocator, false)
 	}
 
 	apiServer := StartRealAPIServerOrDie(t, func(opts *options.ServerRunOptions) {
 		// Disable alphas when emulating previous versions.
 		if v != componentbaseversion.DefaultKubeBinaryVersion {
+			opts.Options.GenericServerRunOptions.EmulationForwardCompatible = true
 			opts.Options.APIEnablement.RuntimeConfig["api/alpha"] = "false"
 		}
 	})