diff --git a/cmd/kube-apiserver/app/apiextensions.go b/cmd/kube-apiserver/app/apiextensions.go
index ee80cd9a7c9..59bc0079f30 100644
--- a/cmd/kube-apiserver/app/apiextensions.go
+++ b/cmd/kube-apiserver/app/apiextensions.go
@@ -23,6 +23,7 @@ import (
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	apiextensionsapiserver "k8s.io/apiextensions-apiserver/pkg/apiserver"
+	"k8s.io/apiextensions-apiserver/pkg/apiserver/conversion"
 	apiextensionsoptions "k8s.io/apiextensions-apiserver/pkg/cmd/server/options"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -71,10 +72,17 @@ func createAPIExtensionsConfig(
 		apiextensionsapiserver.Scheme); err != nil {
 		return nil, err
 	}
+
 	crdRESTOptionsGetter, err := apiextensionsoptions.NewCRDRESTOptionsGetter(etcdOptions)
 	if err != nil {
 		return nil, err
 	}
+
+	conversionFactory, err := conversion.NewCRConverterFactory(serviceResolver, authResolverWrapper)
+	if err != nil {
+		return nil, err
+	}
+
 	apiextensionsConfig := &apiextensionsapiserver.Config{
 		GenericConfig: &genericapiserver.RecommendedConfig{
 			Config:                genericConfig,
@@ -83,8 +91,7 @@ func createAPIExtensionsConfig(
 		ExtraConfig: apiextensionsapiserver.ExtraConfig{
 			CRDRESTOptionsGetter: crdRESTOptionsGetter,
 			MasterCount:          masterCount,
-			AuthResolverWrapper:  authResolverWrapper,
-			ServiceResolver:      serviceResolver,
+			ConversionFactory:    conversionFactory,
 		},
 	}
 
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/apiserver.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/apiserver.go
index dd2d7a51103..8e999b9bf4c 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/apiserver.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/apiserver.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/install"
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
+	"k8s.io/apiextensions-apiserver/pkg/apiserver/conversion"
 	"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	externalinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions"
 	"k8s.io/apiextensions-apiserver/pkg/controller/apiapproval"
@@ -48,7 +49,6 @@ import (
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	serverstorage "k8s.io/apiserver/pkg/server/storage"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
-	"k8s.io/apiserver/pkg/util/webhook"
 )
 
 var (
@@ -83,10 +83,8 @@ type ExtraConfig struct {
 	// the CRD Establishing will be hold by 5 seconds.
 	MasterCount int
 
-	// ServiceResolver is used in CR webhook converters to resolve webhook's service names
-	ServiceResolver webhook.ServiceResolver
-	// AuthResolverWrapper is used in CR webhook converters
-	AuthResolverWrapper webhook.AuthenticationInfoResolverWrapper
+	// ConversionFactory is used to provider converters for CRs.
+	ConversionFactory conversion.Factory
 }
 
 type Config struct {
@@ -197,8 +195,7 @@ func (c completedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 		c.ExtraConfig.CRDRESTOptionsGetter,
 		c.GenericConfig.AdmissionControl,
 		establishingController,
-		c.ExtraConfig.ServiceResolver,
-		c.ExtraConfig.AuthResolverWrapper,
+		c.ExtraConfig.ConversionFactory,
 		c.ExtraConfig.MasterCount,
 		s.GenericAPIServer.Authorizer,
 		c.GenericConfig.RequestTimeout,
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter.go
index 96b2b89e0e8..06c26b3296d 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter.go
@@ -21,13 +21,28 @@ import (
 
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	metav1validation "k8s.io/apimachinery/pkg/apis/meta/v1/validation"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apiserver/pkg/util/webhook"
 	typedscheme "k8s.io/client-go/kubernetes/scheme"
 )
 
+// Factory is able to create a new CRConverter for crd.
+type Factory interface {
+	// NewConverter returns a CRConverter capable of converting crd's versions.
+	//
+	// For proper conversion, the returned CRConverter must be used via NewDelegatingConverter.
+	//
+	// When implementing a CRConverter, you do not need to: test for valid API versions or no-op
+	// conversions, handle field selector logic, or handle scale conversions; these are all handled
+	// via NewDelegatingConverter.
+	NewConverter(crd *apiextensionsv1.CustomResourceDefinition) (CRConverter, error)
+}
+
 // CRConverterFactory is the factory for all CR converters.
 type CRConverterFactory struct {
 	// webhookConverterFactory is the factory for webhook converters.
@@ -39,7 +54,7 @@ type CRConverterFactory struct {
 // apiextensions-apiserver runs.
 var converterMetricFactorySingleton = newConverterMetricFactory()
 
-// NewCRConverterFactory creates a new CRConverterFactory
+// NewCRConverterFactory creates a new CRConverterFactory that supports none and webhook conversion strategies.
 func NewCRConverterFactory(serviceResolver webhook.ServiceResolver, authResolverWrapper webhook.AuthenticationInfoResolverWrapper) (*CRConverterFactory, error) {
 	converterFactory := &CRConverterFactory{}
 	webhookConverterFactory, err := newWebhookConverterFactory(serviceResolver, authResolverWrapper)
@@ -50,30 +65,32 @@ func NewCRConverterFactory(serviceResolver webhook.ServiceResolver, authResolver
 	return converterFactory, nil
 }
 
-// NewConverter returns a new CR converter based on the conversion settings in crd object.
-func (m *CRConverterFactory) NewConverter(crd *apiextensionsv1.CustomResourceDefinition) (safe, unsafe runtime.ObjectConvertor, err error) {
+// NewConverter creates a new CRConverter based on the crd's conversion strategy. Supported strategies are none and
+// webhook.
+func (f *CRConverterFactory) NewConverter(crd *apiextensionsv1.CustomResourceDefinition) (CRConverter, error) {
+	switch crd.Spec.Conversion.Strategy {
+	case apiextensionsv1.NoneConverter:
+		return NewNOPConverter(), nil
+	case apiextensionsv1.WebhookConverter:
+		converter, err := f.webhookConverterFactory.NewWebhookConverter(crd)
+		if err != nil {
+			return nil, err
+		}
+		return converterMetricFactorySingleton.addMetrics(crd.Name, converter)
+	}
+
+	return nil, fmt.Errorf("unknown conversion strategy %q for CRD %s", crd.Spec.Conversion.Strategy, crd.Name)
+}
+
+// NewDelegatingConverter returns new safe and unsafe converters based on the conversion settings in
+// crd. These converters contain logic common to all converters, and they delegate the actual
+// specific version-to-version conversion logic to the delegate.
+func NewDelegatingConverter(crd *apiextensionsv1.CustomResourceDefinition, delegate CRConverter) (safe, unsafe runtime.ObjectConvertor, err error) {
 	validVersions := map[schema.GroupVersion]bool{}
 	for _, version := range crd.Spec.Versions {
 		validVersions[schema.GroupVersion{Group: crd.Spec.Group, Version: version.Name}] = true
 	}
 
-	var converter crConverterInterface
-	switch crd.Spec.Conversion.Strategy {
-	case apiextensionsv1.NoneConverter:
-		converter = &nopConverter{}
-	case apiextensionsv1.WebhookConverter:
-		converter, err = m.webhookConverterFactory.NewWebhookConverter(crd)
-		if err != nil {
-			return nil, nil, err
-		}
-		converter, err = converterMetricFactorySingleton.addMetrics(crd.Name, converter)
-		if err != nil {
-			return nil, nil, err
-		}
-	default:
-		return nil, nil, fmt.Errorf("unknown conversion strategy %q for CRD %s", crd.Spec.Conversion.Strategy, crd.Name)
-	}
-
 	// Determine whether we should expect to be asked to "convert" autoscaling/v1 Scale types
 	convertScale := false
 	for _, version := range crd.Spec.Versions {
@@ -82,33 +99,33 @@ func (m *CRConverterFactory) NewConverter(crd *apiextensionsv1.CustomResourceDef
 		}
 	}
 
-	unsafe = &crConverter{
+	unsafe = &delegatingCRConverter{
 		convertScale:  convertScale,
 		validVersions: validVersions,
 		clusterScoped: crd.Spec.Scope == apiextensionsv1.ClusterScoped,
-		converter:     converter,
+		converter:     delegate,
 	}
 	return &safeConverterWrapper{unsafe}, unsafe, nil
 }
 
-// crConverterInterface is the interface all cr converters must implement
-type crConverterInterface interface {
+// CRConverter is the interface all CR converters must implement
+type CRConverter interface {
 	// Convert converts in object to the given gvk and returns the converted object.
 	// Note that the function may mutate in object and return it. A safe wrapper will make sure
 	// a safe converter will be returned.
-	Convert(in runtime.Object, targetGVK schema.GroupVersion) (runtime.Object, error)
+	Convert(in *unstructured.UnstructuredList, targetGVK schema.GroupVersion) (*unstructured.UnstructuredList, error)
 }
 
-// crConverter extends the delegate converter with generic CR conversion behaviour. The delegate will implement the
+// delegatingCRConverter extends the delegate converter with generic CR conversion behaviour. The delegate will implement the
 // user defined conversion strategy given in the CustomResourceDefinition.
-type crConverter struct {
+type delegatingCRConverter struct {
 	convertScale  bool
-	converter     crConverterInterface
+	converter     CRConverter
 	validVersions map[schema.GroupVersion]bool
 	clusterScoped bool
 }
 
-func (c *crConverter) ConvertFieldLabel(gvk schema.GroupVersionKind, label, value string) (string, string, error) {
+func (c *delegatingCRConverter) ConvertFieldLabel(gvk schema.GroupVersionKind, label, value string) (string, string, error) {
 	// We currently only support metadata.namespace and metadata.name.
 	switch {
 	case label == "metadata.name":
@@ -120,7 +137,7 @@ func (c *crConverter) ConvertFieldLabel(gvk schema.GroupVersionKind, label, valu
 	}
 }
 
-func (c *crConverter) Convert(in, out, context interface{}) error {
+func (c *delegatingCRConverter) Convert(in, out, context interface{}) error {
 	// Special-case typed scale conversion if this custom resource supports a scale endpoint
 	if c.convertScale {
 		_, isInScale := in.(*autoscalingv1.Scale)
@@ -158,13 +175,7 @@ func (c *crConverter) Convert(in, out, context interface{}) error {
 // The in object can be a single object or a UnstructuredList. CRD storage implementation creates an
 // UnstructuredList with the request's GV, populates it from storage, then calls conversion to convert
 // the individual items. This function assumes it never gets a v1.List.
-func (c *crConverter) ConvertToVersion(in runtime.Object, target runtime.GroupVersioner) (runtime.Object, error) {
-	fromGVK := in.GetObjectKind().GroupVersionKind()
-	toGVK, ok := target.KindForGroupVersionKinds([]schema.GroupVersionKind{fromGVK})
-	if !ok {
-		// TODO: should this be a typed error?
-		return nil, fmt.Errorf("%v is unstructured and is not suitable for converting to %q", fromGVK.String(), target)
-	}
+func (c *delegatingCRConverter) ConvertToVersion(in runtime.Object, target runtime.GroupVersioner) (runtime.Object, error) {
 	// Special-case typed scale conversion if this custom resource supports a scale endpoint
 	if c.convertScale {
 		if _, isInScale := in.(*autoscalingv1.Scale); isInScale {
@@ -172,8 +183,28 @@ func (c *crConverter) ConvertToVersion(in runtime.Object, target runtime.GroupVe
 		}
 	}
 
+	fromGVK := in.GetObjectKind().GroupVersionKind()
+	toGVK, ok := target.KindForGroupVersionKinds([]schema.GroupVersionKind{fromGVK})
+	if !ok {
+		// TODO: should this be a typed error?
+		return nil, fmt.Errorf("%v is unstructured and is not suitable for converting to %q", fromGVK.String(), target)
+	}
+
+	isList := false
+	var list *unstructured.UnstructuredList
+	switch t := in.(type) {
+	case *unstructured.Unstructured:
+		list = &unstructured.UnstructuredList{Items: []unstructured.Unstructured{*t}}
+	case *unstructured.UnstructuredList:
+		list = t
+		isList = true
+	default:
+		return nil, fmt.Errorf("unexpected type %T", in)
+	}
+
+	desiredAPIVersion := toGVK.GroupVersion().String()
 	if !c.validVersions[toGVK.GroupVersion()] {
-		return nil, fmt.Errorf("request to convert CR to an invalid group/version: %s", toGVK.GroupVersion().String())
+		return nil, fmt.Errorf("request to convert CR to an invalid group/version: %s", desiredAPIVersion)
 	}
 	// Note that even if the request is for a list, the GV of the request UnstructuredList is what
 	// is expected to convert to. As mentioned in the function's document, it is not expected to
@@ -181,16 +212,101 @@ func (c *crConverter) ConvertToVersion(in runtime.Object, target runtime.GroupVe
 	if !c.validVersions[fromGVK.GroupVersion()] {
 		return nil, fmt.Errorf("request to convert CR from an invalid group/version: %s", fromGVK.GroupVersion().String())
 	}
-	// Check list item's apiVersion
-	if list, ok := in.(*unstructured.UnstructuredList); ok {
-		for i := range list.Items {
-			expectedGV := list.Items[i].GroupVersionKind().GroupVersion()
-			if !c.validVersions[expectedGV] {
-				return nil, fmt.Errorf("request to convert CR list failed, list index %d has invalid group/version: %s", i, expectedGV.String())
-			}
+
+	var objectsToConvert []*unstructured.Unstructured
+	objectsToConvert, err := getObjectsToConvert(list, desiredAPIVersion, c.validVersions)
+	if err != nil {
+		return nil, err
+	}
+
+	objCount := len(objectsToConvert)
+	if objCount == 0 {
+		// no objects needed conversion
+		if !isList {
+			// for a single item, return as-is
+			return in, nil
+		}
+		// for a list, set the version of the top-level list object (all individual objects are already in the correct version)
+		out := list.DeepCopy()
+		out.SetAPIVersion(desiredAPIVersion)
+		return out, nil
+	}
+
+	// A smoke test in API machinery calls the converter on empty objects during startup. The test is initiated here:
+	// https://github.com/kubernetes/kubernetes/blob/dbb448bbdcb9e440eee57024ffa5f1698956a054/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go#L201
+	if isEmptyUnstructuredObject(in) {
+		converted, err := NewNOPConverter().Convert(list, toGVK.GroupVersion())
+		if err != nil {
+			return nil, err
+		}
+		if !isList {
+			return &converted.Items[0], nil
+		}
+		return converted, nil
+	}
+
+	convertedObjects, err := c.converter.Convert(list, toGVK.GroupVersion())
+	if err != nil {
+		return nil, fmt.Errorf("conversion for %v failed: %w", in.GetObjectKind().GroupVersionKind(), err)
+	}
+	if len(convertedObjects.Items) != len(objectsToConvert) {
+		return nil, fmt.Errorf("conversion for %v returned %d objects, expected %d", in.GetObjectKind().GroupVersionKind(), len(convertedObjects.Items), len(objectsToConvert))
+	}
+
+	// start a deepcopy of the input and fill in the converted objects from the response at the right spots.
+	// The response list might be sparse because objects had the right version already.
+	convertedList := list.DeepCopy()
+	convertedList.SetAPIVersion(desiredAPIVersion)
+	convertedIndex := 0
+	for i := range convertedList.Items {
+		original := &convertedList.Items[i]
+		if original.GetAPIVersion() == desiredAPIVersion {
+			// This item has not been sent for conversion, and therefore does not show up in the response.
+			// convertedList has the right item already.
+			continue
+		}
+		converted := &convertedObjects.Items[convertedIndex]
+		convertedIndex++
+		if expected, got := toGVK.GroupVersion(), converted.GetObjectKind().GroupVersionKind().GroupVersion(); expected != got {
+			return nil, fmt.Errorf("conversion for %v returned invalid converted object at index %v: invalid groupVersion (expected %v, received %v)", in.GetObjectKind().GroupVersionKind(), convertedIndex, expected, got)
+		}
+		if expected, got := original.GetObjectKind().GroupVersionKind().Kind, converted.GetObjectKind().GroupVersionKind().Kind; expected != got {
+			return nil, fmt.Errorf("conversion for %v returned invalid converted object at index %v: invalid kind (expected %v, received %v)", in.GetObjectKind().GroupVersionKind(), convertedIndex, expected, got)
+		}
+		if err := validateConvertedObject(original, converted); err != nil {
+			return nil, fmt.Errorf("conversion for %v returned invalid converted object at index %v: %v", in.GetObjectKind().GroupVersionKind(), convertedIndex, err)
+		}
+		if err := restoreObjectMeta(original, converted); err != nil {
+			return nil, fmt.Errorf("conversion for %v returned invalid metadata in object at index %v: %v", in.GetObjectKind().GroupVersionKind(), convertedIndex, err)
+		}
+		convertedList.Items[i] = *converted
+	}
+
+	if isList {
+		return convertedList, nil
+	}
+
+	return &convertedList.Items[0], nil
+}
+
+func getObjectsToConvert(
+	list *unstructured.UnstructuredList,
+	desiredAPIVersion string,
+	validVersions map[schema.GroupVersion]bool,
+) ([]*unstructured.Unstructured, error) {
+	var objectsToConvert []*unstructured.Unstructured
+	for i := range list.Items {
+		expectedGV := list.Items[i].GroupVersionKind().GroupVersion()
+		if !validVersions[expectedGV] {
+			return nil, fmt.Errorf("request to convert CR list failed, list index %d has invalid group/version: %s", i, expectedGV.String())
+		}
+
+		// Only sent item for conversion, if the apiVersion is different
+		if list.Items[i].GetAPIVersion() != desiredAPIVersion {
+			objectsToConvert = append(objectsToConvert, &list.Items[i])
 		}
 	}
-	return c.converter.Convert(in, toGVK.GroupVersion())
+	return objectsToConvert, nil
 }
 
 // safeConverterWrapper is a wrapper over an unsafe object converter that makes copy of the input and then delegate to the unsafe converter.
@@ -218,3 +334,113 @@ func (c *safeConverterWrapper) Convert(in, out, context interface{}) error {
 func (c *safeConverterWrapper) ConvertToVersion(in runtime.Object, target runtime.GroupVersioner) (runtime.Object, error) {
 	return c.unsafe.ConvertToVersion(in.DeepCopyObject(), target)
 }
+
+// isEmptyUnstructuredObject returns true if in is an empty unstructured object, i.e. an unstructured object that does
+// not have any field except apiVersion and kind.
+func isEmptyUnstructuredObject(in runtime.Object) bool {
+	u, ok := in.(*unstructured.Unstructured)
+	if !ok {
+		return false
+	}
+	if len(u.Object) != 2 {
+		return false
+	}
+	if _, ok := u.Object["kind"]; !ok {
+		return false
+	}
+	if _, ok := u.Object["apiVersion"]; !ok {
+		return false
+	}
+	return true
+}
+
+// validateConvertedObject checks that ObjectMeta fields match, with the exception of
+// labels and annotations.
+func validateConvertedObject(in, out *unstructured.Unstructured) error {
+	if e, a := in.GetKind(), out.GetKind(); e != a {
+		return fmt.Errorf("must have the same kind: %v != %v", e, a)
+	}
+	if e, a := in.GetName(), out.GetName(); e != a {
+		return fmt.Errorf("must have the same name: %v != %v", e, a)
+	}
+	if e, a := in.GetNamespace(), out.GetNamespace(); e != a {
+		return fmt.Errorf("must have the same namespace: %v != %v", e, a)
+	}
+	if e, a := in.GetUID(), out.GetUID(); e != a {
+		return fmt.Errorf("must have the same UID: %v != %v", e, a)
+	}
+	return nil
+}
+
+// restoreObjectMeta deep-copies metadata from original into converted, while preserving labels and annotations from converted.
+func restoreObjectMeta(original, converted *unstructured.Unstructured) error {
+	obj, found := converted.Object["metadata"]
+	if !found {
+		return fmt.Errorf("missing metadata in converted object")
+	}
+	responseMetaData, ok := obj.(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid metadata of type %T in converted object", obj)
+	}
+
+	if _, ok := original.Object["metadata"]; !ok {
+		// the original will always have metadata. But just to be safe, let's clear in converted
+		// with an empty object instead of nil, to be able to add labels and annotations below.
+		converted.Object["metadata"] = map[string]interface{}{}
+	} else {
+		converted.Object["metadata"] = runtime.DeepCopyJSONValue(original.Object["metadata"])
+	}
+
+	obj = converted.Object["metadata"]
+	convertedMetaData, ok := obj.(map[string]interface{})
+	if !ok {
+		return fmt.Errorf("invalid metadata of type %T in input object", obj)
+	}
+
+	for _, fld := range []string{"labels", "annotations"} {
+		obj, found := responseMetaData[fld]
+		if !found || obj == nil {
+			delete(convertedMetaData, fld)
+			continue
+		}
+		responseField, ok := obj.(map[string]interface{})
+		if !ok {
+			return fmt.Errorf("invalid metadata.%s of type %T in converted object", fld, obj)
+		}
+
+		originalField, ok := convertedMetaData[fld].(map[string]interface{})
+		if !ok && convertedMetaData[fld] != nil {
+			return fmt.Errorf("invalid metadata.%s of type %T in original object", fld, convertedMetaData[fld])
+		}
+
+		somethingChanged := len(originalField) != len(responseField)
+		for k, v := range responseField {
+			if _, ok := v.(string); !ok {
+				return fmt.Errorf("metadata.%s[%s] must be a string, but is %T in converted object", fld, k, v)
+			}
+			if originalField[k] != interface{}(v) {
+				somethingChanged = true
+			}
+		}
+
+		if somethingChanged {
+			stringMap := make(map[string]string, len(responseField))
+			for k, v := range responseField {
+				stringMap[k] = v.(string)
+			}
+			var errs field.ErrorList
+			if fld == "labels" {
+				errs = metav1validation.ValidateLabels(stringMap, field.NewPath("metadata", "labels"))
+			} else {
+				errs = apivalidation.ValidateAnnotations(stringMap, field.NewPath("metadata", "annotation"))
+			}
+			if len(errs) > 0 {
+				return errs.ToAggregate()
+			}
+		}
+
+		convertedMetaData[fld] = responseField
+	}
+
+	return nil
+}
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter_test.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter_test.go
index 3866dbd36bf..8c60cb7f0da 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter_test.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/converter_test.go
@@ -21,11 +21,11 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apiserver/pkg/util/webhook"
 )
 
 func TestConversion(t *testing.T) {
@@ -154,10 +154,6 @@ func TestConversion(t *testing.T) {
 		},
 	}
 
-	CRConverterFactory, err := NewCRConverterFactory(nil, func(resolver webhook.AuthenticationInfoResolver) webhook.AuthenticationInfoResolver { return nil })
-	if err != nil {
-		t.Fatalf("Cannot create conversion factory: %v", err)
-	}
 	for _, test := range tests {
 		testCRD := apiextensionsv1.CustomResourceDefinition{
 			Spec: apiextensionsv1.CustomResourceDefinitionSpec{
@@ -171,7 +167,7 @@ func TestConversion(t *testing.T) {
 			testCRD.Spec.Versions = append(testCRD.Spec.Versions, apiextensionsv1.CustomResourceDefinitionVersion{Name: gv.Version, Served: true})
 			testCRD.Spec.Group = gv.Group
 		}
-		safeConverter, _, err := CRConverterFactory.NewConverter(&testCRD)
+		safeConverter, _, err := NewDelegatingConverter(&testCRD, NewNOPConverter())
 		if err != nil {
 			t.Fatalf("Cannot create converter: %v", err)
 		}
@@ -193,3 +189,102 @@ func TestConversion(t *testing.T) {
 		}
 	}
 }
+
+func TestGetObjectsToConvert(t *testing.T) {
+	v1Object := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v1", "kind": "Widget", "metadata": map[string]interface{}{"name": "myv1"}}}
+	v2Object := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v2", "kind": "Widget", "metadata": map[string]interface{}{"name": "myv2"}}}
+	v3Object := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v3", "kind": "Widget", "metadata": map[string]interface{}{"name": "myv3"}}}
+
+	testcases := []struct {
+		Name          string
+		List          *unstructured.UnstructuredList
+		APIVersion    string
+		ValidVersions map[schema.GroupVersion]bool
+
+		ExpectObjects []*unstructured.Unstructured
+		ExpectError   bool
+	}{
+		{
+			Name:       "empty list",
+			List:       &unstructured.UnstructuredList{},
+			APIVersion: "foo/v1",
+			ValidVersions: map[schema.GroupVersion]bool{
+				{Group: "foo", Version: "v1"}: true,
+			},
+			ExpectObjects: nil,
+		},
+		{
+			Name: "one-item list, in desired version",
+			List: &unstructured.UnstructuredList{
+				Items: []unstructured.Unstructured{*v1Object},
+			},
+			ValidVersions: map[schema.GroupVersion]bool{
+				{Group: "foo", Version: "v1"}: true,
+			},
+			APIVersion:    "foo/v1",
+			ExpectObjects: nil,
+		},
+		{
+			Name: "one-item list, not in desired version",
+			List: &unstructured.UnstructuredList{
+				Items: []unstructured.Unstructured{*v2Object},
+			},
+			ValidVersions: map[schema.GroupVersion]bool{
+				{Group: "foo", Version: "v1"}: true,
+				{Group: "foo", Version: "v2"}: true,
+			},
+			APIVersion:    "foo/v1",
+			ExpectObjects: []*unstructured.Unstructured{v2Object},
+		},
+		{
+			Name: "multi-item list, in desired version",
+			List: &unstructured.UnstructuredList{
+				Items: []unstructured.Unstructured{*v1Object, *v1Object, *v1Object},
+			},
+			ValidVersions: map[schema.GroupVersion]bool{
+				{Group: "foo", Version: "v1"}: true,
+				{Group: "foo", Version: "v2"}: true,
+			},
+			APIVersion:    "foo/v1",
+			ExpectObjects: nil,
+		},
+		{
+			Name: "multi-item list, mixed versions",
+			List: &unstructured.UnstructuredList{
+				Items: []unstructured.Unstructured{*v1Object, *v2Object, *v3Object},
+			},
+			ValidVersions: map[schema.GroupVersion]bool{
+				{Group: "foo", Version: "v1"}: true,
+				{Group: "foo", Version: "v2"}: true,
+				{Group: "foo", Version: "v3"}: true,
+			},
+			APIVersion:    "foo/v1",
+			ExpectObjects: []*unstructured.Unstructured{v2Object, v3Object},
+		},
+		{
+			Name: "multi-item list, invalid versions",
+			List: &unstructured.UnstructuredList{
+				Items: []unstructured.Unstructured{*v1Object, *v2Object, *v3Object},
+			},
+			ValidVersions: map[schema.GroupVersion]bool{
+				{Group: "foo", Version: "v2"}: true,
+				{Group: "foo", Version: "v3"}: true,
+			},
+			APIVersion:    "foo/v1",
+			ExpectObjects: nil,
+			ExpectError:   true,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.Name, func(t *testing.T) {
+			objects, err := getObjectsToConvert(tc.List, tc.APIVersion, tc.ValidVersions)
+			gotError := err != nil
+			if e, a := tc.ExpectError, gotError; e != a {
+				t.Fatalf("error: expected %t, got %t", e, a)
+			}
+			if !reflect.DeepEqual(objects, tc.ExpectObjects) {
+				t.Errorf("unexpected diff: %s", cmp.Diff(tc.ExpectObjects, objects))
+			}
+		})
+	}
+}
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/metrics.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/metrics.go
index 744046b5b47..b1f68723576 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/metrics.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/metrics.go
@@ -21,7 +21,7 @@ import (
 	"sync"
 	"time"
 
-	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/component-base/metrics"
 	"k8s.io/component-base/metrics/legacyregistry"
@@ -43,15 +43,15 @@ func newConverterMetricFactory() *converterMetricFactory {
 	return &converterMetricFactory{durations: map[string]*metrics.HistogramVec{}, factoryLock: sync.Mutex{}}
 }
 
-var _ crConverterInterface = &converterMetric{}
+var _ CRConverter = &converterMetric{}
 
 type converterMetric struct {
-	delegate  crConverterInterface
+	delegate  CRConverter
 	latencies *metrics.HistogramVec
 	crdName   string
 }
 
-func (c *converterMetricFactory) addMetrics(crdName string, converter crConverterInterface) (crConverterInterface, error) {
+func (c *converterMetricFactory) addMetrics(crdName string, converter CRConverter) (CRConverter, error) {
 	c.factoryLock.Lock()
 	defer c.factoryLock.Unlock()
 	metric, exists := c.durations["webhook"]
@@ -73,7 +73,7 @@ func (c *converterMetricFactory) addMetrics(crdName string, converter crConverte
 	return &converterMetric{latencies: metric, delegate: converter, crdName: crdName}, nil
 }
 
-func (m *converterMetric) Convert(in runtime.Object, targetGV schema.GroupVersion) (runtime.Object, error) {
+func (m *converterMetric) Convert(in *unstructured.UnstructuredList, targetGV schema.GroupVersion) (*unstructured.UnstructuredList, error) {
 	start := time.Now()
 	obj, err := m.delegate.Convert(in, targetGV)
 	fromVersion := in.GetObjectKind().GroupVersionKind().Version
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/nop_converter.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/nop_converter.go
index 8254fdfc0b9..42a27ec1c10 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/nop_converter.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/nop_converter.go
@@ -18,7 +18,6 @@ package conversion
 
 import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
@@ -26,16 +25,18 @@ import (
 type nopConverter struct {
 }
 
-var _ crConverterInterface = &nopConverter{}
+// NewNOPConverter creates a new no-op converter. The only "conversion" it performs is to set the group and version to
+// targetGV.
+func NewNOPConverter() *nopConverter {
+	return &nopConverter{}
+}
+
+var _ CRConverter = &nopConverter{}
 
 // ConvertToVersion converts in object to the given gv in place and returns the same `in` object.
-func (c *nopConverter) Convert(in runtime.Object, targetGV schema.GroupVersion) (runtime.Object, error) {
-	// Run the converter on the list items instead of list itself
-	if list, ok := in.(*unstructured.UnstructuredList); ok {
-		for i := range list.Items {
-			list.Items[i].SetGroupVersionKind(targetGV.WithKind(list.Items[i].GroupVersionKind().Kind))
-		}
+func (c *nopConverter) Convert(list *unstructured.UnstructuredList, targetGV schema.GroupVersion) (*unstructured.UnstructuredList, error) {
+	for i := range list.Items {
+		list.Items[i].SetGroupVersionKind(targetGV.WithKind(list.Items[i].GroupVersionKind().Kind))
 	}
-	in.GetObjectKind().SetGroupVersionKind(targetGV.WithKind(in.GetObjectKind().GroupVersionKind().Kind))
-	return in, nil
+	return list, nil
 }
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter.go
index e72454931fa..8cdfd065f31 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter.go
@@ -26,15 +26,12 @@ import (
 
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
-	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	metav1validation "k8s.io/apimachinery/pkg/apis/meta/v1/validation"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/uuid"
-	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apiserver/pkg/util/webhook"
 	"k8s.io/client-go/rest"
 	"k8s.io/component-base/tracing"
@@ -96,7 +93,7 @@ func webhookClientConfigForCRD(crd *v1.CustomResourceDefinition) *webhook.Client
 	return &ret
 }
 
-var _ crConverterInterface = &webhookConverter{}
+var _ CRConverter = &webhookConverter{}
 
 func (f *webhookConverterFactory) NewWebhookConverter(crd *v1.CustomResourceDefinition) (*webhookConverter, error) {
 	restClient, err := f.clientManager.HookClient(*webhookClientConfigForCRD(crd))
@@ -113,36 +110,21 @@ func (f *webhookConverterFactory) NewWebhookConverter(crd *v1.CustomResourceDefi
 	}, nil
 }
 
-// getObjectsToConvert returns a list of objects requiring conversion.
-// if obj is a list, getObjectsToConvert returns a (potentially empty) list of the items that are not already in the desired version.
-// if obj is not a list, and is already in the desired version, getObjectsToConvert returns an empty list.
-// if obj is not a list, and is not already in the desired version, getObjectsToConvert returns a list containing only obj.
-func getObjectsToConvert(obj runtime.Object, apiVersion string) []runtime.RawExtension {
-	listObj, isList := obj.(*unstructured.UnstructuredList)
-	var objects []runtime.RawExtension
-	if isList {
-		for i := range listObj.Items {
-			// Only sent item for conversion, if the apiVersion is different
-			if listObj.Items[i].GetAPIVersion() != apiVersion {
-				objects = append(objects, runtime.RawExtension{Object: &listObj.Items[i]})
-			}
-		}
-	} else {
-		if obj.GetObjectKind().GroupVersionKind().GroupVersion().String() != apiVersion {
-			objects = []runtime.RawExtension{{Object: obj}}
+// createConversionReviewObjects returns ConversionReview request and response objects for the first supported version found in conversionReviewVersions.
+func createConversionReviewObjects(conversionReviewVersions []string, objects *unstructured.UnstructuredList, apiVersion string, requestUID types.UID) (request, response runtime.Object, err error) {
+	rawObjects := make([]runtime.RawExtension, len(objects.Items))
+	for i := range objects.Items {
+		rawObjects[i] = runtime.RawExtension{
+			Object: &objects.Items[i],
 		}
 	}
-	return objects
-}
 
-// createConversionReviewObjects returns ConversionReview request and response objects for the first supported version found in conversionReviewVersions.
-func createConversionReviewObjects(conversionReviewVersions []string, objects []runtime.RawExtension, apiVersion string, requestUID types.UID) (request, response runtime.Object, err error) {
 	for _, version := range conversionReviewVersions {
 		switch version {
 		case v1beta1.SchemeGroupVersion.Version:
 			return &v1beta1.ConversionReview{
 				Request: &v1beta1.ConversionRequest{
-					Objects:           objects,
+					Objects:           rawObjects,
 					DesiredAPIVersion: apiVersion,
 					UID:               requestUID,
 				},
@@ -151,7 +133,7 @@ func createConversionReviewObjects(conversionReviewVersions []string, objects []
 		case v1.SchemeGroupVersion.Version:
 			return &v1.ConversionReview{
 				Request: &v1.ConversionRequest{
-					Objects:           objects,
+					Objects:           rawObjects,
 					DesiredAPIVersion: apiVersion,
 					UID:               requestUID,
 				},
@@ -162,9 +144,13 @@ func createConversionReviewObjects(conversionReviewVersions []string, objects []
 	return nil, nil, fmt.Errorf("no supported conversion review versions")
 }
 
-func getRawExtensionObject(rx runtime.RawExtension) (runtime.Object, error) {
+func getRawExtensionObject(rx runtime.RawExtension) (*unstructured.Unstructured, error) {
 	if rx.Object != nil {
-		return rx.Object, nil
+		u, ok := rx.Object.(*unstructured.Unstructured)
+		if !ok {
+			return nil, fmt.Errorf("unexpected type %T", rx.Object)
+		}
+		return u, nil
 	}
 	u := unstructured.Unstructured{}
 	err := u.UnmarshalJSON(rx.Raw)
@@ -227,39 +213,16 @@ func getConvertedObjectsFromResponse(expectedUID types.UID, response runtime.Obj
 	}
 }
 
-func (c *webhookConverter) Convert(in runtime.Object, toGV schema.GroupVersion) (runtime.Object, error) {
+func (c *webhookConverter) Convert(in *unstructured.UnstructuredList, toGV schema.GroupVersion) (*unstructured.UnstructuredList, error) {
 	ctx := context.TODO()
-	// In general, the webhook should not do any defaulting or validation. A special case of that is an empty object
-	// conversion that must result an empty object and practically is the same as nopConverter.
-	// A smoke test in API machinery calls the converter on empty objects. As this case happens consistently
-	// it special cased here not to call webhook converter. The test initiated here:
-	// https://github.com/kubernetes/kubernetes/blob/dbb448bbdcb9e440eee57024ffa5f1698956a054/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go#L201
-	if isEmptyUnstructuredObject(in) {
-		return c.nopConverter.Convert(in, toGV)
-	}
-
-	listObj, isList := in.(*unstructured.UnstructuredList)
-
 	requestUID := uuid.NewUUID()
 	desiredAPIVersion := toGV.String()
-	objectsToConvert := getObjectsToConvert(in, desiredAPIVersion)
-	request, response, err := createConversionReviewObjects(c.conversionReviewVersions, objectsToConvert, desiredAPIVersion, requestUID)
+	request, response, err := createConversionReviewObjects(c.conversionReviewVersions, in, desiredAPIVersion, requestUID)
 	if err != nil {
 		return nil, err
 	}
 
-	objCount := len(objectsToConvert)
-	if objCount == 0 {
-		// no objects needed conversion
-		if !isList {
-			// for a single item, return as-is
-			return in, nil
-		}
-		// for a list, set the version of the top-level list object (all individual objects are already in the correct version)
-		out := listObj.DeepCopy()
-		out.SetAPIVersion(toGV.String())
-		return out, nil
-	}
+	objCount := len(in.Items)
 
 	ctx, span := tracing.Start(ctx, "Call conversion webhook",
 		attribute.String("custom-resource-definition", c.name),
@@ -284,189 +247,14 @@ func (c *webhookConverter) Convert(in runtime.Object, toGV schema.GroupVersion)
 		return nil, fmt.Errorf("conversion webhook for %v failed: %v", in.GetObjectKind().GroupVersionKind(), err)
 	}
 
-	if len(convertedObjects) != len(objectsToConvert) {
-		return nil, fmt.Errorf("conversion webhook for %v returned %d objects, expected %d", in.GetObjectKind().GroupVersionKind(), len(convertedObjects), len(objectsToConvert))
-	}
-
-	if isList {
-		// start a deepcopy of the input and fill in the converted objects from the response at the right spots.
-		// The response list might be sparse because objects had the right version already.
-		convertedList := listObj.DeepCopy()
-		convertedIndex := 0
-		for i := range convertedList.Items {
-			original := &convertedList.Items[i]
-			if original.GetAPIVersion() == toGV.String() {
-				// This item has not been sent for conversion, and therefore does not show up in the response.
-				// convertedList has the right item already.
-				continue
-			}
-			converted, err := getRawExtensionObject(convertedObjects[convertedIndex])
-			if err != nil {
-				return nil, fmt.Errorf("conversion webhook for %v returned invalid converted object at index %v: %v", in.GetObjectKind().GroupVersionKind(), convertedIndex, err)
-			}
-			convertedIndex++
-			if expected, got := toGV, converted.GetObjectKind().GroupVersionKind().GroupVersion(); expected != got {
-				return nil, fmt.Errorf("conversion webhook for %v returned invalid converted object at index %v: invalid groupVersion (expected %v, received %v)", in.GetObjectKind().GroupVersionKind(), convertedIndex, expected, got)
-			}
-			if expected, got := original.GetObjectKind().GroupVersionKind().Kind, converted.GetObjectKind().GroupVersionKind().Kind; expected != got {
-				return nil, fmt.Errorf("conversion webhook for %v returned invalid converted object at index %v: invalid kind (expected %v, received %v)", in.GetObjectKind().GroupVersionKind(), convertedIndex, expected, got)
-			}
-			unstructConverted, ok := converted.(*unstructured.Unstructured)
-			if !ok {
-				// this should not happened
-				return nil, fmt.Errorf("conversion webhook for %v returned invalid converted object at index %v: invalid type, expected=Unstructured, got=%T", in.GetObjectKind().GroupVersionKind(), convertedIndex, converted)
-			}
-			if err := validateConvertedObject(original, unstructConverted); err != nil {
-				return nil, fmt.Errorf("conversion webhook for %v returned invalid converted object at index %v: %v", in.GetObjectKind().GroupVersionKind(), convertedIndex, err)
-			}
-			if err := restoreObjectMeta(original, unstructConverted); err != nil {
-				return nil, fmt.Errorf("conversion webhook for %v returned invalid metadata in object at index %v: %v", in.GetObjectKind().GroupVersionKind(), convertedIndex, err)
-			}
-			convertedList.Items[i] = *unstructConverted
+	out := &unstructured.UnstructuredList{}
+	out.Items = make([]unstructured.Unstructured, len(convertedObjects))
+	for i := range convertedObjects {
+		u, err := getRawExtensionObject(convertedObjects[i])
+		if err != nil {
+			return nil, err
 		}
-		convertedList.SetAPIVersion(toGV.String())
-		return convertedList, nil
+		out.Items[i] = *u
 	}
-
-	if len(convertedObjects) != 1 {
-		// This should not happened
-		return nil, fmt.Errorf("conversion webhook for %v failed, no objects returned", in.GetObjectKind())
-	}
-	converted, err := getRawExtensionObject(convertedObjects[0])
-	if err != nil {
-		return nil, err
-	}
-	if e, a := toGV, converted.GetObjectKind().GroupVersionKind().GroupVersion(); e != a {
-		return nil, fmt.Errorf("conversion webhook for %v returned invalid object at index 0: invalid groupVersion (expected %v, received %v)", in.GetObjectKind().GroupVersionKind(), e, a)
-	}
-	if e, a := in.GetObjectKind().GroupVersionKind().Kind, converted.GetObjectKind().GroupVersionKind().Kind; e != a {
-		return nil, fmt.Errorf("conversion webhook for %v returned invalid object at index 0: invalid kind (expected %v, received %v)", in.GetObjectKind().GroupVersionKind(), e, a)
-	}
-	unstructConverted, ok := converted.(*unstructured.Unstructured)
-	if !ok {
-		// this should not happened
-		return nil, fmt.Errorf("conversion webhook for %v failed, unexpected type %T at index 0", in.GetObjectKind().GroupVersionKind(), converted)
-	}
-	unstructIn, ok := in.(*unstructured.Unstructured)
-	if !ok {
-		// this should not happened
-		return nil, fmt.Errorf("conversion webhook for %v failed unexpected input type %T", in.GetObjectKind().GroupVersionKind(), in)
-	}
-	if err := validateConvertedObject(unstructIn, unstructConverted); err != nil {
-		return nil, fmt.Errorf("conversion webhook for %v returned invalid object: %v", in.GetObjectKind().GroupVersionKind(), err)
-	}
-	if err := restoreObjectMeta(unstructIn, unstructConverted); err != nil {
-		return nil, fmt.Errorf("conversion webhook for %v returned invalid metadata: %v", in.GetObjectKind().GroupVersionKind(), err)
-	}
-	return converted, nil
-}
-
-// validateConvertedObject checks that ObjectMeta fields match, with the exception of
-// labels and annotations.
-func validateConvertedObject(in, out *unstructured.Unstructured) error {
-	if e, a := in.GetKind(), out.GetKind(); e != a {
-		return fmt.Errorf("must have the same kind: %v != %v", e, a)
-	}
-	if e, a := in.GetName(), out.GetName(); e != a {
-		return fmt.Errorf("must have the same name: %v != %v", e, a)
-	}
-	if e, a := in.GetNamespace(), out.GetNamespace(); e != a {
-		return fmt.Errorf("must have the same namespace: %v != %v", e, a)
-	}
-	if e, a := in.GetUID(), out.GetUID(); e != a {
-		return fmt.Errorf("must have the same UID: %v != %v", e, a)
-	}
-	return nil
-}
-
-// restoreObjectMeta deep-copies metadata from original into converted, while preserving labels and annotations from converted.
-func restoreObjectMeta(original, converted *unstructured.Unstructured) error {
-	obj, found := converted.Object["metadata"]
-	if !found {
-		return fmt.Errorf("missing metadata in converted object")
-	}
-	responseMetaData, ok := obj.(map[string]interface{})
-	if !ok {
-		return fmt.Errorf("invalid metadata of type %T in converted object", obj)
-	}
-
-	if _, ok := original.Object["metadata"]; !ok {
-		// the original will always have metadata. But just to be safe, let's clear in converted
-		// with an empty object instead of nil, to be able to add labels and annotations below.
-		converted.Object["metadata"] = map[string]interface{}{}
-	} else {
-		converted.Object["metadata"] = runtime.DeepCopyJSONValue(original.Object["metadata"])
-	}
-
-	obj = converted.Object["metadata"]
-	convertedMetaData, ok := obj.(map[string]interface{})
-	if !ok {
-		return fmt.Errorf("invalid metadata of type %T in input object", obj)
-	}
-
-	for _, fld := range []string{"labels", "annotations"} {
-		obj, found := responseMetaData[fld]
-		if !found || obj == nil {
-			delete(convertedMetaData, fld)
-			continue
-		}
-		responseField, ok := obj.(map[string]interface{})
-		if !ok {
-			return fmt.Errorf("invalid metadata.%s of type %T in converted object", fld, obj)
-		}
-
-		originalField, ok := convertedMetaData[fld].(map[string]interface{})
-		if !ok && convertedMetaData[fld] != nil {
-			return fmt.Errorf("invalid metadata.%s of type %T in original object", fld, convertedMetaData[fld])
-		}
-
-		somethingChanged := len(originalField) != len(responseField)
-		for k, v := range responseField {
-			if _, ok := v.(string); !ok {
-				return fmt.Errorf("metadata.%s[%s] must be a string, but is %T in converted object", fld, k, v)
-			}
-			if originalField[k] != interface{}(v) {
-				somethingChanged = true
-			}
-		}
-
-		if somethingChanged {
-			stringMap := make(map[string]string, len(responseField))
-			for k, v := range responseField {
-				stringMap[k] = v.(string)
-			}
-			var errs field.ErrorList
-			if fld == "labels" {
-				errs = metav1validation.ValidateLabels(stringMap, field.NewPath("metadata", "labels"))
-			} else {
-				errs = apivalidation.ValidateAnnotations(stringMap, field.NewPath("metadata", "annotation"))
-			}
-			if len(errs) > 0 {
-				return errs.ToAggregate()
-			}
-		}
-
-		convertedMetaData[fld] = responseField
-	}
-
-	return nil
-}
-
-// isEmptyUnstructuredObject returns true if in is an empty unstructured object, i.e. an unstructured object that does
-// not have any field except apiVersion and kind.
-func isEmptyUnstructuredObject(in runtime.Object) bool {
-	u, ok := in.(*unstructured.Unstructured)
-	if !ok {
-		return false
-	}
-	if len(u.Object) != 2 {
-		return false
-	}
-	if _, ok := u.Object["kind"]; !ok {
-		return false
-	}
-	if _, ok := u.Object["apiVersion"]; !ok {
-		return false
-	}
-	return true
+	return out, nil
 }
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter_test.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter_test.go
index 5c6766c09c1..c53c657a803 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter_test.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter_test.go
@@ -206,81 +206,19 @@ func TestRestoreObjectMeta(t *testing.T) {
 	}
 }
 
-func TestGetObjectsToConvert(t *testing.T) {
-	v1Object := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v1", "kind": "Widget", "metadata": map[string]interface{}{"name": "myv1"}}}
-	v2Object := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v2", "kind": "Widget", "metadata": map[string]interface{}{"name": "myv2"}}}
-	v3Object := &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v3", "kind": "Widget", "metadata": map[string]interface{}{"name": "myv3"}}}
-
-	testcases := []struct {
-		Name       string
-		Object     runtime.Object
-		APIVersion string
-
-		ExpectObjects []runtime.RawExtension
-	}{
-		{
-			Name:          "empty list",
-			Object:        &unstructured.UnstructuredList{},
-			APIVersion:    "foo/v1",
-			ExpectObjects: nil,
-		},
-		{
-			Name: "one-item list, in desired version",
-			Object: &unstructured.UnstructuredList{
-				Items: []unstructured.Unstructured{*v1Object},
-			},
-			APIVersion:    "foo/v1",
-			ExpectObjects: nil,
-		},
-		{
-			Name: "one-item list, not in desired version",
-			Object: &unstructured.UnstructuredList{
-				Items: []unstructured.Unstructured{*v2Object},
-			},
-			APIVersion:    "foo/v1",
-			ExpectObjects: []runtime.RawExtension{{Object: v2Object}},
-		},
-		{
-			Name: "multi-item list, in desired version",
-			Object: &unstructured.UnstructuredList{
-				Items: []unstructured.Unstructured{*v1Object, *v1Object, *v1Object},
-			},
-			APIVersion:    "foo/v1",
-			ExpectObjects: nil,
-		},
-		{
-			Name: "multi-item list, mixed versions",
-			Object: &unstructured.UnstructuredList{
-				Items: []unstructured.Unstructured{*v1Object, *v2Object, *v3Object},
-			},
-			APIVersion:    "foo/v1",
-			ExpectObjects: []runtime.RawExtension{{Object: v2Object}, {Object: v3Object}},
-		},
-		{
-			Name:          "single item, in desired version",
-			Object:        v1Object,
-			APIVersion:    "foo/v1",
-			ExpectObjects: nil,
-		},
-		{
-			Name:          "single item, not in desired version",
-			Object:        v2Object,
-			APIVersion:    "foo/v1",
-			ExpectObjects: []runtime.RawExtension{{Object: v2Object}},
-		},
-	}
-	for _, tc := range testcases {
-		t.Run(tc.Name, func(t *testing.T) {
-			if objects := getObjectsToConvert(tc.Object, tc.APIVersion); !reflect.DeepEqual(objects, tc.ExpectObjects) {
-				t.Errorf("unexpected diff: %s", cmp.Diff(tc.ExpectObjects, objects))
-			}
-		})
-	}
-}
-
 func TestCreateConversionReviewObjects(t *testing.T) {
-	objects := []runtime.RawExtension{
-		{Object: &unstructured.Unstructured{Object: map[string]interface{}{"apiVersion": "foo/v2", "Kind": "Widget"}}},
+	objects := &unstructured.UnstructuredList{
+		Items: []unstructured.Unstructured{
+			{
+				Object: map[string]interface{}{"apiVersion": "foo/v2", "Kind": "Widget"},
+			},
+		},
+	}
+
+	rawObjects := []runtime.RawExtension{
+		{
+			Object: &objects.Items[0],
+		},
 	}
 
 	testcases := []struct {
@@ -300,7 +238,7 @@ func TestCreateConversionReviewObjects(t *testing.T) {
 			Name:     "v1",
 			Versions: []string{"v1", "v1beta1", "v2"},
 			ExpectRequest: &v1.ConversionReview{
-				Request:  &v1.ConversionRequest{UID: "uid", DesiredAPIVersion: "foo/v1", Objects: objects},
+				Request:  &v1.ConversionRequest{UID: "uid", DesiredAPIVersion: "foo/v1", Objects: rawObjects},
 				Response: &v1.ConversionResponse{},
 			},
 			ExpectResponse: &v1.ConversionReview{},
@@ -309,7 +247,7 @@ func TestCreateConversionReviewObjects(t *testing.T) {
 			Name:     "v1beta1",
 			Versions: []string{"v1beta1", "v1", "v2"},
 			ExpectRequest: &v1beta1.ConversionReview{
-				Request:  &v1beta1.ConversionRequest{UID: "uid", DesiredAPIVersion: "foo/v1", Objects: objects},
+				Request:  &v1beta1.ConversionRequest{UID: "uid", DesiredAPIVersion: "foo/v1", Objects: rawObjects},
 				Response: &v1beta1.ConversionResponse{},
 			},
 			ExpectResponse: &v1beta1.ConversionReview{},
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler.go
index dd654ef1a0c..86bfc46641c 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler.go
@@ -72,7 +72,6 @@ import (
 	"k8s.io/apiserver/pkg/registry/generic"
 	genericfilters "k8s.io/apiserver/pkg/server/filters"
 	utilopenapi "k8s.io/apiserver/pkg/util/openapi"
-	"k8s.io/apiserver/pkg/util/webhook"
 	"k8s.io/apiserver/pkg/warning"
 	"k8s.io/client-go/scale"
 	"k8s.io/client-go/scale/scheme/autoscalingv1"
@@ -109,7 +108,7 @@ type crdHandler struct {
 	// CRD establishing process for HA clusters.
 	masterCount int
 
-	converterFactory *conversion.CRConverterFactory
+	converterFactory conversion.Factory
 
 	// so that we can do create on update.
 	authorizer authorizer.Authorizer
@@ -172,14 +171,18 @@ func NewCustomResourceDefinitionHandler(
 	restOptionsGetter generic.RESTOptionsGetter,
 	admission admission.Interface,
 	establishingController *establish.EstablishingController,
-	serviceResolver webhook.ServiceResolver,
-	authResolverWrapper webhook.AuthenticationInfoResolverWrapper,
+	converterFactory conversion.Factory,
 	masterCount int,
 	authorizer authorizer.Authorizer,
 	requestTimeout time.Duration,
 	minRequestTimeout time.Duration,
 	staticOpenAPISpec *spec.Swagger,
 	maxRequestBodyBytes int64) (*crdHandler, error) {
+
+	if converterFactory == nil {
+		return nil, fmt.Errorf("converterFactory is required")
+	}
+
 	ret := &crdHandler{
 		versionDiscoveryHandler: versionDiscoveryHandler,
 		groupDiscoveryHandler:   groupDiscoveryHandler,
@@ -189,6 +192,7 @@ func NewCustomResourceDefinitionHandler(
 		restOptionsGetter:       restOptionsGetter,
 		admission:               admission,
 		establishingController:  establishingController,
+		converterFactory:        converterFactory,
 		masterCount:             masterCount,
 		authorizer:              authorizer,
 		requestTimeout:          requestTimeout,
@@ -203,11 +207,6 @@ func NewCustomResourceDefinitionHandler(
 			ret.removeDeadStorage()
 		},
 	})
-	crConverterFactory, err := conversion.NewCRConverterFactory(serviceResolver, authResolverWrapper)
-	if err != nil {
-		return nil, err
-	}
-	ret.converterFactory = crConverterFactory
 
 	ret.customStorage.Store(crdStorageMap{})
 
@@ -690,7 +689,12 @@ func (r *crdHandler) getOrCreateServingInfoFor(uid types.UID, name string) (*crd
 		}
 	}
 
-	safeConverter, unsafeConverter, err := r.converterFactory.NewConverter(crd)
+	converter, err := r.converterFactory.NewConverter(crd)
+	if err != nil {
+		return nil, fmt.Errorf("error creating converter for %s: %w", crd.Name, err)
+	}
+
+	safeConverter, unsafeConverter, err := conversion.NewDelegatingConverter(crd, converter)
 	if err != nil {
 		return nil, err
 	}
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler_test.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler_test.go
index d4be6a9331f..46470bd2f87 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler_test.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/customresource_handler_test.go
@@ -23,13 +23,10 @@ import (
 	"errors"
 	"io"
 	"io/ioutil"
-	"net"
 	"net/http"
 	"net/http/httptest"
-	"net/url"
 	"os"
 	"path/filepath"
-	"strconv"
 	"testing"
 	"time"
 
@@ -60,7 +57,6 @@ import (
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/server/options"
 	etcd3testing "k8s.io/apiserver/pkg/storage/etcd3/testing"
-	"k8s.io/apiserver/pkg/util/webhook"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/kube-openapi/pkg/validation/spec"
 )
@@ -120,11 +116,7 @@ func TestConvertFieldLabel(t *testing.T) {
 			} else {
 				crd.Spec.Scope = apiextensionsv1.NamespaceScoped
 			}
-			f, err := conversion.NewCRConverterFactory(nil, nil)
-			if err != nil {
-				t.Fatal(err)
-			}
-			_, c, err := f.NewConverter(&crd)
+			_, c, err := conversion.NewDelegatingConverter(&crd, conversion.NewNOPConverter())
 			if err != nil {
 				t.Fatalf("Failed to create CR converter. error: %v", err)
 			}
@@ -466,6 +458,12 @@ func TestHandlerConversionWithoutWatchCache(t *testing.T) {
 	testHandlerConversion(t, false)
 }
 
+type noneConverterFactory struct{}
+
+func (f *noneConverterFactory) NewConverter(_ *apiextensionsv1.CustomResourceDefinition) (conversion.CRConverter, error) {
+	return conversion.NewNOPConverter(), nil
+}
+
 func testHandlerConversion(t *testing.T, enableWatchCache bool) {
 	cl := fake.NewSimpleClientset()
 	informers := informers.NewSharedInformerFactory(fake.NewSimpleClientset(), 0)
@@ -506,8 +504,7 @@ func testHandlerConversion(t *testing.T, enableWatchCache bool) {
 		restOptionsGetter,
 		dummyAdmissionImpl{},
 		&establish.EstablishingController{},
-		dummyServiceResolverImpl{},
-		func(r webhook.AuthenticationInfoResolver) webhook.AuthenticationInfoResolver { return r },
+		&noneConverterFactory{},
 		1,
 		dummyAuthorizerImpl{},
 		time.Minute, time.Minute, nil, 3*1024*1024)
@@ -847,12 +844,6 @@ func (dummyAuthorizerImpl) Authorize(ctx context.Context, a authorizer.Attribute
 	return authorizer.DecisionAllow, "", nil
 }
 
-type dummyServiceResolverImpl struct{}
-
-func (dummyServiceResolverImpl) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
-	return &url.URL{Scheme: "https", Host: net.JoinHostPort(name+"."+namespace+".svc", strconv.Itoa(int(port)))}, nil
-}
-
 var multiVersionFixture = &apiextensionsv1.CustomResourceDefinition{
 	ObjectMeta: metav1.ObjectMeta{Name: "multiversion.stable.example.com", UID: types.UID("12345")},
 	Spec: apiextensionsv1.CustomResourceDefinitionSpec{
diff --git a/staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/options/options.go b/staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/options/options.go
index 158b6e13844..4f7331a6828 100644
--- a/staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/options/options.go
+++ b/staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/options/options.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/spf13/pflag"
 	oteltrace "go.opentelemetry.io/otel/trace"
+	"k8s.io/apiextensions-apiserver/pkg/apiserver/conversion"
 
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
@@ -107,6 +108,14 @@ func (o CustomResourceDefinitionsServerOptions) Config() (*apiserver.Config, err
 	if err := o.APIEnablement.ApplyTo(&serverConfig.Config, apiserver.DefaultAPIResourceConfigSource(), apiserver.Scheme); err != nil {
 		return nil, err
 	}
+
+	serviceResolver := &serviceResolver{serverConfig.SharedInformerFactory.Core().V1().Services().Lister()}
+	authResolverWrapper := webhook.NewDefaultAuthenticationInfoResolverWrapper(nil, nil, serverConfig.LoopbackClientConfig, oteltrace.NewNoopTracerProvider())
+	conversionFactory, err := conversion.NewCRConverterFactory(serviceResolver, authResolverWrapper)
+	if err != nil {
+		return nil, err
+	}
+
 	crdRESTOptionsGetter, err := NewCRDRESTOptionsGetter(*o.RecommendedOptions.Etcd)
 	if err != nil {
 		return nil, err
@@ -115,8 +124,7 @@ func (o CustomResourceDefinitionsServerOptions) Config() (*apiserver.Config, err
 		GenericConfig: serverConfig,
 		ExtraConfig: apiserver.ExtraConfig{
 			CRDRESTOptionsGetter: crdRESTOptionsGetter,
-			ServiceResolver:      &serviceResolver{serverConfig.SharedInformerFactory.Core().V1().Services().Lister()},
-			AuthResolverWrapper:  webhook.NewDefaultAuthenticationInfoResolverWrapper(nil, nil, serverConfig.LoopbackClientConfig, oteltrace.NewNoopTracerProvider()),
+			ConversionFactory:    conversionFactory,
 		},
 	}
 	return config, nil