diff --git a/go.mod b/go.mod
index e9b6814d5..8861b47cc 100644
--- a/go.mod
+++ b/go.mod
@@ -45,9 +45,9 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 	gopkg.in/square/go-jose.v2 v2.6.0
 	k8s.io/api v0.0.0-20240626062052-149781fc54f5
-	k8s.io/apimachinery v0.0.0-20240626061445-a05248b07a6e
+	k8s.io/apimachinery v0.0.0-20240626061446-c225984b7bed
 	k8s.io/client-go v0.0.0-20240626062855-8ffa5314741e
-	k8s.io/component-base v0.0.0-20240626064639-1f2e30104e8a
+	k8s.io/component-base v0.0.0-20240626064641-0eb10f703efe
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kms v0.0.0-20240626065322-b47e46c9b25f
 	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340
diff --git a/go.sum b/go.sum
index d1de610ee..653ba160a 100644
--- a/go.sum
+++ b/go.sum
@@ -372,12 +372,12 @@ honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 k8s.io/api v0.0.0-20240626062052-149781fc54f5 h1:yqS/8fRAivOgCF1sgfIwxp1A/S2VrPpudN0V3RfHj6M=
 k8s.io/api v0.0.0-20240626062052-149781fc54f5/go.mod h1:WzXtjaoUCXrzbvJtK/lCWCih7nCdOFqgkgptCo1OH/w=
-k8s.io/apimachinery v0.0.0-20240626061445-a05248b07a6e h1:DBEwrTm4pQoqxKDZM1fY9oZ4ncSkHlmnKDFbu/icsHE=
-k8s.io/apimachinery v0.0.0-20240626061445-a05248b07a6e/go.mod h1:WJc1RfanAukQew7I55uKC34w5zx50UFDD5qo/JD4dNE=
+k8s.io/apimachinery v0.0.0-20240626061446-c225984b7bed h1:uJ7kyuzfVNEOMtzKLgbR4aUBZ1a++mNQPgBOeNii610=
+k8s.io/apimachinery v0.0.0-20240626061446-c225984b7bed/go.mod h1:WJc1RfanAukQew7I55uKC34w5zx50UFDD5qo/JD4dNE=
 k8s.io/client-go v0.0.0-20240626062855-8ffa5314741e h1:xKbYBQXJjUyvJ7lmp5bJz0Ufnoa/3Ybr4GZ42Mf9COM=
 k8s.io/client-go v0.0.0-20240626062855-8ffa5314741e/go.mod h1:10qMvliiExgeiO4z62WzFt6kQk3F4UlvFyXpjK5e8FA=
-k8s.io/component-base v0.0.0-20240626064639-1f2e30104e8a h1:6Vf1BT3V25kiYJaFOhhTkQyEHXdepiEYWRgPdh+i2AU=
-k8s.io/component-base v0.0.0-20240626064639-1f2e30104e8a/go.mod h1:HLRj6WUqGEeDExNNYzOZUHafJvSpSzPyxZJEhBgiqjI=
+k8s.io/component-base v0.0.0-20240626064641-0eb10f703efe h1:eXzjveqpDKwB8NPy8lyEAzs/7f/xb8Gk/6DCDQ02/0w=
+k8s.io/component-base v0.0.0-20240626064641-0eb10f703efe/go.mod h1:lAq7g1d6gPIb29qbxU/iqz34CASTgVPP0pB1x1/2GAo=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kms v0.0.0-20240626065322-b47e46c9b25f h1:rGbqzko9e+3Ubbn3wxJCv1dLBZpXksiINmPlb1/5NsY=
diff --git a/pkg/cel/environment/base.go b/pkg/cel/environment/base.go
index 915386999..5e2e995c1 100644
--- a/pkg/cel/environment/base.go
+++ b/pkg/cel/environment/base.go
@@ -30,20 +30,27 @@ import (
 	"k8s.io/apimachinery/pkg/util/version"
 	celconfig "k8s.io/apiserver/pkg/apis/cel"
 	"k8s.io/apiserver/pkg/cel/library"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 )
 
 // DefaultCompatibilityVersion returns a default compatibility version for use with EnvSet
 // that guarantees compatibility with CEL features/libraries/parameters understood by
-// an n-1 version
+// the api server min compatibility version
 //
-// This default will be set to no more than n-1 the current Kubernetes major.minor version.
+// This default will be set to no more than the current Kubernetes major.minor version.
 //
-// Note that a default version number less than n-1 indicates a wider range of version
-// compatibility than strictly required for rollback. A wide range of compatibility is
-// desirable because it means that CEL expressions are portable across a wider range
-// of Kubernetes versions.
+// Note that a default version number less than n-1 the current Kubernetes major.minor version
+// indicates a wider range of version compatibility than strictly required for rollback.
+// A wide range of compatibility is desirable because it means that CEL expressions are portable
+// across a wider range of Kubernetes versions.
+// A default version number equal to the current Kubernetes major.minor version
+// indicates fast forward CEL features that can be used when rollback is no longer needed.
 func DefaultCompatibilityVersion() *version.Version {
-	return version.MajorMinor(1, 30)
+	effectiveVer := utilversion.DefaultComponentGlobalsRegistry.EffectiveVersionFor(utilversion.DefaultKubeComponent)
+	if effectiveVer == nil {
+		effectiveVer = utilversion.DefaultKubeEffectiveVersion()
+	}
+	return effectiveVer.MinCompatibilityVersion()
 }
 
 var baseOpts = append(baseOptsWithoutStrictCost, StrictCostOpt)
diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go
index 80dc25cc6..8448e36ec 100644
--- a/pkg/features/kube_features.go
+++ b/pkg/features/kube_features.go
@@ -322,6 +322,17 @@ const (
 
 func init() {
 	runtime.Must(utilfeature.DefaultMutableFeatureGate.Add(defaultKubernetesFeatureGates))
+	runtime.Must(utilfeature.DefaultMutableFeatureGate.AddVersioned(defaultVersionedKubernetesFeatureGates))
+}
+
+// defaultVersionedKubernetesFeatureGates consists of all known Kubernetes-specific feature keys with VersionedSpecs.
+// To add a new feature, define a key for it above and add it here. The features will be
+// available throughout Kubernetes binaries.
+var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate.VersionedSpecs{
+	// Example:
+	// EmulationVersion: {
+	// 	{Version: version.MustParse("1.30"), Default: false, PreRelease: featuregate.Alpha},
+	// },
 }
 
 // defaultKubernetesFeatureGates consists of all known Kubernetes-specific feature keys.
diff --git a/pkg/registry/generic/options.go b/pkg/registry/generic/options.go
index d675a258f..44d07c0e2 100644
--- a/pkg/registry/generic/options.go
+++ b/pkg/registry/generic/options.go
@@ -19,6 +19,7 @@ package generic
 import (
 	"time"
 
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
@@ -39,12 +40,15 @@ type RESTOptions struct {
 }
 
 // Implement RESTOptionsGetter so that RESTOptions can directly be used when available (i.e. tests)
-func (opts RESTOptions) GetRESTOptions(schema.GroupResource) (RESTOptions, error) {
+func (opts RESTOptions) GetRESTOptions(schema.GroupResource, runtime.Object) (RESTOptions, error) {
 	return opts, nil
 }
 
 type RESTOptionsGetter interface {
-	GetRESTOptions(resource schema.GroupResource) (RESTOptions, error)
+	// GetRESTOptions returns the RESTOptions for the given resource and example object.
+	// The example object is used to determine the storage version for the resource.
+	// If the example object is nil, the storage version will be determined by the resource's default storage version.
+	GetRESTOptions(resource schema.GroupResource, example runtime.Object) (RESTOptions, error)
 }
 
 // StoreOptions is set of configuration options used to complete generic registries.
diff --git a/pkg/registry/generic/registry/store.go b/pkg/registry/generic/registry/store.go
index a8e01708a..98efac933 100644
--- a/pkg/registry/generic/registry/store.go
+++ b/pkg/registry/generic/registry/store.go
@@ -1518,7 +1518,7 @@ func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
 		return err
 	}
 
-	opts, err := options.RESTOptions.GetRESTOptions(e.DefaultQualifiedResource)
+	opts, err := options.RESTOptions.GetRESTOptions(e.DefaultQualifiedResource, e.NewFunc())
 	if err != nil {
 		return err
 	}
diff --git a/pkg/server/config.go b/pkg/server/config.go
index 0502448d6..6f0ca1bca 100644
--- a/pkg/server/config.go
+++ b/pkg/server/config.go
@@ -42,8 +42,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/version"
 	utilwaitgroup "k8s.io/apimachinery/pkg/util/waitgroup"
-	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
@@ -70,8 +70,10 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	utilflowcontrol "k8s.io/apiserver/pkg/util/flowcontrol"
 	flowcontrolrequest "k8s.io/apiserver/pkg/util/flowcontrol/request"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 	"k8s.io/client-go/informers"
 	restclient "k8s.io/client-go/rest"
+	"k8s.io/component-base/featuregate"
 	"k8s.io/component-base/logs"
 	"k8s.io/component-base/metrics/features"
 	"k8s.io/component-base/metrics/prometheus/slis"
@@ -147,8 +149,11 @@ type Config struct {
 	// done values in this values for this map are ignored.
 	PostStartHooks map[string]PostStartHookConfigEntry
 
-	// Version will enable the /version endpoint if non-nil
-	Version *version.Info
+	// EffectiveVersion determines which apis and features are available
+	// based on when the api/feature lifecyle.
+	EffectiveVersion utilversion.EffectiveVersion
+	// FeatureGate is a way to plumb feature gate through if you have them.
+	FeatureGate featuregate.FeatureGate
 	// AuditBackend is where audit events are sent to.
 	AuditBackend audit.Backend
 	// AuditPolicyRuleEvaluator makes the decision of whether and how to audit log a request.
@@ -585,7 +590,7 @@ func (c *Config) AddPostStartHookOrDie(name string, hook PostStartHookFunc) {
 	}
 }
 
-func completeOpenAPI(config *openapicommon.Config, version *version.Info) {
+func completeOpenAPI(config *openapicommon.Config, version *version.Version) {
 	if config == nil {
 		return
 	}
@@ -624,7 +629,7 @@ func completeOpenAPI(config *openapicommon.Config, version *version.Info) {
 	}
 }
 
-func completeOpenAPIV3(config *openapicommon.OpenAPIV3Config, version *version.Info) {
+func completeOpenAPIV3(config *openapicommon.OpenAPIV3Config, version *version.Version) {
 	if config == nil {
 		return
 	}
@@ -676,6 +681,9 @@ func (c *Config) ShutdownInitiatedNotify() <-chan struct{} {
 // Complete fills in any fields not set that are required to have valid data and can be derived
 // from other fields. If you're going to `ApplyOptions`, do that first. It's mutating the receiver.
 func (c *Config) Complete(informers informers.SharedInformerFactory) CompletedConfig {
+	if c.FeatureGate == nil {
+		c.FeatureGate = utilfeature.DefaultFeatureGate
+	}
 	if len(c.ExternalAddress) == 0 && c.PublicAddress != nil {
 		c.ExternalAddress = c.PublicAddress.String()
 	}
@@ -691,9 +699,8 @@ func (c *Config) Complete(informers informers.SharedInformerFactory) CompletedCo
 		}
 		c.ExternalAddress = net.JoinHostPort(c.ExternalAddress, strconv.Itoa(port))
 	}
-
-	completeOpenAPI(c.OpenAPIConfig, c.Version)
-	completeOpenAPIV3(c.OpenAPIV3Config, c.Version)
+	completeOpenAPI(c.OpenAPIConfig, c.EffectiveVersion.EmulationVersion())
+	completeOpenAPIV3(c.OpenAPIV3Config, c.EffectiveVersion.EmulationVersion())
 
 	if c.DiscoveryAddresses == nil {
 		c.DiscoveryAddresses = discovery.DefaultAddresses{DefaultAddress: c.ExternalAddress}
@@ -711,7 +718,7 @@ func (c *Config) Complete(informers informers.SharedInformerFactory) CompletedCo
 		} else {
 			c.EquivalentResourceRegistry = runtime.NewEquivalentResourceRegistryWithIdentity(func(groupResource schema.GroupResource) string {
 				// use the storage prefix as the key if possible
-				if opts, err := c.RESTOptionsGetter.GetRESTOptions(groupResource); err == nil {
+				if opts, err := c.RESTOptionsGetter.GetRESTOptions(groupResource, nil); err == nil {
 					return opts.ResourcePrefix
 				}
 				// otherwise return "" to use the default key (parent GV name)
@@ -819,12 +826,13 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 		APIServerID:           c.APIServerID,
 		StorageVersionManager: c.StorageVersionManager,
 
-		Version: c.Version,
+		EffectiveVersion: c.EffectiveVersion,
+		FeatureGate:      c.FeatureGate,
 
 		muxAndDiscoveryCompleteSignals: map[string]<-chan struct{}{},
 	}
 
-	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.AggregatedDiscoveryEndpoint) {
+	if c.FeatureGate.Enabled(genericfeatures.AggregatedDiscoveryEndpoint) {
 		manager := c.AggregatedDiscoveryGroupManager
 		if manager == nil {
 			manager = discoveryendpoint.NewResourceManager("apis")
@@ -1039,14 +1047,14 @@ func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
 		handler = genericfilters.WithRetryAfter(handler, c.lifecycleSignals.NotAcceptingNewRequest.Signaled())
 	}
 	handler = genericfilters.WithHTTPLogging(handler)
-	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerTracing) {
+	if c.FeatureGate.Enabled(genericfeatures.APIServerTracing) {
 		handler = genericapifilters.WithTracing(handler, c.TracerProvider)
 	}
 	handler = genericapifilters.WithLatencyTrackers(handler)
 	// WithRoutine will execute future handlers in a separate goroutine and serving
 	// handler in current goroutine to minimize the stack memory usage. It must be
 	// after WithPanicRecover() to be protected from panics.
-	if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServingWithRoutine) {
+	if c.FeatureGate.Enabled(genericfeatures.APIServingWithRoutine) {
 		handler = genericfilters.WithRoutine(handler, c.LongRunningFunc)
 	}
 	handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver)
@@ -1087,10 +1095,10 @@ func installAPI(s *GenericAPIServer, c *Config) {
 		}
 	}
 
-	routes.Version{Version: c.Version}.Install(s.Handler.GoRestfulContainer)
+	routes.Version{Version: c.EffectiveVersion.BinaryVersion().Info()}.Install(s.Handler.GoRestfulContainer)
 
 	if c.EnableDiscovery {
-		if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.AggregatedDiscoveryEndpoint) {
+		if c.FeatureGate.Enabled(genericfeatures.AggregatedDiscoveryEndpoint) {
 			wrapped := discoveryendpoint.WrapAggregatedDiscoveryToHandler(s.DiscoveryGroupManager, s.AggregatedDiscoveryGroupManager)
 			s.Handler.GoRestfulContainer.Add(wrapped.GenerateWebService("/apis", metav1.APIGroupList{}))
 		} else {
diff --git a/pkg/server/config_test.go b/pkg/server/config_test.go
index f58f3bf9c..a7f76d014 100644
--- a/pkg/server/config_test.go
+++ b/pkg/server/config_test.go
@@ -40,6 +40,8 @@ import (
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/server/healthz"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/rest"
@@ -90,6 +92,7 @@ func TestNewWithDelegate(t *testing.T) {
 	delegateConfig.PublicAddress = netutils.ParseIPSloppy("192.168.10.4")
 	delegateConfig.LegacyAPIGroupPrefixes = sets.NewString("/api")
 	delegateConfig.LoopbackClientConfig = &rest.Config{}
+	delegateConfig.EffectiveVersion = utilversion.NewEffectiveVersion("")
 	clientset := fake.NewSimpleClientset()
 	if clientset == nil {
 		t.Fatal("unable to create fake client set")
@@ -122,6 +125,7 @@ func TestNewWithDelegate(t *testing.T) {
 	wrappingConfig.PublicAddress = netutils.ParseIPSloppy("192.168.10.4")
 	wrappingConfig.LegacyAPIGroupPrefixes = sets.NewString("/api")
 	wrappingConfig.LoopbackClientConfig = &rest.Config{}
+	wrappingConfig.EffectiveVersion = utilversion.NewEffectiveVersion("")
 
 	wrappingConfig.HealthzChecks = append(wrappingConfig.HealthzChecks, healthz.NamedCheck("wrapping-health", func(r *http.Request) error {
 		return fmt.Errorf("wrapping failed healthcheck")
@@ -305,6 +309,7 @@ func TestAuthenticationAuditAnnotationsDefaultChain(t *testing.T) {
 		LongRunningFunc:                func(_ *http.Request, _ *request.RequestInfo) bool { return false },
 		lifecycleSignals:               newLifecycleSignals(),
 		TracerProvider:                 tracing.NewNoopTracerProvider(),
+		FeatureGate:                    utilfeature.DefaultFeatureGate,
 	}
 
 	h := DefaultBuildHandlerChain(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
diff --git a/pkg/server/deleted_kinds.go b/pkg/server/deleted_kinds.go
index f14ed9213..564b5bc43 100644
--- a/pkg/server/deleted_kinds.go
+++ b/pkg/server/deleted_kinds.go
@@ -17,6 +17,7 @@ limitations under the License.
 package server
 
 import (
+	"fmt"
 	"os"
 	"strconv"
 	"strings"
@@ -25,16 +26,15 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
-	apimachineryversion "k8s.io/apimachinery/pkg/version"
+	apimachineryversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/klog/v2"
 )
 
 // resourceExpirationEvaluator holds info for deciding if a particular rest.Storage needs to excluded from the API
 type resourceExpirationEvaluator struct {
-	currentMajor int
-	currentMinor int
-	isAlpha      bool
+	currentVersion *apimachineryversion.Version
+	isAlpha        bool
 	// This is usually set for testing for which tests need to be removed.  This prevent insta-failing CI.
 	// Set KUBE_APISERVER_STRICT_REMOVED_API_HANDLING_IN_ALPHA to see what will be removed when we tag beta
 	strictRemovedHandlingInAlpha bool
@@ -53,30 +53,17 @@ type ResourceExpirationEvaluator interface {
 	ShouldServeForVersion(majorRemoved, minorRemoved int) bool
 }
 
-func NewResourceExpirationEvaluator(currentVersion apimachineryversion.Info) (ResourceExpirationEvaluator, error) {
+func NewResourceExpirationEvaluator(currentVersion *apimachineryversion.Version) (ResourceExpirationEvaluator, error) {
+	if currentVersion == nil {
+		return nil, fmt.Errorf("empty NewResourceExpirationEvaluator currentVersion")
+	}
+	klog.V(1).Infof("NewResourceExpirationEvaluator with currentVersion: %s.", currentVersion)
 	ret := &resourceExpirationEvaluator{
 		strictRemovedHandlingInAlpha: false,
 	}
-	if len(currentVersion.Major) > 0 {
-		currentMajor64, err := strconv.ParseInt(currentVersion.Major, 10, 32)
-		if err != nil {
-			return nil, err
-		}
-		ret.currentMajor = int(currentMajor64)
-	}
-	if len(currentVersion.Minor) > 0 {
-		// split the "normal" + and - for semver stuff
-		minorString := strings.Split(currentVersion.Minor, "+")[0]
-		minorString = strings.Split(minorString, "-")[0]
-		minorString = strings.Split(minorString, ".")[0]
-		currentMinor64, err := strconv.ParseInt(minorString, 10, 32)
-		if err != nil {
-			return nil, err
-		}
-		ret.currentMinor = int(currentMinor64)
-	}
-
-	ret.isAlpha = strings.Contains(currentVersion.GitVersion, "alpha")
+	// Only keeps the major and minor versions from input version.
+	ret.currentVersion = apimachineryversion.MajorMinor(currentVersion.Major(), currentVersion.Minor())
+	ret.isAlpha = strings.Contains(currentVersion.PreRelease(), "alpha")
 
 	if envString, ok := os.LookupEnv("KUBE_APISERVER_STRICT_REMOVED_API_HANDLING_IN_ALPHA"); !ok {
 		// do nothing
@@ -112,6 +99,16 @@ func (e *resourceExpirationEvaluator) shouldServe(gv schema.GroupVersion, versio
 		return false
 	}
 
+	introduced, ok := versionedPtr.(introducedInterface)
+	// skip the introduced check for test when currentVersion is 0.0 to test all apis
+	if ok && (e.currentVersion.Major() > 0 || e.currentVersion.Minor() > 0) {
+		majorIntroduced, minorIntroduced := introduced.APILifecycleIntroduced()
+		verIntroduced := apimachineryversion.MajorMinor(uint(majorIntroduced), uint(minorIntroduced))
+		if e.currentVersion.LessThan(verIntroduced) {
+			return false
+		}
+	}
+
 	removed, ok := versionedPtr.(removedInterface)
 	if !ok {
 		return true
@@ -121,16 +118,11 @@ func (e *resourceExpirationEvaluator) shouldServe(gv schema.GroupVersion, versio
 }
 
 func (e *resourceExpirationEvaluator) ShouldServeForVersion(majorRemoved, minorRemoved int) bool {
-	if e.currentMajor < majorRemoved {
+	removedVer := apimachineryversion.MajorMinor(uint(majorRemoved), uint(minorRemoved))
+	if removedVer.GreaterThan(e.currentVersion) {
 		return true
 	}
-	if e.currentMajor > majorRemoved {
-		return false
-	}
-	if e.currentMinor < minorRemoved {
-		return true
-	}
-	if e.currentMinor > minorRemoved {
+	if removedVer.LessThan(e.currentVersion) {
 		return false
 	}
 	// at this point major and minor are equal, so this API should be removed when the current release GAs.
@@ -152,6 +144,11 @@ type removedInterface interface {
 	APILifecycleRemoved() (major, minor int)
 }
 
+// Object interface generated from "k8s:prerelease-lifecycle-gen:introduced" tags in types.go.
+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.
 // 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) {
@@ -171,6 +168,8 @@ func (e *resourceExpirationEvaluator) RemoveDeletedKinds(groupName string, versi
 			}
 
 			klog.V(1).Infof("Removing resource %v.%v.%v because it is time to stop serving it per APILifecycle.", resourceName, apiVersion, groupName)
+			storage := versionToResource[resourceName]
+			storage.Destroy()
 			delete(versionToResource, resourceName)
 		}
 		versionedResourcesStorageMap[apiVersion] = versionToResource
diff --git a/pkg/server/deleted_kinds_test.go b/pkg/server/deleted_kinds_test.go
index c033990b4..84403fd15 100644
--- a/pkg/server/deleted_kinds_test.go
+++ b/pkg/server/deleted_kinds_test.go
@@ -25,57 +25,41 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/dump"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/apimachinery/pkg/version"
+	apimachineryversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apiserver/pkg/registry/rest"
 )
 
 func Test_newResourceExpirationEvaluator(t *testing.T) {
 	tests := []struct {
 		name           string
-		currentVersion version.Info
+		currentVersion string
 		expected       resourceExpirationEvaluator
 		expectedErr    string
 	}{
 		{
-			name: "beta",
-			currentVersion: version.Info{
-				Major:      "1",
-				Minor:      "20+",
-				GitVersion: "v1.20.0-beta.0.62+a5d22854a2ac21",
-			},
-			expected: resourceExpirationEvaluator{currentMajor: 1, currentMinor: 20},
+			name:           "beta",
+			currentVersion: "v1.20.0-beta.0.62+a5d22854a2ac21",
+			expected:       resourceExpirationEvaluator{currentVersion: apimachineryversion.MajorMinor(1, 20)},
 		},
 		{
-			name: "alpha",
-			currentVersion: version.Info{
-				Major:      "1",
-				Minor:      "20+",
-				GitVersion: "v1.20.0-alpha.0.62+a5d22854a2ac21",
-			},
-			expected: resourceExpirationEvaluator{currentMajor: 1, currentMinor: 20, isAlpha: true},
+			name:           "alpha",
+			currentVersion: "v1.20.0-alpha.0.62+a5d22854a2ac21",
+			expected:       resourceExpirationEvaluator{currentVersion: apimachineryversion.MajorMinor(1, 20), isAlpha: true},
 		},
 		{
-			name: "maintenance",
-			currentVersion: version.Info{
-				Major:      "1",
-				Minor:      "20+",
-				GitVersion: "v1.20.1",
-			},
-			expected: resourceExpirationEvaluator{currentMajor: 1, currentMinor: 20},
+			name:           "maintenance",
+			currentVersion: "v1.20.1",
+			expected:       resourceExpirationEvaluator{currentVersion: apimachineryversion.MajorMinor(1, 20)},
 		},
 		{
-			name: "bad",
-			currentVersion: version.Info{
-				Major:      "1",
-				Minor:      "20something+",
-				GitVersion: "v1.20.1",
-			},
-			expectedErr: `strconv.ParseInt: parsing "20something": invalid syntax`,
+			name:           "no v prefix",
+			currentVersion: "1.20.1",
+			expected:       resourceExpirationEvaluator{currentVersion: apimachineryversion.MajorMinor(1, 20)},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			actual, actualErr := NewResourceExpirationEvaluator(tt.currentVersion)
+			actual, actualErr := NewResourceExpirationEvaluator(apimachineryversion.MustParse(tt.currentVersion))
 
 			checkErr(t, actualErr, tt.expectedErr)
 			if actualErr != nil {
@@ -90,12 +74,12 @@ func Test_newResourceExpirationEvaluator(t *testing.T) {
 	}
 }
 
-func storageRemovedIn(major, minor int) removedInStorage {
-	return removedInStorage{major: major, minor: minor}
+func storageRemovedIn(major, minor int) *removedInStorage {
+	return &removedInStorage{major: major, minor: minor}
 }
 
-func storageNeverRemoved() removedInStorage {
-	return removedInStorage{neverRemoved: true}
+func storageNeverRemoved() *removedInStorage {
+	return &removedInStorage{neverRemoved: true}
 }
 
 type removedInStorage struct {
@@ -103,23 +87,23 @@ type removedInStorage struct {
 	neverRemoved bool
 }
 
-func (r removedInStorage) New() runtime.Object {
+func (r *removedInStorage) New() runtime.Object {
 	if r.neverRemoved {
-		return neverRemovedObj{}
+		return &defaultObj{}
 	}
-	return removedInObj{major: r.major, minor: r.minor}
+	return &removedInObj{major: r.major, minor: r.minor}
 }
 
-func (r removedInStorage) Destroy() {
+func (r *removedInStorage) Destroy() {
 }
 
-type neverRemovedObj struct {
+type defaultObj struct {
 }
 
-func (r neverRemovedObj) GetObjectKind() schema.ObjectKind {
+func (r *defaultObj) GetObjectKind() schema.ObjectKind {
 	panic("don't do this")
 }
-func (r neverRemovedObj) DeepCopyObject() runtime.Object {
+func (r *defaultObj) DeepCopyObject() runtime.Object {
 	panic("don't do this either")
 }
 
@@ -127,13 +111,45 @@ type removedInObj struct {
 	major, minor int
 }
 
-func (r removedInObj) GetObjectKind() schema.ObjectKind {
+func (r *removedInObj) GetObjectKind() schema.ObjectKind {
 	panic("don't do this")
 }
-func (r removedInObj) DeepCopyObject() runtime.Object {
+func (r *removedInObj) DeepCopyObject() runtime.Object {
 	panic("don't do this either")
 }
-func (r removedInObj) APILifecycleRemoved() (major, minor int) {
+func (r *removedInObj) APILifecycleRemoved() (major, minor int) {
+	return r.major, r.minor
+}
+
+func storageIntroducedIn(major, minor int) *introducedInStorage {
+	return &introducedInStorage{major: major, minor: minor}
+}
+
+type introducedInStorage struct {
+	major, minor int
+}
+
+func (r *introducedInStorage) New() runtime.Object {
+	if r.major == 0 && r.minor == 0 {
+		return &defaultObj{}
+	}
+	return &IntroducedInObj{major: r.major, minor: r.minor}
+}
+
+func (r *introducedInStorage) Destroy() {
+}
+
+type IntroducedInObj struct {
+	major, minor int
+}
+
+func (r *IntroducedInObj) GetObjectKind() schema.ObjectKind {
+	panic("don't do this")
+}
+func (r *IntroducedInObj) DeepCopyObject() runtime.Object {
+	panic("don't do this either")
+}
+func (r *IntroducedInObj) APILifecycleIntroduced() (major, minor int) {
 	return r.major, r.minor
 }
 
@@ -147,8 +163,7 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-curr",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
 			},
 			restStorage: storageRemovedIn(1, 20),
 			expected:    false,
@@ -156,8 +171,7 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-curr-but-deferred",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor:                   1,
-				currentMinor:                   20,
+				currentVersion:                 apimachineryversion.MajorMinor(1, 20),
 				serveRemovedAPIsOneMoreRelease: true,
 			},
 			restStorage: storageRemovedIn(1, 20),
@@ -166,9 +180,8 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-curr-but-alpha",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
-				isAlpha:      true,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+				isAlpha:        true,
 			},
 			restStorage: storageRemovedIn(1, 20),
 			expected:    true,
@@ -176,8 +189,7 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-curr-but-alpha-but-strict",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor:                 1,
-				currentMinor:                 20,
+				currentVersion:               apimachineryversion.MajorMinor(1, 20),
 				isAlpha:                      true,
 				strictRemovedHandlingInAlpha: true,
 			},
@@ -187,8 +199,7 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-prev-deferral-does-not-help",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor:                   1,
-				currentMinor:                   21,
+				currentVersion:                 apimachineryversion.MajorMinor(1, 21),
 				serveRemovedAPIsOneMoreRelease: true,
 			},
 			restStorage: storageRemovedIn(1, 20),
@@ -197,8 +208,7 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-prev-major",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor:                   2,
-				currentMinor:                   20,
+				currentVersion:                 apimachineryversion.MajorMinor(2, 20),
 				serveRemovedAPIsOneMoreRelease: true,
 			},
 			restStorage: storageRemovedIn(1, 20),
@@ -207,8 +217,7 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "removed-in-future",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
 			},
 			restStorage: storageRemovedIn(1, 21),
 			expected:    true,
@@ -216,12 +225,43 @@ func Test_resourceExpirationEvaluator_shouldServe(t *testing.T) {
 		{
 			name: "never-removed",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
 			},
 			restStorage: storageNeverRemoved(),
 			expected:    true,
 		},
+		{
+			name: "introduced-in-curr",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			restStorage: storageIntroducedIn(1, 20),
+			expected:    true,
+		},
+		{
+			name: "introduced-in-prev-major",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			restStorage: storageIntroducedIn(1, 19),
+			expected:    true,
+		},
+		{
+			name: "introduced-in-future",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			restStorage: storageIntroducedIn(1, 21),
+			expected:    false,
+		},
+		{
+			name: "missing-introduced",
+			resourceExpirationEvaluator: resourceExpirationEvaluator{
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
+			},
+			restStorage: storageIntroducedIn(0, 0),
+			expected:    true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -269,8 +309,7 @@ func Test_removeDeletedKinds(t *testing.T) {
 		{
 			name: "remove-one-of-two",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
 			},
 			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
 				"v1": {
@@ -287,8 +326,7 @@ func Test_removeDeletedKinds(t *testing.T) {
 		{
 			name: "remove-nested-not-expired",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
 			},
 			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
 				"v1": {
@@ -306,8 +344,7 @@ func Test_removeDeletedKinds(t *testing.T) {
 		{
 			name: "remove-all-of-version",
 			resourceExpirationEvaluator: resourceExpirationEvaluator{
-				currentMajor: 1,
-				currentMinor: 20,
+				currentVersion: apimachineryversion.MajorMinor(1, 20),
 			},
 			versionedResourcesStorageMap: map[string]map[string]rest.Storage{
 				"v1": {
diff --git a/pkg/server/genericapiserver.go b/pkg/server/genericapiserver.go
index e0dcbf758..fed94bf4a 100644
--- a/pkg/server/genericapiserver.go
+++ b/pkg/server/genericapiserver.go
@@ -40,7 +40,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	utilwaitgroup "k8s.io/apimachinery/pkg/util/waitgroup"
-	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/audit"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
@@ -52,8 +51,9 @@ import (
 	"k8s.io/apiserver/pkg/server/healthz"
 	"k8s.io/apiserver/pkg/server/routes"
 	"k8s.io/apiserver/pkg/storageversion"
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 	restclient "k8s.io/client-go/rest"
+	"k8s.io/component-base/featuregate"
 	"k8s.io/klog/v2"
 	openapibuilder3 "k8s.io/kube-openapi/pkg/builder3"
 	openapicommon "k8s.io/kube-openapi/pkg/common"
@@ -236,8 +236,11 @@ type GenericAPIServer struct {
 	// StorageVersionManager holds the storage versions of the API resources installed by this server.
 	StorageVersionManager storageversion.Manager
 
-	// Version will enable the /version endpoint if non-nil
-	Version *version.Info
+	// EffectiveVersion determines which apis and features are available
+	// based on when the api/feature lifecyle.
+	EffectiveVersion utilversion.EffectiveVersion
+	// FeatureGate is a way to plumb feature gate through if you have them.
+	FeatureGate featuregate.FeatureGate
 
 	// lifecycleSignals provides access to the various signals that happen during the life cycle of the apiserver.
 	lifecycleSignals lifecycleSignals
@@ -776,7 +779,7 @@ func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *A
 		}
 		resourceInfos = append(resourceInfos, r...)
 
-		if utilfeature.DefaultFeatureGate.Enabled(features.AggregatedDiscoveryEndpoint) {
+		if s.FeatureGate.Enabled(features.AggregatedDiscoveryEndpoint) {
 			// Aggregated discovery only aggregates resources under /apis
 			if apiPrefix == APIGroupPrefix {
 				s.AggregatedDiscoveryGroupManager.AddGroupVersion(
@@ -804,8 +807,8 @@ func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *A
 
 	s.RegisterDestroyFunc(apiGroupInfo.destroyStorage)
 
-	if utilfeature.DefaultFeatureGate.Enabled(features.StorageVersionAPI) &&
-		utilfeature.DefaultFeatureGate.Enabled(features.APIServerIdentity) {
+	if s.FeatureGate.Enabled(features.StorageVersionAPI) &&
+		s.FeatureGate.Enabled(features.APIServerIdentity) {
 		// API installation happens before we start listening on the handlers,
 		// therefore it is safe to register ResourceInfos here. The handler will block
 		// write requests until the storage versions of the targeting resources are updated.
@@ -835,7 +838,7 @@ func (s *GenericAPIServer) InstallLegacyAPIGroup(apiPrefix string, apiGroupInfo
 	// Install the version handler.
 	// Add a handler at /<apiPrefix> to enumerate the supported api versions.
 	legacyRootAPIHandler := discovery.NewLegacyRootAPIHandler(s.discoveryAddresses, s.Serializer, apiPrefix)
-	if utilfeature.DefaultFeatureGate.Enabled(features.AggregatedDiscoveryEndpoint) {
+	if s.FeatureGate.Enabled(features.AggregatedDiscoveryEndpoint) {
 		wrapped := discoveryendpoint.WrapAggregatedDiscoveryToHandler(legacyRootAPIHandler, s.AggregatedLegacyDiscoveryGroupManager)
 		s.Handler.GoRestfulContainer.Add(wrapped.GenerateWebService("/api", metav1.APIVersions{}))
 	} else {
diff --git a/pkg/server/genericapiserver_test.go b/pkg/server/genericapiserver_test.go
index 3549b238c..c6a7cbdcf 100644
--- a/pkg/server/genericapiserver_test.go
+++ b/pkg/server/genericapiserver_test.go
@@ -48,6 +48,7 @@ import (
 	openapinamer "k8s.io/apiserver/pkg/endpoints/openapi"
 	"k8s.io/apiserver/pkg/registry/rest"
 	genericfilters "k8s.io/apiserver/pkg/server/filters"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 	"k8s.io/apiserver/pkg/warning"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
@@ -137,7 +138,7 @@ func setUp(t *testing.T) (Config, *assert.Assertions) {
 	if clientset == nil {
 		t.Fatal("unable to create fake client set")
 	}
-
+	config.EffectiveVersion = utilversion.NewEffectiveVersion("")
 	config.OpenAPIConfig = DefaultOpenAPIConfig(testGetOpenAPIDefinitions, openapinamer.NewDefinitionNamer(runtime.NewScheme()))
 	config.OpenAPIConfig.Info.Version = "unversioned"
 	config.OpenAPIV3Config = DefaultOpenAPIV3Config(testGetOpenAPIDefinitions, openapinamer.NewDefinitionNamer(runtime.NewScheme()))
@@ -459,7 +460,9 @@ func TestNotRestRoutesHaveAuth(t *testing.T) {
 	config.EnableProfiling = true
 
 	kubeVersion := fakeVersion()
-	config.Version = &kubeVersion
+	effectiveVersion := utilversion.NewEffectiveVersion(kubeVersion.String())
+	effectiveVersion.Set(effectiveVersion.BinaryVersion().WithInfo(kubeVersion), effectiveVersion.EmulationVersion(), effectiveVersion.MinCompatibilityVersion())
+	config.EffectiveVersion = effectiveVersion
 
 	s, err := config.Complete(nil).New("test", NewEmptyDelegate())
 	if err != nil {
@@ -586,7 +589,7 @@ func fakeVersion() version.Info {
 	return version.Info{
 		Major:        "42",
 		Minor:        "42",
-		GitVersion:   "42",
+		GitVersion:   "42.42",
 		GitCommit:    "34973274ccef6ab4dfaaf86599792fa9c3fe4689",
 		GitTreeState: "Dirty",
 		BuildDate:    time.Now().String(),
diff --git a/pkg/server/options/api_enablement_test.go b/pkg/server/options/api_enablement_test.go
index b37931eb2..a14319e53 100644
--- a/pkg/server/options/api_enablement_test.go
+++ b/pkg/server/options/api_enablement_test.go
@@ -32,46 +32,41 @@ func (f fakeGroupRegistry) IsGroupRegistered(group string) bool {
 
 func TestAPIEnablementOptionsValidate(t *testing.T) {
 	testCases := []struct {
-		name        string
-		testOptions *APIEnablementOptions
-		expectErr   string
+		name          string
+		runtimeConfig cliflag.ConfigurationMap
+		expectErr     string
 	}{
 		{
 			name: "test when options is nil",
 		},
 		{
-			name: "test when invalid key with only api/all=false",
-			testOptions: &APIEnablementOptions{
-				RuntimeConfig: cliflag.ConfigurationMap{"api/all": "false"},
-			},
-			expectErr: "invalid key with only api/all=false",
+			name:          "test when invalid key with only api/all=false",
+			runtimeConfig: cliflag.ConfigurationMap{"api/all": "false"},
+			expectErr:     "invalid key with only api/all=false",
 		},
 		{
-			name: "test when ConfigurationMap key is invalid",
-			testOptions: &APIEnablementOptions{
-				RuntimeConfig: cliflag.ConfigurationMap{"apiall": "false"},
-			},
-			expectErr: "runtime-config invalid key",
+			name:          "test when ConfigurationMap key is invalid",
+			runtimeConfig: cliflag.ConfigurationMap{"apiall": "false"},
+			expectErr:     "runtime-config invalid key",
 		},
 		{
-			name: "test when unknown api groups",
-			testOptions: &APIEnablementOptions{
-				RuntimeConfig: cliflag.ConfigurationMap{"api/v1": "true"},
-			},
-			expectErr: "unknown api groups",
+			name:          "test when unknown api groups",
+			runtimeConfig: cliflag.ConfigurationMap{"api/v1": "true"},
+			expectErr:     "unknown api groups",
 		},
 		{
-			name: "test when valid api groups",
-			testOptions: &APIEnablementOptions{
-				RuntimeConfig: cliflag.ConfigurationMap{"apiregistration.k8s.io/v1beta1": "true"},
-			},
+			name:          "test when valid api groups",
+			runtimeConfig: cliflag.ConfigurationMap{"apiregistration.k8s.io/v1beta1": "true"},
 		},
 	}
 	testGroupRegistry := fakeGroupRegistry{}
 
 	for _, testcase := range testCases {
 		t.Run(testcase.name, func(t *testing.T) {
-			errs := testcase.testOptions.Validate(testGroupRegistry)
+			testOptions := &APIEnablementOptions{
+				RuntimeConfig: testcase.runtimeConfig,
+			}
+			errs := testOptions.Validate(testGroupRegistry)
 			if len(testcase.expectErr) != 0 && !strings.Contains(utilerrors.NewAggregate(errs).Error(), testcase.expectErr) {
 				t.Errorf("got err: %v, expected err: %s", errs, testcase.expectErr)
 			}
diff --git a/pkg/server/options/etcd.go b/pkg/server/options/etcd.go
index 10f9775ef..af7696c40 100644
--- a/pkg/server/options/etcd.go
+++ b/pkg/server/options/etcd.go
@@ -383,8 +383,8 @@ type StorageFactoryRestOptionsFactory struct {
 	StorageFactory serverstorage.StorageFactory
 }
 
-func (f *StorageFactoryRestOptionsFactory) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
-	storageConfig, err := f.StorageFactory.NewConfig(resource)
+func (f *StorageFactoryRestOptionsFactory) GetRESTOptions(resource schema.GroupResource, example runtime.Object) (generic.RESTOptions, error) {
+	storageConfig, err := f.StorageFactory.NewConfig(resource, example)
 	if err != nil {
 		return generic.RESTOptions{}, fmt.Errorf("unable to find storage destination for %v, due to %v", resource, err.Error())
 	}
@@ -469,7 +469,7 @@ type SimpleStorageFactory struct {
 	StorageConfig storagebackend.Config
 }
 
-func (s *SimpleStorageFactory) NewConfig(resource schema.GroupResource) (*storagebackend.ConfigForResource, error) {
+func (s *SimpleStorageFactory) NewConfig(resource schema.GroupResource, example runtime.Object) (*storagebackend.ConfigForResource, error) {
 	return s.StorageConfig.ForResource(resource), nil
 }
 
@@ -493,8 +493,8 @@ type transformerStorageFactory struct {
 	resourceTransformers storagevalue.ResourceTransformers
 }
 
-func (t *transformerStorageFactory) NewConfig(resource schema.GroupResource) (*storagebackend.ConfigForResource, error) {
-	config, err := t.delegate.NewConfig(resource)
+func (t *transformerStorageFactory) NewConfig(resource schema.GroupResource, example runtime.Object) (*storagebackend.ConfigForResource, error) {
+	config, err := t.delegate.NewConfig(resource, example)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/server/options/etcd_test.go b/pkg/server/options/etcd_test.go
index 06d0cae0a..ea4464b24 100644
--- a/pkg/server/options/etcd_test.go
+++ b/pkg/server/options/etcd_test.go
@@ -437,7 +437,7 @@ func TestRestOptionsStorageObjectCountTracker(t *testing.T) {
 	if err := etcdOptions.ApplyTo(serverConfig); err != nil {
 		t.Fatalf("Failed to apply etcd options error: %v", err)
 	}
-	restOptions, err := serverConfig.RESTOptionsGetter.GetRESTOptions(schema.GroupResource{Group: "", Resource: ""})
+	restOptions, err := serverConfig.RESTOptionsGetter.GetRESTOptions(schema.GroupResource{Group: "", Resource: ""}, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/pkg/server/options/server_run_options.go b/pkg/server/options/server_run_options.go
index 1373d8a4d..593d59163 100644
--- a/pkg/server/options/server_run_options.go
+++ b/pkg/server/options/server_run_options.go
@@ -25,8 +25,10 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apimachinery/pkg/util/errors"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apiserver/pkg/server"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 
 	"github.com/spf13/pflag"
 )
@@ -89,9 +91,24 @@ type ServerRunOptions struct {
 	// This grace period is orthogonal to other grace periods, and
 	// it is not overridden by any other grace period.
 	ShutdownWatchTerminationGracePeriod time.Duration
+
+	// ComponentGlobalsRegistry is the registry where the effective versions and feature gates for all components are stored.
+	ComponentGlobalsRegistry utilversion.ComponentGlobalsRegistry
+	// ComponentName is name under which the server's global variabled are registered in the ComponentGlobalsRegistry.
+	ComponentName string
 }
 
 func NewServerRunOptions() *ServerRunOptions {
+	if utilversion.DefaultComponentGlobalsRegistry.EffectiveVersionFor(utilversion.DefaultKubeComponent) == nil {
+		featureGate := utilfeature.DefaultMutableFeatureGate
+		effectiveVersion := utilversion.DefaultKubeEffectiveVersion()
+		utilruntime.Must(utilversion.DefaultComponentGlobalsRegistry.Register(utilversion.DefaultKubeComponent, effectiveVersion, featureGate))
+	}
+
+	return NewServerRunOptionsForComponent(utilversion.DefaultKubeComponent, utilversion.DefaultComponentGlobalsRegistry)
+}
+
+func NewServerRunOptionsForComponent(componentName string, componentGlobalsRegistry utilversion.ComponentGlobalsRegistry) *ServerRunOptions {
 	defaults := server.NewConfig(serializer.CodecFactory{})
 	return &ServerRunOptions{
 		MaxRequestsInFlight:                 defaults.MaxRequestsInFlight,
@@ -104,11 +121,16 @@ func NewServerRunOptions() *ServerRunOptions {
 		JSONPatchMaxCopyBytes:               defaults.JSONPatchMaxCopyBytes,
 		MaxRequestBodyBytes:                 defaults.MaxRequestBodyBytes,
 		ShutdownSendRetryAfter:              false,
+		ComponentName:                       componentName,
+		ComponentGlobalsRegistry:            componentGlobalsRegistry,
 	}
 }
 
 // ApplyTo applies the run options to the method receiver and returns self
 func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
+	if err := s.ComponentGlobalsRegistry.SetFallback(); err != nil {
+		return err
+	}
 	c.CorsAllowedOriginList = s.CorsAllowedOriginList
 	c.HSTSDirectives = s.HSTSDirectives
 	c.ExternalAddress = s.ExternalHost
@@ -124,6 +146,8 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.PublicAddress = s.AdvertiseAddress
 	c.ShutdownSendRetryAfter = s.ShutdownSendRetryAfter
 	c.ShutdownWatchTerminationGracePeriod = s.ShutdownWatchTerminationGracePeriod
+	c.EffectiveVersion = s.ComponentGlobalsRegistry.EffectiveVersionFor(s.ComponentName)
+	c.FeatureGate = s.ComponentGlobalsRegistry.FeatureGateFor(s.ComponentName)
 
 	return nil
 }
@@ -196,6 +220,9 @@ func (s *ServerRunOptions) Validate() []error {
 	if err := validateCorsAllowedOriginList(s.CorsAllowedOriginList); err != nil {
 		errors = append(errors, err)
 	}
+	if errs := s.ComponentGlobalsRegistry.Validate(); len(errs) != 0 {
+		errors = append(errors, errs...)
+	}
 	return errors
 }
 
@@ -337,5 +364,10 @@ func (s *ServerRunOptions) AddUniversalFlags(fs *pflag.FlagSet) {
 		"This option, if set, represents the maximum amount of grace period the apiserver will wait "+
 		"for active watch request(s) to drain during the graceful server shutdown window.")
 
-	utilfeature.DefaultMutableFeatureGate.AddFlag(fs)
+	s.ComponentGlobalsRegistry.AddFlags(fs)
+}
+
+// Complete fills missing fields with defaults.
+func (s *ServerRunOptions) Complete() error {
+	return s.ComponentGlobalsRegistry.SetFallback()
 }
diff --git a/pkg/server/options/server_run_options_test.go b/pkg/server/options/server_run_options_test.go
index d1f67fe07..7bfcf0245 100644
--- a/pkg/server/options/server_run_options_test.go
+++ b/pkg/server/options/server_run_options_test.go
@@ -23,10 +23,21 @@ import (
 	"time"
 
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/version"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 	netutils "k8s.io/utils/net"
 )
 
 func TestServerRunOptionsValidate(t *testing.T) {
+	testRegistry := utilversion.NewComponentGlobalsRegistry()
+	featureGate := utilfeature.DefaultFeatureGate.DeepCopy()
+	effectiveVersion := utilversion.NewEffectiveVersion("1.30")
+	effectiveVersion.SetEmulationVersion(version.MajorMinor(1, 32))
+	testComponent := "test"
+	utilruntime.Must(testRegistry.Register(testComponent, effectiveVersion, featureGate))
+
 	testCases := []struct {
 		name        string
 		testOptions *ServerRunOptions
@@ -43,6 +54,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--max-requests-inflight can not be negative value",
 		},
@@ -57,6 +69,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--max-mutating-requests-inflight can not be negative value",
 		},
@@ -71,6 +84,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--request-timeout can not be negative value",
 		},
@@ -85,6 +99,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           -1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--min-request-timeout can not be negative value",
 		},
@@ -99,6 +114,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       -10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "ServerRunOptions.JSONPatchMaxCopyBytes can not be negative value",
 		},
@@ -113,6 +129,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         -10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "ServerRunOptions.MaxRequestBodyBytes can not be negative value",
 		},
@@ -128,6 +145,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				LivezGracePeriod:            -time.Second,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--livez-grace-period can not be a negative value",
 		},
@@ -143,6 +161,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				ShutdownDelayDuration:       -time.Second,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--shutdown-delay-duration can not be negative value",
 		},
@@ -158,9 +177,27 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 			expectErr: "--strict-transport-security-directives invalid, allowed values: max-age=expireTime, includeSubDomains, preload. see https://tools.ietf.org/html/rfc6797#section-6.1 for more information",
 		},
+		{
+			name: "Test when emulation version is invalid",
+			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,
+				ComponentName:               testComponent,
+				ComponentGlobalsRegistry:    testRegistry,
+			},
+			expectErr: "emulation version 1.32 is not between [1.29, 1.30.0]",
+		},
 		{
 			name: "Test when ServerRunOptions is valid",
 			testOptions: &ServerRunOptions{
@@ -173,6 +210,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
 				MaxRequestBodyBytes:         10 * 1024 * 1024,
+				ComponentGlobalsRegistry:    utilversion.DefaultComponentGlobalsRegistry,
 			},
 		},
 	}
diff --git a/pkg/server/options/serving_test.go b/pkg/server/options/serving_test.go
index a08ce2b3b..dc87524ef 100644
--- a/pkg/server/options/serving_test.go
+++ b/pkg/server/options/serving_test.go
@@ -43,6 +43,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/apiserver/pkg/server"
+	utilversion "k8s.io/apiserver/pkg/util/version"
 	"k8s.io/client-go/discovery"
 	restclient "k8s.io/client-go/rest"
 	cliflag "k8s.io/component-base/cli/flag"
@@ -276,9 +277,8 @@ func TestServerRunWithSNI(t *testing.T) {
 
 			// launch server
 			config := setUp(t)
-
 			v := fakeVersion()
-			config.Version = &v
+			config.EffectiveVersion = utilversion.NewEffectiveVersion(v.String())
 
 			config.EnableIndex = true
 			secureOptions := (&SecureServingOptions{
@@ -463,11 +463,9 @@ func certSignature(cert tls.Certificate) (string, error) {
 
 func fakeVersion() version.Info {
 	return version.Info{
-		Major:        "42",
-		Minor:        "42",
-		GitVersion:   "42",
-		GitCommit:    "34973274ccef6ab4dfaaf86599792fa9c3fe4689",
-		GitTreeState: "Dirty",
+		Major:      "42",
+		Minor:      "42",
+		GitVersion: "42.42",
 	}
 }
 
diff --git a/pkg/server/storage/resource_encoding_config.go b/pkg/server/storage/resource_encoding_config.go
index efb22fbc8..d73c8e62c 100644
--- a/pkg/server/storage/resource_encoding_config.go
+++ b/pkg/server/storage/resource_encoding_config.go
@@ -21,6 +21,8 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	apimachineryversion "k8s.io/apimachinery/pkg/util/version"
+	"k8s.io/apiserver/pkg/util/version"
 )
 
 type ResourceEncodingConfig interface {
@@ -33,10 +35,15 @@ type ResourceEncodingConfig interface {
 	InMemoryEncodingFor(schema.GroupResource) (schema.GroupVersion, error)
 }
 
+type CompatibilityResourceEncodingConfig interface {
+	BackwardCompatibileStorageEncodingFor(schema.GroupResource, runtime.Object) (schema.GroupVersion, error)
+}
+
 type DefaultResourceEncodingConfig struct {
 	// resources records the overriding encoding configs for individual resources.
-	resources map[schema.GroupResource]*OverridingResourceEncoding
-	scheme    *runtime.Scheme
+	resources        map[schema.GroupResource]*OverridingResourceEncoding
+	scheme           *runtime.Scheme
+	effectiveVersion version.EffectiveVersion
 }
 
 type OverridingResourceEncoding struct {
@@ -47,7 +54,7 @@ type OverridingResourceEncoding struct {
 var _ ResourceEncodingConfig = &DefaultResourceEncodingConfig{}
 
 func NewDefaultResourceEncodingConfig(scheme *runtime.Scheme) *DefaultResourceEncodingConfig {
-	return &DefaultResourceEncodingConfig{resources: map[schema.GroupResource]*OverridingResourceEncoding{}, scheme: scheme}
+	return &DefaultResourceEncodingConfig{resources: map[schema.GroupResource]*OverridingResourceEncoding{}, scheme: scheme, effectiveVersion: version.DefaultKubeEffectiveVersion()}
 }
 
 func (o *DefaultResourceEncodingConfig) SetResourceEncoding(resourceBeingStored schema.GroupResource, externalEncodingVersion, internalVersion schema.GroupVersion) {
@@ -57,6 +64,10 @@ func (o *DefaultResourceEncodingConfig) SetResourceEncoding(resourceBeingStored
 	}
 }
 
+func (o *DefaultResourceEncodingConfig) SetEffectiveVersion(effectiveVersion version.EffectiveVersion) {
+	o.effectiveVersion = effectiveVersion
+}
+
 func (o *DefaultResourceEncodingConfig) StorageEncodingFor(resource schema.GroupResource) (schema.GroupVersion, error) {
 	if !o.scheme.IsGroupRegistered(resource.Group) {
 		return schema.GroupVersion{}, fmt.Errorf("group %q is not registered in scheme", resource.Group)
@@ -71,6 +82,24 @@ func (o *DefaultResourceEncodingConfig) StorageEncodingFor(resource schema.Group
 	return o.scheme.PrioritizedVersionsForGroup(resource.Group)[0], nil
 }
 
+func (o *DefaultResourceEncodingConfig) BackwardCompatibileStorageEncodingFor(resource schema.GroupResource, example runtime.Object) (schema.GroupVersion, error) {
+	if !o.scheme.IsGroupRegistered(resource.Group) {
+		return schema.GroupVersion{}, fmt.Errorf("group %q is not registered in scheme", resource.Group)
+	}
+
+	// Always respect overrides
+	resourceOverride, resourceExists := o.resources[resource]
+	if resourceExists {
+		return resourceOverride.ExternalResourceEncoding, nil
+	}
+
+	return emulatedStorageVersion(
+		o.scheme.PrioritizedVersionsForGroup(resource.Group)[0],
+		example,
+		o.effectiveVersion,
+		o.scheme)
+}
+
 func (o *DefaultResourceEncodingConfig) InMemoryEncodingFor(resource schema.GroupResource) (schema.GroupVersion, error) {
 	if !o.scheme.IsGroupRegistered(resource.Group) {
 		return schema.GroupVersion{}, fmt.Errorf("group %q is not registered in scheme", resource.Group)
@@ -82,3 +111,79 @@ func (o *DefaultResourceEncodingConfig) InMemoryEncodingFor(resource schema.Grou
 	}
 	return schema.GroupVersion{Group: resource.Group, Version: runtime.APIVersionInternal}, nil
 }
+
+// Object interface generated from "k8s:prerelease-lifecycle-gen:introduced" tags in types.go.
+type introducedInterface interface {
+	APILifecycleIntroduced() (major, minor int)
+}
+
+func emulatedStorageVersion(binaryVersionOfResource schema.GroupVersion, example runtime.Object, effectiveVersion version.EffectiveVersion, scheme *runtime.Scheme) (schema.GroupVersion, error) {
+	if example == nil || effectiveVersion == nil {
+		return binaryVersionOfResource, nil
+	}
+
+	// Look up example in scheme to find all objects of the same Group-Kind
+	// Use the highest priority version for that group-kind whose lifecycle window
+	// includes the current emulation version.
+	// If no version is found, use the binary version
+	// (in this case the API should be disabled anyway)
+	gvks, _, err := scheme.ObjectKinds(example)
+	if err != nil {
+		return schema.GroupVersion{}, err
+	} else if len(gvks) == 0 {
+		// Probably shouldn't happen if err is non-nil
+		return schema.GroupVersion{}, fmt.Errorf("object %T has no GVKs registered in scheme", example)
+	}
+
+	// VersionsForGroupKind returns versions in priority order
+	versions := scheme.VersionsForGroupKind(schema.GroupKind{Group: gvks[0].Group, Kind: gvks[0].Kind})
+
+	compatibilityVersion := effectiveVersion.MinCompatibilityVersion()
+
+	for _, gv := range versions {
+		if gv.Version == runtime.APIVersionInternal {
+			continue
+		}
+
+		gvk := schema.GroupVersionKind{
+			Group:   gv.Group,
+			Version: gv.Version,
+			Kind:    gvks[0].Kind,
+		}
+
+		exampleOfGVK, err := scheme.New(gvk)
+		if err != nil {
+			return schema.GroupVersion{}, err
+		}
+
+		// If it was introduced after current compatibility version, don't use it
+		// skip the introduced check for test when currentVersion is 0.0 to test all apis
+		if introduced, hasIntroduced := exampleOfGVK.(introducedInterface); hasIntroduced && (compatibilityVersion.Major() > 0 || compatibilityVersion.Minor() > 0) {
+			// API resource lifecycles should be relative to k8s api version
+			majorIntroduced, minorIntroduced := introduced.APILifecycleIntroduced()
+			introducedVer := apimachineryversion.MajorMinor(uint(majorIntroduced), uint(minorIntroduced))
+			if introducedVer.GreaterThan(compatibilityVersion) {
+				continue
+			}
+		}
+
+		// versions is returned in priority order, so just use first result
+		return gvk.GroupVersion(), nil
+	}
+
+	// Getting here means we're serving a version that is unknown to the
+	// min-compatibility-version server.
+	//
+	// This is only expected to happen when serving an alpha API type due
+	// to missing pre-release lifecycle information
+	// (which doesn't happen by default), or when emulation-version and
+	// min-compatibility-version are several versions apart so a beta or GA API
+	// was being served which didn't exist at all in min-compatibility-version.
+	//
+	// In the alpha case - we do not support compatibility versioning of
+	// alpha types and recommend users do not mix the two.
+	// In the skip-level case - The version of apiserver we are retaining
+	// compatibility with has no knowledge of the type,
+	// so storing it in another type is no issue.
+	return binaryVersionOfResource, nil
+}
diff --git a/pkg/server/storage/storage_factory.go b/pkg/server/storage/storage_factory.go
index 0dc50cea6..ad01c5a5d 100644
--- a/pkg/server/storage/storage_factory.go
+++ b/pkg/server/storage/storage_factory.go
@@ -42,7 +42,7 @@ type Backend struct {
 type StorageFactory interface {
 	// New finds the storage destination for the given group and resource. It will
 	// return an error if the group has no storage destination configured.
-	NewConfig(groupResource schema.GroupResource) (*storagebackend.ConfigForResource, error)
+	NewConfig(groupResource schema.GroupResource, example runtime.Object) (*storagebackend.ConfigForResource, error)
 
 	// ResourcePrefix returns the overridden resource prefix for the GroupResource
 	// This allows for cohabitation of resources with different native types and provides
@@ -226,7 +226,7 @@ func (s *DefaultStorageFactory) getStorageGroupResource(groupResource schema.Gro
 
 // New finds the storage destination for the given group and resource. It will
 // return an error if the group has no storage destination configured.
-func (s *DefaultStorageFactory) NewConfig(groupResource schema.GroupResource) (*storagebackend.ConfigForResource, error) {
+func (s *DefaultStorageFactory) NewConfig(groupResource schema.GroupResource, example runtime.Object) (*storagebackend.ConfigForResource, error) {
 	chosenStorageResource := s.getStorageGroupResource(groupResource)
 
 	// operate on copy
@@ -244,14 +244,23 @@ func (s *DefaultStorageFactory) NewConfig(groupResource schema.GroupResource) (*
 	}
 
 	var err error
-	codecConfig.StorageVersion, err = s.ResourceEncodingConfig.StorageEncodingFor(chosenStorageResource)
-	if err != nil {
-		return nil, err
+	if backwardCompatibleInterface, ok := s.ResourceEncodingConfig.(CompatibilityResourceEncodingConfig); ok {
+		codecConfig.StorageVersion, err = backwardCompatibleInterface.BackwardCompatibileStorageEncodingFor(groupResource, example)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		codecConfig.StorageVersion, err = s.ResourceEncodingConfig.StorageEncodingFor(chosenStorageResource)
+		if err != nil {
+			return nil, err
+		}
 	}
+
 	codecConfig.MemoryVersion, err = s.ResourceEncodingConfig.InMemoryEncodingFor(groupResource)
 	if err != nil {
 		return nil, err
 	}
+
 	codecConfig.Config = storageConfig
 
 	storageConfig.Codec, storageConfig.EncodeVersioner, err = s.newStorageCodecFn(codecConfig)
diff --git a/pkg/server/storage/storage_factory_test.go b/pkg/server/storage/storage_factory_test.go
index c52049ca0..a12e865f0 100644
--- a/pkg/server/storage/storage_factory_test.go
+++ b/pkg/server/storage/storage_factory_test.go
@@ -26,10 +26,12 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
+	apimachineryversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apiserver/pkg/apis/example"
 	exampleinstall "k8s.io/apiserver/pkg/apis/example/install"
 	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
 	"k8s.io/apiserver/pkg/storage/storagebackend"
+	"k8s.io/apiserver/pkg/util/version"
 )
 
 var (
@@ -118,7 +120,7 @@ func TestConfigurableStorageFactory(t *testing.T) {
 	f.SetEtcdLocation(example.Resource("*"), []string{"/server2"})
 	f.SetEtcdPrefix(example.Resource("test"), "/prefix_for_test")
 
-	config, err := f.NewConfig(example.Resource("test"))
+	config, err := f.NewConfig(example.Resource("test"), nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -163,7 +165,7 @@ func TestUpdateEtcdOverrides(t *testing.T) {
 		storageFactory.SetEtcdLocation(test.resource, test.servers)
 
 		var err error
-		config, err := storageFactory.NewConfig(test.resource)
+		config, err := storageFactory.NewConfig(test.resource, nil)
 		if err != nil {
 			t.Errorf("%d: unexpected error %v", i, err)
 			continue
@@ -173,7 +175,7 @@ func TestUpdateEtcdOverrides(t *testing.T) {
 			continue
 		}
 
-		config, err = storageFactory.NewConfig(schema.GroupResource{Group: examplev1.GroupName, Resource: "unlikely"})
+		config, err = storageFactory.NewConfig(schema.GroupResource{Group: examplev1.GroupName, Resource: "unlikely"}, nil)
 		if err != nil {
 			t.Errorf("%d: unexpected error %v", i, err)
 			continue
@@ -244,3 +246,241 @@ func TestConfigs(t *testing.T) {
 		}
 	}
 }
+
+var introducedLifecycles = map[reflect.Type]*apimachineryversion.Version{}
+var removedLifecycles = map[reflect.Type]*apimachineryversion.Version{}
+
+type fakeLifecycler[T, V any] struct {
+	metav1.TypeMeta
+	metav1.ObjectMeta
+}
+
+type removedLifecycler[T, V any] struct {
+	fakeLifecycler[T, V]
+}
+
+func (f *fakeLifecycler[T, V]) GetObjectKind() schema.ObjectKind { return f }
+func (f *fakeLifecycler[T, V]) DeepCopyObject() runtime.Object   { return f }
+func (f *fakeLifecycler[T, V]) APILifecycleIntroduced() (major, minor int) {
+	if introduced, ok := introducedLifecycles[reflect.TypeOf(f)]; ok {
+		return int(introduced.Major()), int(introduced.Minor())
+	}
+	panic("no lifecycle version set")
+}
+func (f *removedLifecycler[T, V]) APILifecycleRemoved() (major, minor int) {
+	if removed, ok := removedLifecycles[reflect.TypeOf(f)]; ok {
+		return int(removed.Major()), int(removed.Minor())
+	}
+	panic("no lifecycle version set")
+}
+
+func registerFakeLifecycle[T, V any](sch *runtime.Scheme, group, introduced, removed string) {
+	f := fakeLifecycler[T, V]{}
+
+	introducedLifecycles[reflect.TypeOf(&f)] = apimachineryversion.MustParseSemantic(introduced)
+
+	var res runtime.Object
+	if removed != "" {
+		removedLifecycles[reflect.TypeOf(&f)] = apimachineryversion.MustParseSemantic(removed)
+		res = &removedLifecycler[T, V]{fakeLifecycler: f}
+	} else {
+		res = &f
+	}
+
+	var v V
+	var t T
+	sch.AddKnownTypeWithName(
+		schema.GroupVersionKind{
+			Group:   group,
+			Version: strings.ToLower(reflect.TypeOf(v).Name()),
+			Kind:    reflect.TypeOf(t).Name(),
+		},
+		res,
+	)
+
+	// Also ensure internal version is registered
+	// If it is registertd multiple times, it will ignore subsequent registrations
+	internalInstance := &fakeLifecycler[T, struct{}]{}
+	sch.AddKnownTypeWithName(
+		schema.GroupVersionKind{
+			Group:   group,
+			Version: runtime.APIVersionInternal,
+			Kind:    reflect.TypeOf(t).Name(),
+		},
+		internalInstance,
+	)
+}
+
+func TestStorageFactoryCompatibilityVersion(t *testing.T) {
+	// Creates a scheme with stub types for unit test
+	sch := runtime.NewScheme()
+	codecs := serializer.NewCodecFactory(sch)
+
+	type Internal = struct{}
+	type V1beta1 struct{}
+	type V1beta2 struct{}
+	type V1beta3 struct{}
+	type V1 struct{}
+
+	type Pod struct{}
+	type FlowSchema struct{}
+	type ValidatingAdmisisonPolicy struct{}
+	type CronJob struct{}
+
+	// Order dictates priority order
+	registerFakeLifecycle[FlowSchema, V1](sch, "flowcontrol.apiserver.k8s.io", "1.29.0", "")
+	registerFakeLifecycle[FlowSchema, V1beta3](sch, "flowcontrol.apiserver.k8s.io", "1.26.0", "1.32.0")
+	registerFakeLifecycle[FlowSchema, V1beta2](sch, "flowcontrol.apiserver.k8s.io", "1.23.0", "1.29.0")
+	registerFakeLifecycle[FlowSchema, V1beta1](sch, "flowcontrol.apiserver.k8s.io", "1.20.0", "1.26.0")
+	registerFakeLifecycle[CronJob, V1](sch, "batch", "1.21.0", "")
+	registerFakeLifecycle[CronJob, V1beta1](sch, "batch", "1.8.0", "1.21.0")
+	registerFakeLifecycle[ValidatingAdmisisonPolicy, V1](sch, "admissionregistration.k8s.io", "1.30.0", "")
+	registerFakeLifecycle[ValidatingAdmisisonPolicy, V1beta1](sch, "admissionregistration.k8s.io", "1.28.0", "1.34.0")
+	registerFakeLifecycle[Pod, V1](sch, "", "1.31.0", "")
+
+	// FlowSchema
+	//   - v1beta1: 1.20.0 - 1.23.0
+	//   - v1beta2: 1.23.0 - 1.26.0
+	//   - v1beta3: 1.26.0 - 1.30.0
+	//   - v1: 1.29.0+
+	// CronJob
+	//	 - v1beta1: 1.8.0 - 1.21.0
+	//	 - v1: 1.21.0+
+	// ValidatingAdmissionPolicy
+	//	 - v1beta1: 1.28.0 - 1.31.0
+	//	 - v1: 1.30.0+
+
+	testcases := []struct {
+		effectiveVersion string
+		example          runtime.Object
+		expectedVersion  schema.GroupVersion
+	}{
+		{
+			// Basic case. Beta version for long time
+			effectiveVersion: "1.14.0",
+			example:          &fakeLifecycler[CronJob, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "batch", Version: "v1beta1"},
+		},
+		{
+			// Basic case. Beta version for long time
+			effectiveVersion: "1.20.0",
+			example:          &fakeLifecycler[CronJob, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "batch", Version: "v1beta1"},
+		},
+		{
+			// Basic case. GA version for long time
+			effectiveVersion: "1.28.0",
+			example:          &fakeLifecycler[CronJob, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "batch", Version: "v1"},
+		},
+		{
+			// Basic core/v1
+			effectiveVersion: "1.31.0",
+			example:          &fakeLifecycler[Pod, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "", Version: "v1"},
+		},
+		{
+			// Corner case: 1.1.0 has no flowcontrol. Options are to error
+			// out or to use the latest version. This test assumes the latter.
+			effectiveVersion: "1.1.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1"},
+		},
+		{
+			effectiveVersion: "1.21.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1beta1"},
+		},
+		{
+			// v2Beta1 introduced this version, but minCompatibility should
+			// force v1beta1
+			effectiveVersion: "1.23.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1beta1"},
+		},
+		{
+			effectiveVersion: "1.24.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1beta2"},
+		},
+		{
+			effectiveVersion: "1.26.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1beta2"},
+		},
+		{
+			effectiveVersion: "1.27.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1beta3"},
+		},
+		{
+			// GA API introduced 1.29 but must keep storing in v1beta3 for downgrades
+			effectiveVersion: "1.29.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1beta3"},
+		},
+		{
+			// Version after GA api is introduced
+			effectiveVersion: "1.30.0",
+			example:          &fakeLifecycler[FlowSchema, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "flowcontrol.apiserver.k8s.io", Version: "v1"},
+		},
+		{
+			effectiveVersion: "1.30.0",
+			example:          &fakeLifecycler[ValidatingAdmisisonPolicy, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "admissionregistration.k8s.io", Version: "v1beta1"},
+		},
+		{
+			effectiveVersion: "1.31.0",
+			example:          &fakeLifecycler[ValidatingAdmisisonPolicy, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "admissionregistration.k8s.io", Version: "v1"},
+		},
+		{
+			effectiveVersion: "1.29.0",
+			example:          &fakeLifecycler[ValidatingAdmisisonPolicy, Internal]{},
+			expectedVersion:  schema.GroupVersion{Group: "admissionregistration.k8s.io", Version: "v1beta1"},
+		},
+	}
+
+	for _, tc := range testcases {
+		gvks, _, err := sch.ObjectKinds(tc.example)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+		gvk := gvks[0]
+		t.Run(gvk.GroupKind().String()+"@"+tc.effectiveVersion, func(t *testing.T) {
+			config := NewDefaultResourceEncodingConfig(sch)
+			config.SetEffectiveVersion(version.NewEffectiveVersion(tc.effectiveVersion))
+			f := NewDefaultStorageFactory(
+				storagebackend.Config{},
+				"",
+				codecs,
+				config,
+				NewResourceConfig(),
+				nil)
+
+			cfg, err := f.NewConfig(schema.GroupResource{
+				Group:    gvk.Group,
+				Resource: gvk.Kind, // doesnt really matter here
+			}, tc.example)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+
+			gvks, _, err := sch.ObjectKinds(tc.example)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			expectEncodeVersioner := runtime.NewMultiGroupVersioner(tc.expectedVersion,
+				schema.GroupKind{
+					Group: gvks[0].Group,
+				}, schema.GroupKind{
+					Group: gvks[0].Group,
+				})
+			if cfg.EncodeVersioner.Identifier() != expectEncodeVersioner.Identifier() {
+				t.Errorf("expected %v, got %v", expectEncodeVersioner, cfg.EncodeVersioner)
+			}
+		})
+	}
+}
diff --git a/pkg/util/feature/feature_gate.go b/pkg/util/feature/feature_gate.go
index 7dd3df589..00a9e099b 100644
--- a/pkg/util/feature/feature_gate.go
+++ b/pkg/util/feature/feature_gate.go
@@ -25,7 +25,7 @@ var (
 	// Only top-level commands/options setup and the k8s.io/component-base/featuregate/testing package should make use of this.
 	// Tests that need to modify feature gates for the duration of their test should use:
 	//   featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.<FeatureName>, <value>)
-	DefaultMutableFeatureGate featuregate.MutableFeatureGate = featuregate.NewFeatureGate()
+	DefaultMutableFeatureGate featuregate.MutableVersionedFeatureGate = featuregate.NewFeatureGate()
 
 	// DefaultFeatureGate is a shared global FeatureGate.
 	// Top-level commands/options setup that needs to modify this feature gate should use DefaultMutableFeatureGate.
diff --git a/pkg/util/version/registry.go b/pkg/util/version/registry.go
new file mode 100644
index 000000000..e6655277c
--- /dev/null
+++ b/pkg/util/version/registry.go
@@ -0,0 +1,454 @@
+/*
+Copyright 2024 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 version
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/spf13/pflag"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/version"
+	cliflag "k8s.io/component-base/cli/flag"
+	"k8s.io/component-base/featuregate"
+	"k8s.io/klog/v2"
+)
+
+// DefaultComponentGlobalsRegistry is the global var to store the effective versions and feature gates for all components for easy access.
+// Example usage:
+// // register the component effective version and feature gate first
+// _, _ = utilversion.DefaultComponentGlobalsRegistry.ComponentGlobalsOrRegister(utilversion.DefaultKubeComponent, utilversion.DefaultKubeEffectiveVersion(), utilfeature.DefaultMutableFeatureGate)
+// wardleEffectiveVersion := utilversion.NewEffectiveVersion("1.2")
+// wardleFeatureGate := featuregate.NewFeatureGate()
+// utilruntime.Must(utilversion.DefaultComponentGlobalsRegistry.Register(apiserver.WardleComponentName, wardleEffectiveVersion, wardleFeatureGate, false))
+//
+//	cmd := &cobra.Command{
+//	 ...
+//		// call DefaultComponentGlobalsRegistry.Set() in PersistentPreRunE
+//		PersistentPreRunE: func(*cobra.Command, []string) error {
+//			if err := utilversion.DefaultComponentGlobalsRegistry.Set(); err != nil {
+//				return err
+//			}
+//	 ...
+//		},
+//		RunE: func(c *cobra.Command, args []string) error {
+//			// call utilversion.DefaultComponentGlobalsRegistry.Validate() somewhere
+//		},
+//	}
+//
+// flags := cmd.Flags()
+// // add flags
+// utilversion.DefaultComponentGlobalsRegistry.AddFlags(flags)
+var DefaultComponentGlobalsRegistry ComponentGlobalsRegistry = NewComponentGlobalsRegistry()
+
+const (
+	DefaultKubeComponent = "kube"
+
+	klogLevel = 2
+)
+
+type VersionMapping func(from *version.Version) *version.Version
+
+// ComponentGlobals stores the global variables for a component for easy access.
+type ComponentGlobals struct {
+	effectiveVersion MutableEffectiveVersion
+	featureGate      featuregate.MutableVersionedFeatureGate
+
+	// emulationVersionMapping contains the mapping from the emulation version of this component
+	// to the emulation version of another component.
+	emulationVersionMapping map[string]VersionMapping
+	// dependentEmulationVersion stores whether or not this component's EmulationVersion is dependent through mapping on another component.
+	// If true, the emulation version cannot be set from the flag, or version mapping from another component.
+	dependentEmulationVersion bool
+	// minCompatibilityVersionMapping contains the mapping from the min compatibility version of this component
+	// to the min compatibility version of another component.
+	minCompatibilityVersionMapping map[string]VersionMapping
+	// dependentMinCompatibilityVersion stores whether or not this component's MinCompatibilityVersion is dependent through mapping on another component
+	// If true, the min compatibility version cannot be set from the flag, or version mapping from another component.
+	dependentMinCompatibilityVersion bool
+}
+
+type ComponentGlobalsRegistry interface {
+	// EffectiveVersionFor returns the EffectiveVersion registered under the component.
+	// Returns nil if the component is not registered.
+	EffectiveVersionFor(component string) EffectiveVersion
+	// FeatureGateFor returns the FeatureGate registered under the component.
+	// Returns nil if the component is not registered.
+	FeatureGateFor(component string) featuregate.FeatureGate
+	// Register registers the EffectiveVersion and FeatureGate for a component.
+	// returns error if the component is already registered.
+	Register(component string, effectiveVersion MutableEffectiveVersion, featureGate featuregate.MutableVersionedFeatureGate) error
+	// ComponentGlobalsOrRegister would return the registered global variables for the component if it already exists in the registry.
+	// Otherwise, the provided variables would be registered under the component, and the same variables would be returned.
+	ComponentGlobalsOrRegister(component string, effectiveVersion MutableEffectiveVersion, featureGate featuregate.MutableVersionedFeatureGate) (MutableEffectiveVersion, featuregate.MutableVersionedFeatureGate)
+	// AddFlags adds flags of "--emulated-version" and "--feature-gates"
+	AddFlags(fs *pflag.FlagSet)
+	// Set sets the flags for all global variables for all components registered.
+	Set() error
+	// SetFallback calls Set() if it has never been called.
+	SetFallback() error
+	// Validate calls the Validate() function for all the global variables for all components registered.
+	Validate() []error
+	// Reset removes all stored ComponentGlobals, configurations, and version mappings.
+	Reset()
+	// SetEmulationVersionMapping sets the mapping from the emulation version of one component
+	// to the emulation version of another component.
+	// Once set, the emulation version of the toComponent will be determined by the emulation version of the fromComponent,
+	// and cannot be set from cmd flags anymore.
+	// For a given component, its emulation version can only depend on one other component, no multiple dependency is allowed.
+	SetEmulationVersionMapping(fromComponent, toComponent string, f VersionMapping) error
+}
+
+type componentGlobalsRegistry struct {
+	componentGlobals map[string]*ComponentGlobals
+	mutex            sync.RWMutex
+	// list of component name to emulation version set from the flag.
+	emulationVersionConfig []string
+	// map of component name to the list of feature gates set from the flag.
+	featureGatesConfig map[string][]string
+	// set stores if the Set() function for the registry is already called.
+	set bool
+}
+
+func NewComponentGlobalsRegistry() *componentGlobalsRegistry {
+	return &componentGlobalsRegistry{
+		componentGlobals:       make(map[string]*ComponentGlobals),
+		emulationVersionConfig: nil,
+		featureGatesConfig:     nil,
+	}
+}
+
+func (r *componentGlobalsRegistry) Reset() {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	r.componentGlobals = make(map[string]*ComponentGlobals)
+	r.emulationVersionConfig = nil
+	r.featureGatesConfig = nil
+	r.set = false
+}
+
+func (r *componentGlobalsRegistry) EffectiveVersionFor(component string) EffectiveVersion {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	globals, ok := r.componentGlobals[component]
+	if !ok {
+		return nil
+	}
+	return globals.effectiveVersion
+}
+
+func (r *componentGlobalsRegistry) FeatureGateFor(component string) featuregate.FeatureGate {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	globals, ok := r.componentGlobals[component]
+	if !ok {
+		return nil
+	}
+	return globals.featureGate
+}
+
+func (r *componentGlobalsRegistry) unsafeRegister(component string, effectiveVersion MutableEffectiveVersion, featureGate featuregate.MutableVersionedFeatureGate) error {
+	if _, ok := r.componentGlobals[component]; ok {
+		return fmt.Errorf("component globals of %s already registered", component)
+	}
+	if featureGate != nil {
+		if err := featureGate.SetEmulationVersion(effectiveVersion.EmulationVersion()); err != nil {
+			return err
+		}
+	}
+	c := ComponentGlobals{
+		effectiveVersion:               effectiveVersion,
+		featureGate:                    featureGate,
+		emulationVersionMapping:        make(map[string]VersionMapping),
+		minCompatibilityVersionMapping: make(map[string]VersionMapping),
+	}
+	r.componentGlobals[component] = &c
+	return nil
+}
+
+func (r *componentGlobalsRegistry) Register(component string, effectiveVersion MutableEffectiveVersion, featureGate featuregate.MutableVersionedFeatureGate) error {
+	if effectiveVersion == nil {
+		return fmt.Errorf("cannot register nil effectiveVersion")
+	}
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return r.unsafeRegister(component, effectiveVersion, featureGate)
+}
+
+func (r *componentGlobalsRegistry) ComponentGlobalsOrRegister(component string, effectiveVersion MutableEffectiveVersion, featureGate featuregate.MutableVersionedFeatureGate) (MutableEffectiveVersion, featuregate.MutableVersionedFeatureGate) {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	globals, ok := r.componentGlobals[component]
+	if ok {
+		return globals.effectiveVersion, globals.featureGate
+	}
+	utilruntime.Must(r.unsafeRegister(component, effectiveVersion, featureGate))
+	return effectiveVersion, featureGate
+}
+
+func (r *componentGlobalsRegistry) unsafeKnownFeatures() []string {
+	var known []string
+	for component, globals := range r.componentGlobals {
+		if globals.featureGate == nil {
+			continue
+		}
+		for _, f := range globals.featureGate.KnownFeatures() {
+			known = append(known, component+":"+f)
+		}
+	}
+	sort.Strings(known)
+	return known
+}
+
+func (r *componentGlobalsRegistry) unsafeVersionFlagOptions(isEmulation bool) []string {
+	var vs []string
+	for component, globals := range r.componentGlobals {
+		binaryVer := globals.effectiveVersion.BinaryVersion()
+		if isEmulation {
+			if globals.dependentEmulationVersion {
+				continue
+			}
+			// emulated version could be between binaryMajor.{binaryMinor} and binaryMajor.{binaryMinor}
+			// TODO: change to binaryMajor.{binaryMinor-1} and binaryMajor.{binaryMinor} in 1.32
+			vs = append(vs, fmt.Sprintf("%s=%s..%s (default=%s)", component,
+				binaryVer.SubtractMinor(0).String(), binaryVer.String(), globals.effectiveVersion.EmulationVersion().String()))
+		} else {
+			if globals.dependentMinCompatibilityVersion {
+				continue
+			}
+			// min compatibility version could be between binaryMajor.{binaryMinor-1} and binaryMajor.{binaryMinor}
+			vs = append(vs, fmt.Sprintf("%s=%s..%s (default=%s)", component,
+				binaryVer.SubtractMinor(1).String(), binaryVer.String(), globals.effectiveVersion.MinCompatibilityVersion().String()))
+		}
+	}
+	sort.Strings(vs)
+	return vs
+}
+
+func (r *componentGlobalsRegistry) AddFlags(fs *pflag.FlagSet) {
+	if r == nil {
+		return
+	}
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	for _, globals := range r.componentGlobals {
+		if globals.featureGate != nil {
+			globals.featureGate.Close()
+		}
+	}
+	if r.emulationVersionConfig != nil || r.featureGatesConfig != nil {
+		klog.Warning("calling componentGlobalsRegistry.AddFlags more than once, the registry will be set by the latest flags")
+	}
+	r.emulationVersionConfig = []string{}
+	r.featureGatesConfig = make(map[string][]string)
+
+	fs.StringSliceVar(&r.emulationVersionConfig, "emulated-version", r.emulationVersionConfig, ""+
+		"The versions different components emulate their capabilities (APIs, features, ...) of.\n"+
+		"If set, the component will emulate the behavior of this version instead of the underlying binary version.\n"+
+		"Version format could only be major.minor, for example: '--emulated-version=wardle=1.2,kube=1.31'. Options are:\n"+strings.Join(r.unsafeVersionFlagOptions(true), "\n")+
+		"If the component is not specified, defaults to \"kube\"")
+
+	fs.Var(cliflag.NewColonSeparatedMultimapStringStringAllowDefaultEmptyKey(&r.featureGatesConfig), "feature-gates", "Comma-separated list of component:key=value pairs that describe feature gates for alpha/experimental features of different components.\n"+
+		"If the component is not specified, defaults to \"kube\". This flag can be repeatedly invoked. For example: --feature-gates 'wardle:featureA=true,wardle:featureB=false' --feature-gates 'kube:featureC=true'"+
+		"Options are:\n"+strings.Join(r.unsafeKnownFeatures(), "\n"))
+}
+
+type componentVersion struct {
+	component string
+	ver       *version.Version
+}
+
+// getFullEmulationVersionConfig expands the given version config with version registered version mapping,
+// and returns the map of component to Version.
+func (r *componentGlobalsRegistry) getFullEmulationVersionConfig(
+	versionConfigMap map[string]*version.Version) (map[string]*version.Version, error) {
+	result := map[string]*version.Version{}
+	setQueue := []componentVersion{}
+	for comp, ver := range versionConfigMap {
+		if _, ok := r.componentGlobals[comp]; !ok {
+			return result, fmt.Errorf("component not registered: %s", comp)
+		}
+		klog.V(klogLevel).Infof("setting version %s=%s", comp, ver.String())
+		setQueue = append(setQueue, componentVersion{comp, ver})
+	}
+	for len(setQueue) > 0 {
+		cv := setQueue[0]
+		if _, visited := result[cv.component]; visited {
+			return result, fmt.Errorf("setting version of %s more than once, probably version mapping loop", cv.component)
+		}
+		setQueue = setQueue[1:]
+		result[cv.component] = cv.ver
+		for toComp, f := range r.componentGlobals[cv.component].emulationVersionMapping {
+			toVer := f(cv.ver)
+			if toVer == nil {
+				return result, fmt.Errorf("got nil version from mapping of %s=%s to component:%s", cv.component, cv.ver.String(), toComp)
+			}
+			klog.V(klogLevel).Infof("setting version %s=%s from version mapping of %s=%s", toComp, toVer.String(), cv.component, cv.ver.String())
+			setQueue = append(setQueue, componentVersion{toComp, toVer})
+		}
+	}
+	return result, nil
+}
+
+func toVersionMap(versionConfig []string) (map[string]*version.Version, error) {
+	m := map[string]*version.Version{}
+	for _, compVer := range versionConfig {
+		// default to "kube" of component is not specified
+		k := "kube"
+		v := compVer
+		if strings.Contains(compVer, "=") {
+			arr := strings.SplitN(compVer, "=", 2)
+			if len(arr) != 2 {
+				return m, fmt.Errorf("malformed pair, expect string=string")
+			}
+			k = strings.TrimSpace(arr[0])
+			v = strings.TrimSpace(arr[1])
+		}
+		ver, err := version.Parse(v)
+		if err != nil {
+			return m, err
+		}
+		if ver.Patch() != 0 {
+			return m, fmt.Errorf("patch version not allowed, got: %s=%s", k, ver.String())
+		}
+		if existingVer, ok := m[k]; ok {
+			return m, fmt.Errorf("duplicate version flag, %s=%s and %s=%s", k, existingVer.String(), k, ver.String())
+		}
+		m[k] = ver
+	}
+	return m, nil
+}
+
+func (r *componentGlobalsRegistry) SetFallback() error {
+	r.mutex.Lock()
+	set := r.set
+	r.mutex.Unlock()
+	if set {
+		return nil
+	}
+	klog.Warning("setting componentGlobalsRegistry in SetFallback. We recommend calling componentGlobalsRegistry.Set()" +
+		" right after parsing flags to avoid using feature gates before their final values are set by the flags.")
+	return r.Set()
+}
+
+func (r *componentGlobalsRegistry) Set() error {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.set = true
+	emulationVersionConfigMap, err := toVersionMap(r.emulationVersionConfig)
+	if err != nil {
+		return err
+	}
+	for comp := range emulationVersionConfigMap {
+		if _, ok := r.componentGlobals[comp]; !ok {
+			return fmt.Errorf("component not registered: %s", comp)
+		}
+		// only components without any dependencies can be set from the flag.
+		if r.componentGlobals[comp].dependentEmulationVersion {
+			return fmt.Errorf("EmulationVersion of %s is set by mapping, cannot set it by flag", comp)
+		}
+	}
+	if emulationVersions, err := r.getFullEmulationVersionConfig(emulationVersionConfigMap); err != nil {
+		return err
+	} else {
+		for comp, ver := range emulationVersions {
+			r.componentGlobals[comp].effectiveVersion.SetEmulationVersion(ver)
+		}
+	}
+	// Set feature gate emulation version before setting feature gate flag values.
+	for comp, globals := range r.componentGlobals {
+		if globals.featureGate == nil {
+			continue
+		}
+		klog.V(klogLevel).Infof("setting %s:feature gate emulation version to %s", comp, globals.effectiveVersion.EmulationVersion().String())
+		if err := globals.featureGate.SetEmulationVersion(globals.effectiveVersion.EmulationVersion()); err != nil {
+			return err
+		}
+	}
+	for comp, fg := range r.featureGatesConfig {
+		if comp == "" {
+			if _, ok := r.featureGatesConfig[DefaultKubeComponent]; ok {
+				return fmt.Errorf("set kube feature gates with default empty prefix or kube: prefix consistently, do not mix use")
+			}
+			comp = DefaultKubeComponent
+		}
+		if _, ok := r.componentGlobals[comp]; !ok {
+			return fmt.Errorf("component not registered: %s", comp)
+		}
+		featureGate := r.componentGlobals[comp].featureGate
+		if featureGate == nil {
+			return fmt.Errorf("component featureGate not registered: %s", comp)
+		}
+		flagVal := strings.Join(fg, ",")
+		klog.V(klogLevel).Infof("setting %s:feature-gates=%s", comp, flagVal)
+		if err := featureGate.Set(flagVal); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *componentGlobalsRegistry) Validate() []error {
+	var errs []error
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	for _, globals := range r.componentGlobals {
+		errs = append(errs, globals.effectiveVersion.Validate()...)
+		if globals.featureGate != nil {
+			errs = append(errs, globals.featureGate.Validate()...)
+		}
+	}
+	return errs
+}
+
+func (r *componentGlobalsRegistry) SetEmulationVersionMapping(fromComponent, toComponent string, f VersionMapping) error {
+	if f == nil {
+		return nil
+	}
+	klog.V(klogLevel).Infof("setting EmulationVersion mapping from %s to %s", fromComponent, toComponent)
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	if _, ok := r.componentGlobals[fromComponent]; !ok {
+		return fmt.Errorf("component not registered: %s", fromComponent)
+	}
+	if _, ok := r.componentGlobals[toComponent]; !ok {
+		return fmt.Errorf("component not registered: %s", toComponent)
+	}
+	// check multiple dependency
+	if r.componentGlobals[toComponent].dependentEmulationVersion {
+		return fmt.Errorf("mapping of %s already exists from another component", toComponent)
+	}
+	r.componentGlobals[toComponent].dependentEmulationVersion = true
+
+	versionMapping := r.componentGlobals[fromComponent].emulationVersionMapping
+	if _, ok := versionMapping[toComponent]; ok {
+		return fmt.Errorf("EmulationVersion from %s to %s already exists", fromComponent, toComponent)
+	}
+	versionMapping[toComponent] = f
+	klog.V(klogLevel).Infof("setting the default EmulationVersion of %s based on mapping from the default EmulationVersion of %s", fromComponent, toComponent)
+	defaultFromVersion := r.componentGlobals[fromComponent].effectiveVersion.EmulationVersion()
+	emulationVersions, err := r.getFullEmulationVersionConfig(map[string]*version.Version{fromComponent: defaultFromVersion})
+	if err != nil {
+		return err
+	}
+	for comp, ver := range emulationVersions {
+		r.componentGlobals[comp].effectiveVersion.SetEmulationVersion(ver)
+	}
+	return nil
+}
diff --git a/pkg/util/version/registry_test.go b/pkg/util/version/registry_test.go
new file mode 100644
index 000000000..1badd5344
--- /dev/null
+++ b/pkg/util/version/registry_test.go
@@ -0,0 +1,418 @@
+/*
+Copyright 2024 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 version
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/spf13/pflag"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/version"
+	"k8s.io/component-base/featuregate"
+)
+
+const (
+	testComponent = "test"
+)
+
+func TestEffectiveVersionRegistry(t *testing.T) {
+	r := NewComponentGlobalsRegistry()
+	ver1 := NewEffectiveVersion("1.31")
+	ver2 := NewEffectiveVersion("1.28")
+
+	if r.EffectiveVersionFor(testComponent) != nil {
+		t.Fatalf("expected nil EffectiveVersion initially")
+	}
+	if err := r.Register(testComponent, ver1, nil); err != nil {
+		t.Fatalf("expected no error to register new component, but got err: %v", err)
+	}
+	if !r.EffectiveVersionFor(testComponent).EqualTo(ver1) {
+		t.Fatalf("expected EffectiveVersionFor to return the version registered")
+	}
+	// overwrite
+	if err := r.Register(testComponent, ver2, nil); err == nil {
+		t.Fatalf("expected error to register existing component when override is false")
+	}
+	if !r.EffectiveVersionFor(testComponent).EqualTo(ver1) {
+		t.Fatalf("expected EffectiveVersionFor to return the version overridden")
+	}
+}
+
+func testRegistry(t *testing.T) *componentGlobalsRegistry {
+	r := NewComponentGlobalsRegistry()
+	verKube := NewEffectiveVersion("1.31")
+	fgKube := featuregate.NewVersionedFeatureGate(version.MustParse("0.0"))
+	err := fgKube.AddVersioned(map[featuregate.Feature]featuregate.VersionedSpecs{
+		"kubeA": {
+			{Version: version.MustParse("1.31"), Default: true, LockToDefault: true, PreRelease: featuregate.GA},
+			{Version: version.MustParse("1.28"), Default: false, PreRelease: featuregate.Beta},
+			{Version: version.MustParse("1.27"), Default: false, PreRelease: featuregate.Alpha},
+		},
+		"kubeB": {
+			{Version: version.MustParse("1.30"), Default: false, PreRelease: featuregate.Alpha},
+		},
+		"commonC": {
+			{Version: version.MustParse("1.29"), Default: true, PreRelease: featuregate.Beta},
+			{Version: version.MustParse("1.27"), Default: false, PreRelease: featuregate.Alpha},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	verTest := NewEffectiveVersion("2.8")
+	fgTest := featuregate.NewVersionedFeatureGate(version.MustParse("0.0"))
+	err = fgTest.AddVersioned(map[featuregate.Feature]featuregate.VersionedSpecs{
+		"testA": {
+			{Version: version.MustParse("2.10"), Default: true, PreRelease: featuregate.GA},
+			{Version: version.MustParse("2.8"), Default: false, PreRelease: featuregate.Beta},
+			{Version: version.MustParse("2.7"), Default: false, PreRelease: featuregate.Alpha},
+		},
+		"testB": {
+			{Version: version.MustParse("2.9"), Default: false, PreRelease: featuregate.Alpha},
+		},
+		"commonC": {
+			{Version: version.MustParse("2.9"), Default: true, PreRelease: featuregate.Beta},
+			{Version: version.MustParse("2.7"), Default: false, PreRelease: featuregate.Alpha},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	utilruntime.Must(r.Register(DefaultKubeComponent, verKube, fgKube))
+	utilruntime.Must(r.Register(testComponent, verTest, fgTest))
+	return r
+}
+
+func TestVersionFlagOptions(t *testing.T) {
+	r := testRegistry(t)
+	emuVers := strings.Join(r.unsafeVersionFlagOptions(true), "\n")
+	expectedEmuVers := "kube=1.31..1.31 (default=1.31)\ntest=2.8..2.8 (default=2.8)"
+	if emuVers != expectedEmuVers {
+		t.Errorf("wanted emulation version flag options to be: %s, got %s", expectedEmuVers, emuVers)
+	}
+	minCompVers := strings.Join(r.unsafeVersionFlagOptions(false), "\n")
+	expectedMinCompVers := "kube=1.30..1.31 (default=1.30)\ntest=2.7..2.8 (default=2.7)"
+	if minCompVers != expectedMinCompVers {
+		t.Errorf("wanted min compatibility version flag options to be: %s, got %s", expectedMinCompVers, minCompVers)
+	}
+}
+
+func TestVersionFlagOptionsWithMapping(t *testing.T) {
+	r := testRegistry(t)
+	utilruntime.Must(r.SetEmulationVersionMapping(testComponent, DefaultKubeComponent,
+		func(from *version.Version) *version.Version { return from.OffsetMinor(3) }))
+	emuVers := strings.Join(r.unsafeVersionFlagOptions(true), "\n")
+	expectedEmuVers := "test=2.8..2.8 (default=2.8)"
+	if emuVers != expectedEmuVers {
+		t.Errorf("wanted emulation version flag options to be: %s, got %s", expectedEmuVers, emuVers)
+	}
+	minCompVers := strings.Join(r.unsafeVersionFlagOptions(false), "\n")
+	expectedMinCompVers := "kube=1.30..1.31 (default=1.30)\ntest=2.7..2.8 (default=2.7)"
+	if minCompVers != expectedMinCompVers {
+		t.Errorf("wanted min compatibility version flag options to be: %s, got %s", expectedMinCompVers, minCompVers)
+	}
+}
+
+func TestVersionedFeatureGateFlag(t *testing.T) {
+	r := testRegistry(t)
+	known := strings.Join(r.unsafeKnownFeatures(), "\n")
+	expectedKnown := "kube:AllAlpha=true|false (ALPHA - default=false)\n" +
+		"kube:AllBeta=true|false (BETA - default=false)\n" +
+		"kube:commonC=true|false (BETA - default=true)\n" +
+		"kube:kubeB=true|false (ALPHA - default=false)\n" +
+		"test:AllAlpha=true|false (ALPHA - default=false)\n" +
+		"test:AllBeta=true|false (BETA - default=false)\n" +
+		"test:commonC=true|false (ALPHA - default=false)\n" +
+		"test:testA=true|false (BETA - default=false)"
+	if known != expectedKnown {
+		t.Errorf("wanted min compatibility version flag options to be:\n%s, got:\n%s", expectedKnown, known)
+	}
+}
+
+func TestFlags(t *testing.T) {
+	tests := []struct {
+		name                         string
+		flags                        []string
+		parseError                   string
+		expectedKubeEmulationVersion string
+		expectedTestEmulationVersion string
+		expectedKubeFeatureValues    map[featuregate.Feature]bool
+		expectedTestFeatureValues    map[featuregate.Feature]bool
+	}{
+		{
+			name:                         "setting kube emulation version",
+			flags:                        []string{"--emulated-version=kube=1.30"},
+			expectedKubeEmulationVersion: "1.30",
+		},
+		{
+			name: "setting kube emulation version twice",
+			flags: []string{
+				"--emulated-version=kube=1.30",
+				"--emulated-version=kube=1.32",
+			},
+			parseError: "duplicate version flag, kube=1.30 and kube=1.32",
+		},
+		{
+			name:                         "prefix v ok",
+			flags:                        []string{"--emulated-version=kube=v1.30"},
+			expectedKubeEmulationVersion: "1.30",
+		},
+		{
+			name:       "patch version not ok",
+			flags:      []string{"--emulated-version=kube=1.30.2"},
+			parseError: "patch version not allowed, got: kube=1.30.2",
+		},
+		{
+			name:                         "setting test emulation version",
+			flags:                        []string{"--emulated-version=test=2.7"},
+			expectedKubeEmulationVersion: "1.31",
+			expectedTestEmulationVersion: "2.7",
+		},
+		{
+			name:                         "version missing component default to kube",
+			flags:                        []string{"--emulated-version=1.30"},
+			expectedKubeEmulationVersion: "1.30",
+		},
+		{
+			name:       "version missing component default to kube with duplicate",
+			flags:      []string{"--emulated-version=1.30", "--emulated-version=kube=1.30"},
+			parseError: "duplicate version flag, kube=1.30 and kube=1.30",
+		},
+		{
+			name:       "version unregistered component",
+			flags:      []string{"--emulated-version=test3=1.31"},
+			parseError: "component not registered: test3",
+		},
+		{
+			name:       "invalid version",
+			flags:      []string{"--emulated-version=test=1.foo"},
+			parseError: "illegal version string \"1.foo\"",
+		},
+		{
+			name: "setting test feature flag",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--feature-gates=test:testA=true",
+			},
+			expectedKubeEmulationVersion: "1.31",
+			expectedTestEmulationVersion: "2.7",
+			expectedKubeFeatureValues:    map[featuregate.Feature]bool{"kubeA": true, "kubeB": false, "commonC": true},
+			expectedTestFeatureValues:    map[featuregate.Feature]bool{"testA": true, "testB": false, "commonC": false},
+		},
+		{
+			name: "setting future test feature flag",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--feature-gates=test:testA=true,test:testB=true",
+			},
+			parseError: "cannot set feature gate testB to true, feature is PreAlpha at emulated version 2.7",
+		},
+		{
+			name: "setting kube feature flag",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--emulated-version=kube=1.30",
+				"--feature-gates=kubeB=false,test:commonC=true",
+				"--feature-gates=commonC=false,kubeB=true",
+			},
+			expectedKubeEmulationVersion: "1.30",
+			expectedTestEmulationVersion: "2.7",
+			expectedKubeFeatureValues:    map[featuregate.Feature]bool{"kubeA": false, "kubeB": true, "commonC": false},
+			expectedTestFeatureValues:    map[featuregate.Feature]bool{"testA": false, "testB": false, "commonC": true},
+		},
+		{
+			name: "setting kube feature flag with different prefix",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--emulated-version=kube=1.30",
+				"--feature-gates=kube:kubeB=false,test:commonC=true",
+				"--feature-gates=commonC=false,kubeB=true",
+			},
+			parseError: "set kube feature gates with default empty prefix or kube: prefix consistently, do not mix use",
+		},
+		{
+			name: "setting locked kube feature flag",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--feature-gates=kubeA=false",
+			},
+			parseError: "cannot set feature gate kubeA to false, feature is locked to true",
+		},
+		{
+			name: "setting unknown test feature flag",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--feature-gates=test:testD=true",
+			},
+			parseError: "unrecognized feature gate: testD",
+		},
+		{
+			name: "setting unknown component feature flag",
+			flags: []string{
+				"--emulated-version=test=2.7",
+				"--feature-gates=test3:commonC=true",
+			},
+			parseError: "component not registered: test3",
+		},
+	}
+	for i, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fs := pflag.NewFlagSet("testflag", pflag.ContinueOnError)
+			r := testRegistry(t)
+			r.AddFlags(fs)
+			err := fs.Parse(test.flags)
+			if err == nil {
+				err = r.Set()
+			}
+			if test.parseError != "" {
+				if err == nil || !strings.Contains(err.Error(), test.parseError) {
+					t.Fatalf("%d: Parse() expected: %v, got: %v", i, test.parseError, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("%d: Parse() expected: nil, got: %v", i, err)
+			}
+			if len(test.expectedKubeEmulationVersion) > 0 {
+				assertVersionEqualTo(t, r.EffectiveVersionFor(DefaultKubeComponent).EmulationVersion(), test.expectedKubeEmulationVersion)
+			}
+			if len(test.expectedTestEmulationVersion) > 0 {
+				assertVersionEqualTo(t, r.EffectiveVersionFor(testComponent).EmulationVersion(), test.expectedTestEmulationVersion)
+			}
+			for f, v := range test.expectedKubeFeatureValues {
+				if r.FeatureGateFor(DefaultKubeComponent).Enabled(f) != v {
+					t.Errorf("%d: expected kube feature Enabled(%s)=%v", i, f, v)
+				}
+			}
+			for f, v := range test.expectedTestFeatureValues {
+				if r.FeatureGateFor(testComponent).Enabled(f) != v {
+					t.Errorf("%d: expected test feature Enabled(%s)=%v", i, f, v)
+				}
+			}
+		})
+	}
+}
+
+func TestVersionMapping(t *testing.T) {
+	r := NewComponentGlobalsRegistry()
+	ver1 := NewEffectiveVersion("0.58")
+	ver2 := NewEffectiveVersion("1.28")
+	ver3 := NewEffectiveVersion("2.10")
+
+	utilruntime.Must(r.Register("test1", ver1, nil))
+	utilruntime.Must(r.Register("test2", ver2, nil))
+	utilruntime.Must(r.Register("test3", ver3, nil))
+
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test1").EmulationVersion(), "0.58")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test2").EmulationVersion(), "1.28")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test3").EmulationVersion(), "2.10")
+
+	utilruntime.Must(r.SetEmulationVersionMapping("test2", "test3",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()+1, from.Minor()-19)
+		}))
+	utilruntime.Must(r.SetEmulationVersionMapping("test1", "test2",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()+1, from.Minor()-28)
+		}))
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test1").EmulationVersion(), "0.58")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test2").EmulationVersion(), "1.30")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test3").EmulationVersion(), "2.11")
+
+	fs := pflag.NewFlagSet("testflag", pflag.ContinueOnError)
+	r.AddFlags(fs)
+
+	if err := fs.Parse([]string{fmt.Sprintf("--emulated-version=%s", "test1=0.56")}); err != nil {
+		t.Fatal(err)
+		return
+	}
+	if err := r.Set(); err != nil {
+		t.Fatal(err)
+		return
+	}
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test1").EmulationVersion(), "0.56")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test2").EmulationVersion(), "1.28")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test3").EmulationVersion(), "2.09")
+}
+
+func TestVersionMappingWithMultipleDependency(t *testing.T) {
+	r := NewComponentGlobalsRegistry()
+	ver1 := NewEffectiveVersion("0.58")
+	ver2 := NewEffectiveVersion("1.28")
+	ver3 := NewEffectiveVersion("2.10")
+
+	utilruntime.Must(r.Register("test1", ver1, nil))
+	utilruntime.Must(r.Register("test2", ver2, nil))
+	utilruntime.Must(r.Register("test3", ver3, nil))
+
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test1").EmulationVersion(), "0.58")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test2").EmulationVersion(), "1.28")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test3").EmulationVersion(), "2.10")
+
+	utilruntime.Must(r.SetEmulationVersionMapping("test1", "test2",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()+1, from.Minor()-28)
+		}))
+	err := r.SetEmulationVersionMapping("test3", "test2",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()-1, from.Minor()+19)
+		})
+	if err == nil {
+		t.Errorf("expect error when setting 2nd mapping to test2")
+	}
+}
+
+func TestVersionMappingWithCyclicDependency(t *testing.T) {
+	r := NewComponentGlobalsRegistry()
+	ver1 := NewEffectiveVersion("0.58")
+	ver2 := NewEffectiveVersion("1.28")
+	ver3 := NewEffectiveVersion("2.10")
+
+	utilruntime.Must(r.Register("test1", ver1, nil))
+	utilruntime.Must(r.Register("test2", ver2, nil))
+	utilruntime.Must(r.Register("test3", ver3, nil))
+
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test1").EmulationVersion(), "0.58")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test2").EmulationVersion(), "1.28")
+	assertVersionEqualTo(t, r.EffectiveVersionFor("test3").EmulationVersion(), "2.10")
+
+	utilruntime.Must(r.SetEmulationVersionMapping("test1", "test2",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()+1, from.Minor()-28)
+		}))
+	utilruntime.Must(r.SetEmulationVersionMapping("test2", "test3",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()+1, from.Minor()-19)
+		}))
+	err := r.SetEmulationVersionMapping("test3", "test1",
+		func(from *version.Version) *version.Version {
+			return version.MajorMinor(from.Major()-2, from.Minor()+48)
+		})
+	if err == nil {
+		t.Errorf("expect cyclic version mapping error")
+	}
+}
+
+func assertVersionEqualTo(t *testing.T, ver *version.Version, expectedVer string) {
+	if ver.EqualTo(version.MustParse(expectedVer)) {
+		return
+	}
+	t.Errorf("expected: %s, got %s", expectedVer, ver.String())
+}
diff --git a/pkg/util/version/version.go b/pkg/util/version/version.go
new file mode 100644
index 000000000..a7a5fda87
--- /dev/null
+++ b/pkg/util/version/version.go
@@ -0,0 +1,157 @@
+/*
+Copyright 2024 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 version
+
+import (
+	"fmt"
+	"sync/atomic"
+
+	"k8s.io/apimachinery/pkg/util/version"
+	baseversion "k8s.io/component-base/version"
+)
+
+type EffectiveVersion interface {
+	BinaryVersion() *version.Version
+	EmulationVersion() *version.Version
+	MinCompatibilityVersion() *version.Version
+	EqualTo(other EffectiveVersion) bool
+	String() string
+	Validate() []error
+}
+
+type MutableEffectiveVersion interface {
+	EffectiveVersion
+	Set(binaryVersion, emulationVersion, minCompatibilityVersion *version.Version)
+	SetEmulationVersion(emulationVersion *version.Version)
+	SetMinCompatibilityVersion(minCompatibilityVersion *version.Version)
+}
+
+type effectiveVersion struct {
+	binaryVersion atomic.Pointer[version.Version]
+	// If the emulationVersion is set by the users, it could only contain major and minor versions.
+	// In tests, emulationVersion could be the same as the binary version, or set directly,
+	// which can have "alpha" as pre-release to continue serving expired apis while we clean up the test.
+	emulationVersion atomic.Pointer[version.Version]
+	// minCompatibilityVersion could only contain major and minor versions.
+	minCompatibilityVersion atomic.Pointer[version.Version]
+}
+
+func (m *effectiveVersion) BinaryVersion() *version.Version {
+	return m.binaryVersion.Load()
+}
+
+func (m *effectiveVersion) EmulationVersion() *version.Version {
+	ver := m.emulationVersion.Load()
+	if ver != nil {
+		// Emulation version can have "alpha" as pre-release to continue serving expired apis while we clean up the test.
+		// The pre-release should not be accessible to the users.
+		return ver.WithPreRelease(m.BinaryVersion().PreRelease())
+	}
+	return ver
+}
+
+func (m *effectiveVersion) MinCompatibilityVersion() *version.Version {
+	return m.minCompatibilityVersion.Load()
+}
+
+func (m *effectiveVersion) EqualTo(other EffectiveVersion) bool {
+	return m.BinaryVersion().EqualTo(other.BinaryVersion()) && m.EmulationVersion().EqualTo(other.EmulationVersion()) && m.MinCompatibilityVersion().EqualTo(other.MinCompatibilityVersion())
+}
+
+func (m *effectiveVersion) String() string {
+	if m == nil {
+		return "<nil>"
+	}
+	return fmt.Sprintf("{BinaryVersion: %s, EmulationVersion: %s, MinCompatibilityVersion: %s}",
+		m.BinaryVersion().String(), m.EmulationVersion().String(), m.MinCompatibilityVersion().String())
+}
+
+func majorMinor(ver *version.Version) *version.Version {
+	if ver == nil {
+		return ver
+	}
+	return version.MajorMinor(ver.Major(), ver.Minor())
+}
+
+func (m *effectiveVersion) Set(binaryVersion, emulationVersion, minCompatibilityVersion *version.Version) {
+	m.binaryVersion.Store(binaryVersion)
+	m.emulationVersion.Store(majorMinor(emulationVersion))
+	m.minCompatibilityVersion.Store(majorMinor(minCompatibilityVersion))
+}
+
+func (m *effectiveVersion) SetEmulationVersion(emulationVersion *version.Version) {
+	m.emulationVersion.Store(majorMinor(emulationVersion))
+}
+
+func (m *effectiveVersion) SetMinCompatibilityVersion(minCompatibilityVersion *version.Version) {
+	m.minCompatibilityVersion.Store(majorMinor(minCompatibilityVersion))
+}
+
+func (m *effectiveVersion) Validate() []error {
+	var errs []error
+	// Validate only checks the major and minor versions.
+	binaryVersion := m.binaryVersion.Load().WithPatch(0)
+	emulationVersion := m.emulationVersion.Load()
+	minCompatibilityVersion := m.minCompatibilityVersion.Load()
+
+	// emulationVersion can only be 1.{binaryMinor-1}...1.{binaryMinor}.
+	maxEmuVer := binaryVersion
+	minEmuVer := binaryVersion.SubtractMinor(1)
+	if emulationVersion.GreaterThan(maxEmuVer) || emulationVersion.LessThan(minEmuVer) {
+		errs = append(errs, fmt.Errorf("emulation version %s is not between [%s, %s]", emulationVersion.String(), minEmuVer.String(), maxEmuVer.String()))
+	}
+	// minCompatibilityVersion can only be 1.{binaryMinor-1} for alpha.
+	maxCompVer := binaryVersion.SubtractMinor(1)
+	minCompVer := binaryVersion.SubtractMinor(1)
+	if minCompatibilityVersion.GreaterThan(maxCompVer) || minCompatibilityVersion.LessThan(minCompVer) {
+		errs = append(errs, fmt.Errorf("minCompatibilityVersion version %s is not between [%s, %s]", minCompatibilityVersion.String(), minCompVer.String(), maxCompVer.String()))
+	}
+	return errs
+}
+
+func newEffectiveVersion(binaryVersion *version.Version) MutableEffectiveVersion {
+	effective := &effectiveVersion{}
+	compatVersion := binaryVersion.SubtractMinor(1)
+	effective.Set(binaryVersion, binaryVersion, compatVersion)
+	return effective
+}
+
+func NewEffectiveVersion(binaryVer string) MutableEffectiveVersion {
+	if binaryVer == "" {
+		return &effectiveVersion{}
+	}
+	binaryVersion := version.MustParse(binaryVer)
+	return newEffectiveVersion(binaryVersion)
+}
+
+// DefaultBuildEffectiveVersion returns the MutableEffectiveVersion based on the
+// current build information.
+func DefaultBuildEffectiveVersion() MutableEffectiveVersion {
+	verInfo := baseversion.Get()
+	binaryVersion := version.MustParse(verInfo.String()).WithInfo(verInfo)
+	if binaryVersion.Major() == 0 && binaryVersion.Minor() == 0 {
+		return DefaultKubeEffectiveVersion()
+	}
+	return newEffectiveVersion(binaryVersion)
+}
+
+// DefaultKubeEffectiveVersion returns the MutableEffectiveVersion based on the
+// latest K8s release.
+func DefaultKubeEffectiveVersion() MutableEffectiveVersion {
+	binaryVersion := version.MustParse(baseversion.DefaultKubeBinaryVersion).WithInfo(baseversion.Get())
+	return newEffectiveVersion(binaryVersion)
+}
diff --git a/pkg/util/version/version_test.go b/pkg/util/version/version_test.go
new file mode 100644
index 000000000..aff8a8e4a
--- /dev/null
+++ b/pkg/util/version/version_test.go
@@ -0,0 +1,107 @@
+/*
+Copyright 2024 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 version
+
+import (
+	"testing"
+
+	"k8s.io/apimachinery/pkg/util/version"
+)
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		name                    string
+		binaryVersion           string
+		emulationVersion        string
+		minCompatibilityVersion string
+		expectErrors            bool
+	}{
+		{
+			name:                    "patch version diff ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.32.1",
+			minCompatibilityVersion: "v1.31.5",
+		},
+		{
+			name:                    "emulation version one minor lower than binary ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.31.0",
+			minCompatibilityVersion: "v1.31.0",
+		},
+		{
+			name:                    "emulation version two minor lower than binary not ok",
+			binaryVersion:           "v1.33.2",
+			emulationVersion:        "v1.31.0",
+			minCompatibilityVersion: "v1.32.0",
+			expectErrors:            true,
+		},
+		{
+			name:                    "emulation version one minor higher than binary not ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.33.0",
+			minCompatibilityVersion: "v1.31.0",
+			expectErrors:            true,
+		},
+		{
+			name:                    "emulation version two minor higher than binary not ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.34.0",
+			minCompatibilityVersion: "v1.31.0",
+			expectErrors:            true,
+		},
+		{
+			name:                    "compatibility version same as binary not ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.32.0",
+			minCompatibilityVersion: "v1.32.0",
+			expectErrors:            true,
+		},
+		{
+			name:                    "compatibility version two minor lower than binary not ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.32.0",
+			minCompatibilityVersion: "v1.30.0",
+			expectErrors:            true,
+		},
+		{
+			name:                    "compatibility version one minor higher than binary not ok",
+			binaryVersion:           "v1.32.2",
+			emulationVersion:        "v1.32.0",
+			minCompatibilityVersion: "v1.33.0",
+			expectErrors:            true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			binaryVersion := version.MustParseGeneric(test.binaryVersion)
+			effective := &effectiveVersion{}
+			emulationVersion := version.MustParseGeneric(test.emulationVersion)
+			minCompatibilityVersion := version.MustParseGeneric(test.minCompatibilityVersion)
+			effective.Set(binaryVersion, emulationVersion, minCompatibilityVersion)
+
+			errs := effective.Validate()
+			if len(errs) > 0 && !test.expectErrors {
+				t.Errorf("expected no errors, errors found %+v", errs)
+			}
+
+			if len(errs) == 0 && test.expectErrors {
+				t.Errorf("expected errors, no errors found")
+			}
+		})
+	}
+}