Repository: kubernetes/kubernetes (language: go)

Issue:
StatefulSet Autodelete owner references should respect controller
### What happened?

StatefulSet owner references don't set or respect the controller flag.

That should either be used, preferably through controller-runtime/controllerutil as is used in controllers like RabbitMq, crunchy postgres and Neo4jCluster.

We have observed these controllers being used on stateful sets. We haven't seen any reports of controller contention problems but the potential is there.

This should delay GA promotion of the statefulset autodelete feature (https://github.com/kubernetes/enhancements/issues/1847).

/assign @mattcary
/sig apps

### What did you expect to happen?

n/a

### How can we reproduce it (as minimally and precisely as possible)?

n/a

### Anything else we need to know?

_No response_

### Kubernetes version

n/a

### Cloud provider

<details>

</details>


### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>

StatefulSet Autodelete owner references should respect controller
### What happened?

StatefulSet owner references don't set or respect the controller flag.

That should either be used, preferably through controller-runtime/controllerutil as is used in controllers like RabbitMq, crunchy postgres and Neo4jCluster.

We have observed these controllers being used on stateful sets. We haven't seen any reports of controller contention problems but the potential is there.

This should delay GA promotion of the statefulset autodelete feature (https://github.com/kubernetes/enhancements/issues/1847).

/assign @mattcary
/sig apps

### What did you expect to happen?

n/a

### How can we reproduce it (as minimally and precisely as possible)?

n/a

### Anything else we need to know?

_No response_

### Kubernetes version

n/a

### Cloud provider

<details>

</details>


### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>



Discussion / hints:
/triage accepted
/cc @msau42 
Hmm, I don't see controller-runtime used in k8s. What's in pkg/controller/controller_utils.go doesn't seem appropriate so I guess I'll just roll it manually.
/triage accepted
/cc @msau42 
Hmm, I don't see controller-runtime used in k8s. What's in pkg/controller/controller_utils.go doesn't seem appropriate so I guess I'll just roll it manually.

Fix patch:
diff --git a/pkg/controller/statefulset/stateful_pod_control.go b/pkg/controller/statefulset/stateful_pod_control.go
index e3433619d91a4..76945d336932e 100644
--- a/pkg/controller/statefulset/stateful_pod_control.go
+++ b/pkg/controller/statefulset/stateful_pod_control.go
@@ -220,7 +220,7 @@ func (spc *StatefulPodControl) ClaimsMatchRetentionPolicy(ctx context.Context, s
 		case err != nil:
 			return false, fmt.Errorf("Could not retrieve claim %s for %s when checking PVC deletion policy", claimName, pod.Name)
 		default:
-			if !claimOwnerMatchesSetAndPod(logger, claim, set, pod) {
+			if !isClaimOwnerUpToDate(logger, claim, set, pod) {
 				return false, nil
 			}
 		}
@@ -242,14 +242,16 @@ func (spc *StatefulPodControl) UpdatePodClaimForRetentionPolicy(ctx context.Cont
 		case err != nil:
 			return fmt.Errorf("Could not retrieve claim %s not found for %s when checking PVC deletion policy: %w", claimName, pod.Name, err)
 		default:
-			if !claimOwnerMatchesSetAndPod(logger, claim, set, pod) {
+			if hasUnexpectedController(claim, set, pod) {
+				// Add an event so the user knows they're in a strange configuration. The claim will be cleaned up below.
+				msg := fmt.Sprintf("PersistentVolumeClaim %s has a conflicting OwnerReference that acts as a manging controller, the retention policy is ignored for this claim", claimName)
+				spc.recorder.Event(set, v1.EventTypeWarning, "ConflictingController", msg)
+			}
+			if !isClaimOwnerUpToDate(logger, claim, set, pod) {
 				claim = claim.DeepCopy() // Make a copy so we don't mutate the shared cache.
-				needsUpdate := updateClaimOwnerRefForSetAndPod(logger, claim, set, pod)
-				if needsUpdate {
-					err := spc.objectMgr.UpdateClaim(claim)
-					if err != nil {
-						return fmt.Errorf("Could not update claim %s for delete policy ownerRefs: %w", claimName, err)
-					}
+				updateClaimOwnerRefForSetAndPod(logger, claim, set, pod)
+				if err := spc.objectMgr.UpdateClaim(claim); err != nil {
+					return fmt.Errorf("could not update claim %s for delete policy ownerRefs: %w", claimName, err)
 				}
 			}
 		}
@@ -275,8 +277,7 @@ func (spc *StatefulPodControl) PodClaimIsStale(set *apps.StatefulSet, pod *v1.Po
 		case err != nil:
 			return false, err
 		case err == nil:
-			// A claim is stale if it doesn't match the pod's UID, including if the pod has no UID.
-			if hasStaleOwnerRef(pvc, pod) {
+			if hasStaleOwnerRef(pvc, pod, podKind) {
 				return true, nil
 			}
 		}
diff --git a/pkg/controller/statefulset/stateful_set.go b/pkg/controller/statefulset/stateful_set.go
index 3f67ef15da055..0b0875f97ae36 100644
--- a/pkg/controller/statefulset/stateful_set.go
+++ b/pkg/controller/statefulset/stateful_set.go
@@ -49,6 +49,9 @@ import (
 // controllerKind contains the schema.GroupVersionKind for this controller type.
 var controllerKind = apps.SchemeGroupVersion.WithKind("StatefulSet")
 
+// podKind contains the schema.GroupVersionKind for pods.
+var podKind = v1.SchemeGroupVersion.WithKind("Pod")
+
 // StatefulSetController controls statefulsets.
 type StatefulSetController struct {
 	// client interface
diff --git a/pkg/controller/statefulset/stateful_set_utils.go b/pkg/controller/statefulset/stateful_set_utils.go
index f222cf65ef576..c12ff28e2aa48 100644
--- a/pkg/controller/statefulset/stateful_set_utils.go
+++ b/pkg/controller/statefulset/stateful_set_utils.go
@@ -26,6 +26,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
@@ -170,9 +171,100 @@ func getPersistentVolumeClaimRetentionPolicy(set *apps.StatefulSet) apps.Statefu
 	return policy
 }
 
-// claimOwnerMatchesSetAndPod returns false if the ownerRefs of the claim are not set consistently with the
+// matchesRef returns true when the object matches the owner reference, that is the name and GVK are the same.
+func matchesRef(ref *metav1.OwnerReference, obj metav1.Object, gvk schema.GroupVersionKind) bool {
+	return gvk.GroupVersion().String() == ref.APIVersion && gvk.Kind == ref.Kind && ref.Name == obj.GetName()
+}
+
+// hasUnexpectedController returns true if the set has a retention policy and there is a controller
+// for the claim that's not the set or pod. Since the retention policy may have been changed, it is
+// always valid for the set or pod to be a controller.
+func hasUnexpectedController(claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) bool {
+	policy := getPersistentVolumeClaimRetentionPolicy(set)
+	const retain = apps.RetainPersistentVolumeClaimRetentionPolicyType
+	if policy.WhenScaled == retain && policy.WhenDeleted == retain {
+		// On a retain policy, it's not a problem for different controller to be managing the claims.
+		return false
+	}
+	for _, ownerRef := range claim.GetOwnerReferences() {
+		if matchesRef(&ownerRef, set, controllerKind) {
+			if ownerRef.UID != set.GetUID() {
+				// A UID mismatch means that pods were incorrectly orphaned. Treating this as an unexpected
+				// controller means we won't touch the PVCs (eg, leave it to the garbage collector to clean
+				// up if appropriate).
+				return true
+			}
+			continue // This is us.
+		}
+
+		if matchesRef(&ownerRef, pod, podKind) {
+			if ownerRef.UID != pod.GetUID() {
+				// This is the same situation as the set UID mismatch, above.
+				return true
+			}
+			continue // This is us.
+		}
+		if ownerRef.Controller != nil && *ownerRef.Controller {
+			return true // This is another controller.
+		}
+	}
+	return false
+}
+
+// hasNonControllerOwner returns true if the pod or set is an owner but not controller of the claim.
+func hasNonControllerOwner(claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) bool {
+	for _, ownerRef := range claim.GetOwnerReferences() {
+		if ownerRef.UID == set.GetUID() || ownerRef.UID == pod.GetUID() {
+			if ownerRef.Controller == nil || !*ownerRef.Controller {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// removeRefs removes any owner refs from the list matching predicate. Returns true if the list was changed and
+// the new (or unchanged list).
+func removeRefs(refs []metav1.OwnerReference, predicate func(ref *metav1.OwnerReference) bool) []metav1.OwnerReference {
+	newRefs := []metav1.OwnerReference{}
+	for _, ownerRef := range refs {
+		if !predicate(&ownerRef) {
+			newRefs = append(newRefs, ownerRef)
+		}
+	}
+	return newRefs
+}
+
+// isClaimOwnerUpToDate returns false if the ownerRefs of the claim are not set consistently with the
 // PVC deletion policy for the StatefulSet.
-func claimOwnerMatchesSetAndPod(logger klog.Logger, claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) bool {
+//
+// If there are stale references or unexpected controllers, this returns true in order to not touch
+// PVCs that have gotten into this unknown state. Otherwise the ownerships are checked to match the
+// PVC retention policy:
+//
+//	Retain on scaling and set deletion: no owner ref
+//	Retain on scaling and delete on set deletion: owner ref on the set only
+//	Delete on scaling and retain on set deletion: owner ref on the pod only
+//	Delete on scaling and set deletion: owner refs on both set and pod.
+func isClaimOwnerUpToDate(logger klog.Logger, claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) bool {
+	if hasStaleOwnerRef(claim, set, controllerKind) || hasStaleOwnerRef(claim, pod, podKind) {
+		// The claim is being managed by previous, presumably deleted, version of the controller. It should not be touched.
+		return true
+	}
+
+	if hasUnexpectedController(claim, set, pod) {
+		if hasOwnerRef(claim, set) || hasOwnerRef(claim, pod) {
+			return false // Need to clean up the conflicting controllers
+		}
+		// The claim refs are good, we don't want to add any controllers on top of the unexpected one.
+		return true
+	}
+
+	if hasNonControllerOwner(claim, set, pod) {
+		// Some resource has an owner ref, but there is no controller. This needs to be updated.
+		return false
+	}
+
 	policy := getPersistentVolumeClaimRetentionPolicy(set)
 	const retain = apps.RetainPersistentVolumeClaimRetentionPolicyType
 	const delete = apps.DeletePersistentVolumeClaimRetentionPolicyType
@@ -214,64 +306,53 @@ func claimOwnerMatchesSetAndPod(logger klog.Logger, claim *v1.PersistentVolumeCl
 
 // updateClaimOwnerRefForSetAndPod updates the ownerRefs for the claim according to the deletion policy of
 // the StatefulSet. Returns true if the claim was changed and should be updated and false otherwise.
-func updateClaimOwnerRefForSetAndPod(logger klog.Logger, claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) bool {
-	needsUpdate := false
-	// Sometimes the version and kind are not set {pod,set}.TypeMeta. These are necessary for the ownerRef.
-	// This is the case both in real clusters and the unittests.
-	// TODO: there must be a better way to do this other than hardcoding the pod version?
-	updateMeta := func(tm *metav1.TypeMeta, kind string) {
-		if tm.APIVersion == "" {
-			if kind == "StatefulSet" {
-				tm.APIVersion = "apps/v1"
-			} else {
-				tm.APIVersion = "v1"
-			}
-		}
-		if tm.Kind == "" {
-			tm.Kind = kind
-		}
+// isClaimOwnerUpToDate should be called before this to avoid an expensive update operation.
+func updateClaimOwnerRefForSetAndPod(logger klog.Logger, claim *v1.PersistentVolumeClaim, set *apps.StatefulSet, pod *v1.Pod) {
+	refs := claim.GetOwnerReferences()
+
+	unexpectedController := hasUnexpectedController(claim, set, pod)
+
+	// Scrub any ownerRefs to our set & pod.
+	refs = removeRefs(refs, func(ref *metav1.OwnerReference) bool {
+		return matchesRef(ref, set, controllerKind) || matchesRef(ref, pod, podKind)
+	})
+
+	if unexpectedController {
+		// Leave ownerRefs to our set & pod scrubed and return without creating new ones.
+		claim.SetOwnerReferences(refs)
+		return
 	}
-	podMeta := pod.TypeMeta
-	updateMeta(&podMeta, "Pod")
-	setMeta := set.TypeMeta
-	updateMeta(&setMeta, "StatefulSet")
+
 	policy := getPersistentVolumeClaimRetentionPolicy(set)
 	const retain = apps.RetainPersistentVolumeClaimRetentionPolicyType
 	const delete = apps.DeletePersistentVolumeClaimRetentionPolicyType
 	switch {
 	default:
 		logger.Error(nil, "Unknown policy, treating as Retain", "policy", set.Spec.PersistentVolumeClaimRetentionPolicy)
-		fallthrough
+		// Nothing to do
 	case policy.WhenScaled == retain && policy.WhenDeleted == retain:
-		needsUpdate = removeOwnerRef(claim, set) || needsUpdate
-		needsUpdate = removeOwnerRef(claim, pod) || needsUpdate
+		// Nothing to do
 	case policy.WhenScaled == retain && policy.WhenDeleted == delete:
-		needsUpdate = setOwnerRef(claim, set, &setMeta) || needsUpdate
-		needsUpdate = removeOwnerRef(claim, pod) || needsUpdate
+		refs = addControllerRef(refs, set, controllerKind)
 	case policy.WhenScaled == delete && policy.WhenDeleted == retain:
-		needsUpdate = removeOwnerRef(claim, set) || needsUpdate
 		podScaledDown := !podInOrdinalRange(pod, set)
 		if podScaledDown {
-			needsUpdate = setOwnerRef(claim, pod, &podMeta) || needsUpdate
-		}
-		if !podScaledDown {
-			needsUpdate = removeOwnerRef(claim, pod) || needsUpdate
+			refs = addControllerRef(refs, pod, podKind)
 		}
 	case policy.WhenScaled == delete && policy.WhenDeleted == delete:
 		podScaledDown := !podInOrdinalRange(pod, set)
 		if podScaledDown {
-			needsUpdate = removeOwnerRef(claim, set) || needsUpdate
-			needsUpdate = setOwnerRef(claim, pod, &podMeta) || needsUpdate
+			refs = addControllerRef(refs, pod, podKind)
 		}
 		if !podScaledDown {
-			needsUpdate = setOwnerRef(claim, set, &setMeta) || needsUpdate
-			needsUpdate = removeOwnerRef(claim, pod) || needsUpdate
+			refs = addControllerRef(refs, set, controllerKind)
 		}
 	}
-	return needsUpdate
+	claim.SetOwnerReferences(refs)
 }
 
-// hasOwnerRef returns true if target has an ownerRef to owner.
+// hasOwnerRef returns true if target has an ownerRef to owner (as its UID).
+// This does not check if the owner is a controller.
 func hasOwnerRef(target, owner metav1.Object) bool {
 	ownerUID := owner.GetUID()
 	for _, ownerRef := range target.GetOwnerReferences() {
@@ -282,53 +363,28 @@ func hasOwnerRef(target, owner metav1.Object) bool {
 	return false
 }
 
-// hasStaleOwnerRef returns true if target has a ref to owner that appears to be stale.
-func hasStaleOwnerRef(target, owner metav1.Object) bool {
+// hasStaleOwnerRef returns true if target has a ref to owner that appears to be stale, that is,
+// the ref matches the object but not the UID.
+func hasStaleOwnerRef(target *v1.PersistentVolumeClaim, obj metav1.Object, gvk schema.GroupVersionKind) bool {
 	for _, ownerRef := range target.GetOwnerReferences() {
-		if ownerRef.Name == owner.GetName() && ownerRef.UID != owner.GetUID() {
-			return true
+		if matchesRef(&ownerRef, obj, gvk) {
+			return ownerRef.UID != obj.GetUID()
 		}
 	}
 	return false
 }
 
-// setOwnerRef adds owner to the ownerRefs of target, if necessary. Returns true if target needs to be
-// updated and false otherwise.
-func setOwnerRef(target, owner metav1.Object, ownerType *metav1.TypeMeta) bool {
-	if hasOwnerRef(target, owner) {
-		return false
-	}
-	ownerRefs := append(
-		target.GetOwnerReferences(),
-		metav1.OwnerReference{
-			APIVersion: ownerType.APIVersion,
-			Kind:       ownerType.Kind,
-			Name:       owner.GetName(),
-			UID:        owner.GetUID(),
-		})
-	target.SetOwnerReferences(ownerRefs)
-	return true
-}
-
-// removeOwnerRef removes owner from the ownerRefs of target, if necessary. Returns true if target needs
-// to be updated and false otherwise.
-func removeOwnerRef(target, owner metav1.Object) bool {
-	if !hasOwnerRef(target, owner) {
-		return false
-	}
-	ownerUID := owner.GetUID()
-	oldRefs := target.GetOwnerReferences()
-	newRefs := make([]metav1.OwnerReference, len(oldRefs)-1)
-	skip := 0
-	for i := range oldRefs {
-		if oldRefs[i].UID == ownerUID {
-			skip = -1
-		} else {
-			newRefs[i+skip] = oldRefs[i]
+// addControllerRef returns refs with owner added as a controller, if necessary.
+func addControllerRef(refs []metav1.OwnerReference, owner metav1.Object, gvk schema.GroupVersionKind) []metav1.OwnerReference {
+	for _, ref := range refs {
+		if ref.UID == owner.GetUID() {
+			// Already added. Since we scrub our refs before making any changes, we know it's already
+			// a controller if appropriate.
+			return refs
 		}
 	}
-	target.SetOwnerReferences(newRefs)
-	return true
+
+	return append(refs, *metav1.NewControllerRef(owner, gvk))
 }
 
 // getPersistentVolumeClaims gets a map of PersistentVolumeClaims to their template names, as defined in set. The


Test patch:
diff --git a/pkg/controller/statefulset/stateful_pod_control_test.go b/pkg/controller/statefulset/stateful_pod_control_test.go
index 1cc9c2782c4b1..12eb079111035 100644
--- a/pkg/controller/statefulset/stateful_pod_control_test.go
+++ b/pkg/controller/statefulset/stateful_pod_control_test.go
@@ -41,6 +41,7 @@ import (
 	_ "k8s.io/kubernetes/pkg/apis/apps/install"
 	_ "k8s.io/kubernetes/pkg/apis/core/install"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/utils/ptr"
 )
 
 func TestStatefulPodControlCreatesPods(t *testing.T) {
@@ -502,7 +503,7 @@ func TestStatefulPodControlDeleteFailure(t *testing.T) {
 }
 
 func TestStatefulPodControlClaimsMatchDeletionPolcy(t *testing.T) {
-	// The claimOwnerMatchesSetAndPod is tested exhaustively in stateful_set_utils_test; this
+	// The isClaimOwnerUpToDate is tested exhaustively in stateful_set_utils_test; this
 	// test is for the wiring to the method tested there.
 	_, ctx := ktesting.NewTestContext(t)
 	fakeClient := &fake.Clientset{}
@@ -539,52 +540,67 @@ func TestStatefulPodControlClaimsMatchDeletionPolcy(t *testing.T) {
 func TestStatefulPodControlUpdatePodClaimForRetentionPolicy(t *testing.T) {
 	// All the update conditions are tested exhaustively in stateful_set_utils_test. This
 	// tests the wiring from the pod control to that method.
-	testFn := func(t *testing.T) {
-		_, ctx := ktesting.NewTestContext(t)
-		defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetAutoDeletePVC, true)()
+	_, ctx := ktesting.NewTestContext(t)
+	testCases := []struct {
+		name      string
+		ownerRef  []metav1.OwnerReference
+		expectRef bool
+	}{
+		{
+			name:      "bare PVC",
+			expectRef: true,
+		},
+		{
+			name:      "PVC already controller",
+			ownerRef:  []metav1.OwnerReference{{Controller: ptr.To(true), Name: "foobar"}},
+			expectRef: false,
+		},
+	}
+
+	for _, tc := range testCases {
 		fakeClient := &fake.Clientset{}
 		indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
 		claimLister := corelisters.NewPersistentVolumeClaimLister(indexer)
 		fakeClient.AddReactor("update", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) {
 			update := action.(core.UpdateAction)
-			indexer.Update(update.GetObject())
+			if err := indexer.Update(update.GetObject()); err != nil {
+				t.Fatalf("could not update index: %v", err)
+			}
 			return true, update.GetObject(), nil
 		})
 		set := newStatefulSet(3)
 		set.GetObjectMeta().SetUID("set-123")
-		pod := newStatefulSetPod(set, 0)
-		claims := getPersistentVolumeClaims(set, pod)
-		for k := range claims {
-			claim := claims[k]
-			indexer.Add(&claim)
+		pod0 := newStatefulSetPod(set, 0)
+		claims0 := getPersistentVolumeClaims(set, pod0)
+		for k := range claims0 {
+			claim := claims0[k]
+			if tc.ownerRef != nil {
+				claim.SetOwnerReferences(tc.ownerRef)
+			}
+			if err := indexer.Add(&claim); err != nil {
+				t.Errorf("Could not add claim %s: %v", k, err)
+			}
 		}
 		control := NewStatefulPodControl(fakeClient, nil, claimLister, &noopRecorder{})
 		set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
 			WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
 			WhenScaled:  apps.RetainPersistentVolumeClaimRetentionPolicyType,
 		}
-		if err := control.UpdatePodClaimForRetentionPolicy(ctx, set, pod); err != nil {
-			t.Errorf("Unexpected error for UpdatePodClaimForRetentionPolicy (retain): %v", err)
+		if err := control.UpdatePodClaimForRetentionPolicy(ctx, set, pod0); err != nil {
+			t.Errorf("Unexpected error for UpdatePodClaimForRetentionPolicy (retain), pod0: %v", err)
 		}
-		expectRef := utilfeature.DefaultFeatureGate.Enabled(features.StatefulSetAutoDeletePVC)
-		for k := range claims {
-			claim, err := claimLister.PersistentVolumeClaims(claims[k].Namespace).Get(claims[k].Name)
+		expectRef := tc.expectRef
+		for k := range claims0 {
+			claim, err := claimLister.PersistentVolumeClaims(claims0[k].Namespace).Get(claims0[k].Name)
 			if err != nil {
 				t.Errorf("Unexpected error getting Claim %s/%s: %v", claim.Namespace, claim.Name, err)
 			}
 			if hasOwnerRef(claim, set) != expectRef {
-				t.Errorf("Claim %s/%s bad set owner ref", claim.Namespace, claim.Name)
+				t.Logf("Claim ref %v", claim.GetOwnerReferences())
+				t.Errorf("%s: Claim %s/%s bad set owner ref", tc.name, claim.Namespace, claim.Name)
 			}
 		}
 	}
-	t.Run("StatefulSetAutoDeletePVCEnabled", func(t *testing.T) {
-		defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetAutoDeletePVC, true)()
-		testFn(t)
-	})
-	t.Run("StatefulSetAutoDeletePVCDisabled", func(t *testing.T) {
-		defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.StatefulSetAutoDeletePVC, false)()
-		testFn(t)
-	})
 }
 
 func TestPodClaimIsStale(t *testing.T) {
@@ -663,12 +679,22 @@ func TestPodClaimIsStale(t *testing.T) {
 				claimIndexer.Add(&claim)
 			case stale:
 				claim.SetOwnerReferences([]metav1.OwnerReference{
-					{Name: "set-3", UID: types.UID("stale")},
+					{
+						Name:       "set-3",
+						UID:        types.UID("stale"),
+						APIVersion: "v1",
+						Kind:       "Pod",
+					},
 				})
 				claimIndexer.Add(&claim)
 			case withRef:
 				claim.SetOwnerReferences([]metav1.OwnerReference{
-					{Name: "set-3", UID: types.UID("123")},
+					{
+						Name:       "set-3",
+						UID:        types.UID("123"),
+						APIVersion: "v1",
+						Kind:       "Pod",
+					},
 				})
 				claimIndexer.Add(&claim)
 			}
@@ -710,7 +736,8 @@ func TestStatefulPodControlRetainDeletionPolicyUpdate(t *testing.T) {
 		}
 		for k := range claims {
 			claim := claims[k]
-			setOwnerRef(&claim, set, &set.TypeMeta) // This ownerRef should be removed in the update.
+			// This ownerRef should be removed in the update.
+			claim.SetOwnerReferences(addControllerRef(claim.GetOwnerReferences(), set, controllerKind))
 			claimIndexer.Add(&claim)
 		}
 		control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder)
diff --git a/pkg/controller/statefulset/stateful_set_utils_test.go b/pkg/controller/statefulset/stateful_set_utils_test.go
index d5d87724dad43..e0fc0f7cfdffe 100644
--- a/pkg/controller/statefulset/stateful_set_utils_test.go
+++ b/pkg/controller/statefulset/stateful_set_utils_test.go
@@ -23,19 +23,20 @@ import (
 	"regexp"
 	"sort"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
+	apps "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/klog/v2"
 	"k8s.io/klog/v2/ktesting"
-
-	apps "k8s.io/api/apps/v1"
-	v1 "k8s.io/api/core/v1"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	"k8s.io/kubernetes/pkg/controller/history"
 	"k8s.io/utils/ptr"
@@ -63,6 +64,88 @@ func getClaimPodName(set *apps.StatefulSet, claim *v1.PersistentVolumeClaim) str
 	return matches[1]
 }
 
+// ownerRefsChanged returns true if newRefs does not match originalRefs.
+func ownerRefsChanged(originalRefs, newRefs []metav1.OwnerReference) bool {
+	if len(originalRefs) != len(newRefs) {
+		return true
+	}
+	key := func(ref *metav1.OwnerReference) string {
+		return fmt.Sprintf("%s-%s-%s", ref.APIVersion, ref.Kind, ref.Name)
+	}
+	refs := map[string]bool{}
+	for i := range originalRefs {
+		refs[key(&originalRefs[i])] = true
+	}
+	for i := range newRefs {
+		k := key(&newRefs[i])
+		if val, found := refs[k]; !found || !val {
+			return true
+		}
+		refs[k] = false
+	}
+	return false
+}
+
+func TestOwnerRefsChanged(t *testing.T) {
+	toRefs := func(strs []string) []metav1.OwnerReference {
+		refs := []metav1.OwnerReference{}
+		for _, s := range strs {
+			pieces := strings.Split(s, "/")
+			refs = append(refs, metav1.OwnerReference{
+				APIVersion: pieces[0],
+				Kind:       pieces[1],
+				Name:       pieces[2],
+			})
+		}
+		return refs
+	}
+	testCases := []struct {
+		orig, new []string
+		changed   bool
+	}{
+		{
+			orig:    []string{"v1/pod/foo"},
+			new:     []string{},
+			changed: true,
+		},
+		{
+			orig:    []string{"v1/pod/foo"},
+			new:     []string{"v1/pod/foo"},
+			changed: false,
+		},
+		{
+			orig:    []string{"v1/pod/foo"},
+			new:     []string{"v1/pod/bar"},
+			changed: true,
+		},
+		{
+			orig:    []string{"v1/pod/foo", "v1/set/bob"},
+			new:     []string{"v1/pod/foo", "v1/set/alice"},
+			changed: true,
+		},
+		{
+			orig:    []string{"v1/pod/foo", "v1/set/bob"},
+			new:     []string{"v1/pod/foo", "v1/set/bob"},
+			changed: false,
+		},
+		{
+			orig:    []string{"v1/pod/foo", "v1/set/bob"},
+			new:     []string{"v1/pod/foo", "v1/set/bob", "v1/set/bob"},
+			changed: true,
+		},
+		{
+			orig:    []string{"v1/pod/foo", "v1/set/bob"},
+			new:     []string{"v1/set/bob", "v1/pod/foo"},
+			changed: false,
+		},
+	}
+	for _, tc := range testCases {
+		if ownerRefsChanged(toRefs(tc.orig), toRefs(tc.new)) != tc.changed {
+			t.Errorf("Expected change=%t but got %t for %v vs %v", tc.changed, !tc.changed, tc.orig, tc.new)
+		}
+	}
+}
+
 func TestGetParentNameAndOrdinal(t *testing.T) {
 	set := newStatefulSet(3)
 	pod := newStatefulSetPod(set, 1)
@@ -253,7 +336,84 @@ func TestGetPersistentVolumeClaimRetentionPolicy(t *testing.T) {
 	}
 }
 
-func TestClaimOwnerMatchesSetAndPod(t *testing.T) {
+func TestMatchesRef(t *testing.T) {
+	testCases := []struct {
+		name        string
+		ref         metav1.OwnerReference
+		obj         metav1.ObjectMeta
+		schema      schema.GroupVersionKind
+		shouldMatch bool
+	}{
+		{
+			name: "full match",
+			ref: metav1.OwnerReference{
+				APIVersion: "v1",
+				Kind:       "Pod",
+				Name:       "fred",
+				UID:        "abc",
+			},
+			obj: metav1.ObjectMeta{
+				Name: "fred",
+				UID:  "abc",
+			},
+			schema:      podKind,
+			shouldMatch: true,
+		},
+		{
+			name: "match without UID",
+			ref: metav1.OwnerReference{
+				APIVersion: "v1",
+				Kind:       "Pod",
+				Name:       "fred",
+				UID:        "abc",
+			},
+			obj: metav1.ObjectMeta{
+				Name: "fred",
+				UID:  "not-matching",
+			},
+			schema:      podKind,
+			shouldMatch: true,
+		},
+		{
+			name: "mismatch name",
+			ref: metav1.OwnerReference{
+				APIVersion: "v1",
+				Kind:       "Pod",
+				Name:       "fred",
+				UID:        "abc",
+			},
+			obj: metav1.ObjectMeta{
+				Name: "joan",
+				UID:  "abc",
+			},
+			schema:      podKind,
+			shouldMatch: false,
+		},
+		{
+			name: "wrong schema",
+			ref: metav1.OwnerReference{
+				APIVersion: "beta2",
+				Kind:       "Pod",
+				Name:       "fred",
+				UID:        "abc",
+			},
+			obj: metav1.ObjectMeta{
+				Name: "fred",
+				UID:  "abc",
+			},
+			schema:      podKind,
+			shouldMatch: false,
+		},
+	}
+	for _, tc := range testCases {
+		got := matchesRef(&tc.ref, &tc.obj, tc.schema)
+		if got != tc.shouldMatch {
+			t.Errorf("Failed %s: got %t, expected %t", tc.name, got, tc.shouldMatch)
+		}
+	}
+}
+
+func TestIsClaimOwnerUpToDate(t *testing.T) {
 	testCases := []struct {
 		name            string
 		scaleDownPolicy apps.PersistentVolumeClaimRetentionPolicyType
@@ -334,24 +494,32 @@ func TestClaimOwnerMatchesSetAndPod(t *testing.T) {
 						WhenDeleted: tc.setDeletePolicy,
 					}
 					set.Spec.Replicas = &tc.replicas
+					claimRefs := claim.GetOwnerReferences()
 					if setPodRef {
-						setOwnerRef(&claim, &pod, &pod.TypeMeta)
+						claimRefs = addControllerRef(claimRefs, &pod, podKind)
 					}
 					if setSetRef {
-						setOwnerRef(&claim, &set, &set.TypeMeta)
+						claimRefs = addControllerRef(claimRefs, &set, controllerKind)
 					}
 					if useOtherRefs {
-						randomObject1 := v1.Pod{}
-						randomObject1.Name = "rand1"
-						randomObject1.GetObjectMeta().SetUID("rand1-abc")
-						randomObject2 := v1.Pod{}
-						randomObject2.Name = "rand2"
-						randomObject2.GetObjectMeta().SetUID("rand2-def")
-						setOwnerRef(&claim, &randomObject1, &randomObject1.TypeMeta)
-						setOwnerRef(&claim, &randomObject2, &randomObject2.TypeMeta)
+						claimRefs = append(
+							claimRefs,
+							metav1.OwnerReference{
+								Name:       "rand1",
+								APIVersion: "v1",
+								Kind:       "Pod",
+								UID:        "rand1-uid",
+							},
+							metav1.OwnerReference{
+								Name:       "rand2",
+								APIVersion: "v1",
+								Kind:       "Pod",
+								UID:        "rand2-uid",
+							})
 					}
+					claim.SetOwnerReferences(claimRefs)
 					shouldMatch := setPodRef == tc.needsPodRef && setSetRef == tc.needsSetRef
-					if claimOwnerMatchesSetAndPod(logger, &claim, &set, &pod) != shouldMatch {
+					if isClaimOwnerUpToDate(logger, &claim, &set, &pod) != shouldMatch {
 						t.Errorf("Bad match for %s with pod=%v,set=%v,others=%v", tc.name, setPodRef, setSetRef, useOtherRefs)
 					}
 				}
@@ -360,14 +528,194 @@ func TestClaimOwnerMatchesSetAndPod(t *testing.T) {
 	}
 }
 
+func TestClaimOwnerUpToDateEdgeCases(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
+	logger := klog.FromContext(ctx)
+
+	testCases := []struct {
+		name        string
+		ownerRefs   []metav1.OwnerReference
+		policy      apps.StatefulSetPersistentVolumeClaimRetentionPolicy
+		shouldMatch bool
+	}{
+		{
+			name: "normal controller, pod",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "pod-1",
+					APIVersion: "v1",
+					Kind:       "Pod",
+					UID:        "pod-123",
+					Controller: ptr.To(true),
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.DeletePersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: true,
+		},
+		{
+			name: "non-controller causes policy mismatch, pod",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "pod-1",
+					APIVersion: "v1",
+					Kind:       "Pod",
+					UID:        "pod-123",
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.DeletePersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: false,
+		},
+		{
+			name: "stale controller does not affect policy, pod",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "pod-1",
+					APIVersion: "v1",
+					Kind:       "Pod",
+					UID:        "pod-stale",
+					Controller: ptr.To(true),
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.DeletePersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: true,
+		},
+		{
+			name: "unexpected controller causes policy mismatch, pod",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "pod-1",
+					APIVersion: "v1",
+					Kind:       "Pod",
+					UID:        "pod-123",
+					Controller: ptr.To(true),
+				},
+				{
+					Name:       "Random",
+					APIVersion: "v1",
+					Kind:       "Pod",
+					UID:        "random",
+					Controller: ptr.To(true),
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.DeletePersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: false,
+		},
+		{
+			name: "normal controller, set",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "stateful-set",
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					UID:        "ss-456",
+					Controller: ptr.To(true),
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.RetainPersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: true,
+		},
+		{
+			name: "non-controller causes policy mismatch, set",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "stateful-set",
+					APIVersion: "appsv1",
+					Kind:       "StatefulSet",
+					UID:        "ss-456",
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.RetainPersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: false,
+		},
+		{
+			name: "stale controller ignored, set",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "stateful-set",
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					UID:        "set-stale",
+					Controller: ptr.To(true),
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.RetainPersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: true,
+		},
+		{
+			name: "unexpected controller causes policy mismatch, set",
+			ownerRefs: []metav1.OwnerReference{
+				{
+					Name:       "stateful-set",
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					UID:        "ss-456",
+					Controller: ptr.To(true),
+				},
+				{
+					Name:       "Random",
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					UID:        "random",
+					Controller: ptr.To(true),
+				},
+			},
+			policy: apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  apps.RetainPersistentVolumeClaimRetentionPolicyType,
+				WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			},
+			shouldMatch: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		claim := v1.PersistentVolumeClaim{}
+		claim.Name = "target-claim"
+		pod := v1.Pod{}
+		pod.Name = "pod-1"
+		pod.GetObjectMeta().SetUID("pod-123")
+		set := apps.StatefulSet{}
+		set.Name = "stateful-set"
+		set.GetObjectMeta().SetUID("ss-456")
+		set.Spec.PersistentVolumeClaimRetentionPolicy = &tc.policy
+		set.Spec.Replicas = ptr.To(int32(1))
+		claim.SetOwnerReferences(tc.ownerRefs)
+		got := isClaimOwnerUpToDate(logger, &claim, &set, &pod)
+		if got != tc.shouldMatch {
+			t.Errorf("Unexpected match for %s, got %t expected %t", tc.name, got, tc.shouldMatch)
+		}
+	}
+}
+
 func TestUpdateClaimOwnerRefForSetAndPod(t *testing.T) {
 	testCases := []struct {
-		name            string
-		scaleDownPolicy apps.PersistentVolumeClaimRetentionPolicyType
-		setDeletePolicy apps.PersistentVolumeClaimRetentionPolicyType
-		condemned       bool
-		needsPodRef     bool
-		needsSetRef     bool
+		name                 string
+		scaleDownPolicy      apps.PersistentVolumeClaimRetentionPolicyType
+		setDeletePolicy      apps.PersistentVolumeClaimRetentionPolicyType
+		condemned            bool
+		needsPodRef          bool
+		needsSetRef          bool
+		unexpectedController bool
 	}{
 		{
 			name:            "retain",
@@ -417,47 +765,228 @@ func TestUpdateClaimOwnerRefForSetAndPod(t *testing.T) {
 			needsPodRef:     true,
 			needsSetRef:     false,
 		},
+		{
+			name:                 "unexpected controller",
+			scaleDownPolicy:      apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			setDeletePolicy:      apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			condemned:            true,
+			needsPodRef:          false,
+			needsSetRef:          false,
+			unexpectedController: true,
+		},
 	}
 	for _, tc := range testCases {
-		for _, hasPodRef := range []bool{true, false} {
-			for _, hasSetRef := range []bool{true, false} {
-				_, ctx := ktesting.NewTestContext(t)
-				logger := klog.FromContext(ctx)
-				set := apps.StatefulSet{}
-				set.Name = "ss"
-				numReplicas := int32(5)
-				set.Spec.Replicas = &numReplicas
-				set.SetUID("ss-123")
-				set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
-					WhenScaled:  tc.scaleDownPolicy,
-					WhenDeleted: tc.setDeletePolicy,
-				}
-				pod := v1.Pod{}
-				if tc.condemned {
-					pod.Name = "pod-8"
-				} else {
-					pod.Name = "pod-1"
-				}
-				pod.SetUID("pod-456")
-				claim := v1.PersistentVolumeClaim{}
-				if hasPodRef {
-					setOwnerRef(&claim, &pod, &pod.TypeMeta)
-				}
-				if hasSetRef {
-					setOwnerRef(&claim, &set, &set.TypeMeta)
-				}
-				needsUpdate := hasPodRef != tc.needsPodRef || hasSetRef != tc.needsSetRef
-				shouldUpdate := updateClaimOwnerRefForSetAndPod(logger, &claim, &set, &pod)
-				if shouldUpdate != needsUpdate {
-					t.Errorf("Bad update for %s hasPodRef=%v hasSetRef=%v", tc.name, hasPodRef, hasSetRef)
-				}
-				if hasOwnerRef(&claim, &pod) != tc.needsPodRef {
-					t.Errorf("Bad pod ref for %s hasPodRef=%v hasSetRef=%v", tc.name, hasPodRef, hasSetRef)
-				}
-				if hasOwnerRef(&claim, &set) != tc.needsSetRef {
-					t.Errorf("Bad set ref for %s hasPodRef=%v hasSetRef=%v", tc.name, hasPodRef, hasSetRef)
+		for variations := 0; variations < 8; variations++ {
+			hasPodRef := (variations & 1) != 0
+			hasSetRef := (variations & 2) != 0
+			extraOwner := (variations & 3) != 0
+			_, ctx := ktesting.NewTestContext(t)
+			logger := klog.FromContext(ctx)
+			set := apps.StatefulSet{}
+			set.Name = "ss"
+			numReplicas := int32(5)
+			set.Spec.Replicas = &numReplicas
+			set.SetUID("ss-123")
+			set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+				WhenScaled:  tc.scaleDownPolicy,
+				WhenDeleted: tc.setDeletePolicy,
+			}
+			pod := v1.Pod{}
+			if tc.condemned {
+				pod.Name = "pod-8"
+			} else {
+				pod.Name = "pod-1"
+			}
+			pod.SetUID("pod-456")
+			claim := v1.PersistentVolumeClaim{}
+			claimRefs := claim.GetOwnerReferences()
+			if hasPodRef {
+				claimRefs = addControllerRef(claimRefs, &pod, podKind)
+			}
+			if hasSetRef {
+				claimRefs = addControllerRef(claimRefs, &set, controllerKind)
+			}
+			if extraOwner {
+				// Note the extra owner should not affect our owner references.
+				claimRefs = append(claimRefs, metav1.OwnerReference{
+					APIVersion: "custom/v1",
+					Kind:       "random",
+					Name:       "random",
+					UID:        "abc",
+				})
+			}
+			if tc.unexpectedController {
+				claimRefs = append(claimRefs, metav1.OwnerReference{
+					APIVersion: "custom/v1",
+					Kind:       "Unknown",
+					Name:       "unknown",
+					UID:        "xyz",
+					Controller: ptr.To(true),
+				})
+			}
+			claim.SetOwnerReferences(claimRefs)
+			updateClaimOwnerRefForSetAndPod(logger, &claim, &set, &pod)
+			// Confirm that after the update, the specified owner is set as the only controller.
+			// Any other controllers will be cleaned update by the update.
+			check := func(target, owner metav1.Object) bool {
+				for _, ref := range target.GetOwnerReferences() {
+					if ref.UID == owner.GetUID() {
+						return ref.Controller != nil && *ref.Controller
+					}
 				}
+				return false
 			}
+			if check(&claim, &pod) != tc.needsPodRef {
+				t.Errorf("Bad pod ref for %s hasPodRef=%v hasSetRef=%v", tc.name, hasPodRef, hasSetRef)
+			}
+			if check(&claim, &set) != tc.needsSetRef {
+				t.Errorf("Bad set ref for %s hasPodRef=%v hasSetRef=%v", tc.name, hasPodRef, hasSetRef)
+			}
+		}
+	}
+}
+
+func TestUpdateClaimControllerRef(t *testing.T) {
+	testCases := []struct {
+		name         string
+		originalRefs []metav1.OwnerReference
+		expectedRefs []metav1.OwnerReference
+	}{
+		{
+			name: "set correctly",
+			originalRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					UID:        "123",
+					Controller: ptr.To(true),
+				},
+				{
+					APIVersion: "someone",
+					Kind:       "Else",
+					Name:       "foo",
+				},
+			},
+			expectedRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					UID:        "123",
+					Controller: ptr.To(true),
+				},
+				{
+					APIVersion: "someone",
+					Kind:       "Else",
+					Name:       "foo",
+				},
+			},
+		},
+		{
+			name: "missing controller",
+			originalRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					UID:        "123",
+				},
+			},
+			expectedRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					UID:        "123",
+					Controller: ptr.To(true),
+				},
+			},
+		},
+		{
+			name: "matching name but missing",
+			originalRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "someone",
+					Kind:       "else",
+					Name:       "sts",
+					UID:        "456",
+				},
+			},
+			expectedRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "someone",
+					Kind:       "else",
+					Name:       "sts",
+					UID:        "456",
+				},
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					UID:        "123",
+					Controller: ptr.To(true),
+				},
+			},
+		},
+		{
+			name:         "not present",
+			originalRefs: []metav1.OwnerReference{},
+			expectedRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					UID:        "123",
+					Controller: ptr.To(true),
+				},
+			},
+		},
+		{
+			name: "controller, but no UID",
+			originalRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+					Controller: ptr.To(true),
+				},
+			},
+			// The missing UID is interpreted as an unexpected stale reference.
+			expectedRefs: []metav1.OwnerReference{},
+		},
+		{
+			name: "neither controller nor UID",
+			originalRefs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "sts",
+				},
+			},
+			// The missing UID is interpreted as an unexpected stale reference.
+			expectedRefs: []metav1.OwnerReference{},
+		},
+	}
+	for _, tc := range testCases {
+		_, ctx := ktesting.NewTestContext(t)
+		logger := klog.FromContext(ctx)
+		set := apps.StatefulSet{}
+		set.Name = "sts"
+		set.Spec.Replicas = ptr.To(int32(1))
+		set.SetUID("123")
+		set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+			WhenScaled:  apps.DeletePersistentVolumeClaimRetentionPolicyType,
+			WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType,
+		}
+		pod := v1.Pod{}
+		pod.Name = "pod-0"
+		pod.SetUID("456")
+		claim := v1.PersistentVolumeClaim{}
+		claim.SetOwnerReferences(tc.originalRefs)
+		updateClaimOwnerRefForSetAndPod(logger, &claim, &set, &pod)
+		if ownerRefsChanged(tc.expectedRefs, claim.GetOwnerReferences()) {
+			t.Errorf("%s: expected %v, got %v", tc.name, tc.expectedRefs, claim.GetOwnerReferences())
 		}
 	}
 }
@@ -465,23 +994,427 @@ func TestUpdateClaimOwnerRefForSetAndPod(t *testing.T) {
 func TestHasOwnerRef(t *testing.T) {
 	target := v1.Pod{}
 	target.SetOwnerReferences([]metav1.OwnerReference{
-		{UID: "123"}, {UID: "456"}})
-	ownerA := v1.Pod{}
-	ownerA.GetObjectMeta().SetUID("123")
-	ownerB := v1.Pod{}
-	ownerB.GetObjectMeta().SetUID("789")
-	if !hasOwnerRef(&target, &ownerA) {
-		t.Error("Missing owner")
+		{UID: "123", Controller: ptr.To(true)},
+		{UID: "456", Controller: ptr.To(false)},
+		{UID: "789"},
+	})
+	testCases := []struct {
+		uid    types.UID
+		hasRef bool
+	}{
+		{
+			uid:    "123",
+			hasRef: true,
+		},
+		{
+			uid:    "456",
+			hasRef: true,
+		},
+		{
+			uid:    "789",
+			hasRef: true,
+		},
+		{
+			uid:    "012",
+			hasRef: false,
+		},
 	}
-	if hasOwnerRef(&target, &ownerB) {
-		t.Error("Unexpected owner")
+	for _, tc := range testCases {
+		owner := v1.Pod{}
+		owner.GetObjectMeta().SetUID(tc.uid)
+		got := hasOwnerRef(&target, &owner)
+		if got != tc.hasRef {
+			t.Errorf("Expected %t for %s, got %t", tc.hasRef, tc.uid, got)
+		}
+	}
+}
+
+func TestHasUnexpectedController(t *testing.T) {
+	// Each test case will be tested against a StatefulSet named "set" and a Pod named "pod" with UIDs "123".
+	testCases := []struct {
+		name                             string
+		refs                             []metav1.OwnerReference
+		shouldReportUnexpectedController bool
+	}{
+		{
+			name: "custom controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "chipmunks/v1",
+					Kind:       "CustomController",
+					Name:       "simon",
+					UID:        "other-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: true,
+		},
+		{
+			name: "custom non-controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "chipmunks/v1",
+					Kind:       "CustomController",
+					Name:       "simon",
+					UID:        "other-uid",
+					Controller: ptr.To(false),
+				},
+			},
+			shouldReportUnexpectedController: false,
+		},
+		{
+			name: "custom unspecified controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "chipmunks/v1",
+					Kind:       "CustomController",
+					Name:       "simon",
+					UID:        "other-uid",
+				},
+			},
+			shouldReportUnexpectedController: false,
+		},
+		{
+			name: "other pod controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "simon",
+					UID:        "other-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: true,
+		},
+		{
+			name: "other set controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "simon",
+					UID:        "other-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: true,
+		},
+		{
+			name: "own set controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "set",
+					UID:        "set-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: false,
+		},
+		{
+			name: "own set controller, stale uid",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "StatefulSet",
+					Name:       "set",
+					UID:        "stale-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: true,
+		},
+		{
+			name: "own pod controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: false,
+		},
+		{
+			name: "own pod controller, stale uid",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "stale-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: true,
+		},
+		{
+			// API validation should prevent two controllers from being set,
+			// but for completeness it is still tested.
+			name: "own controller and another",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod-uid",
+					Controller: ptr.To(true),
+				},
+				{
+					APIVersion: "chipmunks/v1",
+					Kind:       "CustomController",
+					Name:       "simon",
+					UID:        "other-uid",
+					Controller: ptr.To(true),
+				},
+			},
+			shouldReportUnexpectedController: true,
+		},
+		{
+			name: "own controller and a non-controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod-uid",
+					Controller: ptr.To(true),
+				},
+				{
+					APIVersion: "chipmunks/v1",
+					Kind:       "CustomController",
+					Name:       "simon",
+					UID:        "other-uid",
+					Controller: ptr.To(false),
+				},
+			},
+			shouldReportUnexpectedController: false,
+		},
+	}
+	for _, tc := range testCases {
+		target := &v1.PersistentVolumeClaim{}
+		target.SetOwnerReferences(tc.refs)
+		set := &apps.StatefulSet{}
+		set.SetName("set")
+		set.SetUID("set-uid")
+		pod := &v1.Pod{}
+		pod.SetName("pod")
+		pod.SetUID("pod-uid")
+		set.Spec.PersistentVolumeClaimRetentionPolicy = nil
+		if hasUnexpectedController(target, set, pod) {
+			t.Errorf("Any controller should be allowed when no retention policy (retain behavior) is specified. Incorrectly identified unexpected controller at %s", tc.name)
+		}
+		for _, policy := range []apps.StatefulSetPersistentVolumeClaimRetentionPolicy{
+			{WhenDeleted: "Retain", WhenScaled: "Delete"},
+			{WhenDeleted: "Delete", WhenScaled: "Retain"},
+			{WhenDeleted: "Delete", WhenScaled: "Delete"},
+		} {
+			set.Spec.PersistentVolumeClaimRetentionPolicy = &policy
+			got := hasUnexpectedController(target, set, pod)
+			if got != tc.shouldReportUnexpectedController {
+				t.Errorf("Unexpected controller mismatch at %s (policy %v)", tc.name, policy)
+			}
+		}
+	}
+}
+
+func TestNonController(t *testing.T) {
+	testCases := []struct {
+		name string
+		refs []metav1.OwnerReference
+		// The set and pod objets will be created with names "set" and "pod", respectively.
+		setUID        types.UID
+		podUID        types.UID
+		nonController bool
+	}{
+		{
+			// API validation should prevent two controllers from being set,
+			// but for completeness the semantics here are tested.
+			name: "set and pod controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod",
+					Controller: ptr.To(true),
+				},
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "set",
+					UID:        "set",
+					Controller: ptr.To(true),
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: false,
+		},
+		{
+			name: "set controller, pod noncontroller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod",
+				},
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "set",
+					UID:        "set",
+					Controller: ptr.To(true),
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: true,
+		},
+		{
+			name: "set noncontroller, pod controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod",
+					Controller: ptr.To(true),
+				},
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "set",
+					UID:        "set",
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: true,
+		},
+		{
+			name: "set controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "set",
+					UID:        "set",
+					Controller: ptr.To(true),
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: false,
+		},
+		{
+			name: "pod controller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "Pod",
+					Name:       "pod",
+					UID:        "pod",
+					Controller: ptr.To(true),
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: false,
+		},
+		{
+			name:          "nothing",
+			refs:          []metav1.OwnerReference{},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: false,
+		},
+		{
+			name: "set noncontroller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "set",
+					UID:        "set",
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: true,
+		},
+		{
+			name: "set noncontroller with ptr",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "apps/v1",
+					Kind:       "Set",
+					Name:       "set",
+					UID:        "set",
+					Controller: ptr.To(false),
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: true,
+		},
+		{
+			name: "pod noncontroller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "pod",
+					Name:       "pod",
+					UID:        "pod",
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: true,
+		},
+		{
+			name: "other noncontroller",
+			refs: []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "pod",
+					Name:       "pod",
+					UID:        "not-matching",
+				},
+			},
+			setUID:        "set",
+			podUID:        "pod",
+			nonController: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		claim := v1.PersistentVolumeClaim{}
+		claim.SetOwnerReferences(tc.refs)
+		pod := v1.Pod{}
+		pod.SetUID(tc.podUID)
+		pod.SetName("pod")
+		set := apps.StatefulSet{}
+		set.SetUID(tc.setUID)
+		set.SetName("set")
+		got := hasNonControllerOwner(&claim, &set, &pod)
+		if got != tc.nonController {
+			t.Errorf("Failed %s: got %t, expected %t", tc.name, got, tc.nonController)
+		}
 	}
 }
 
 func TestHasStaleOwnerRef(t *testing.T) {
-	target := v1.Pod{}
+	target := v1.PersistentVolumeClaim{}
 	target.SetOwnerReferences([]metav1.OwnerReference{
-		{Name: "bob", UID: "123"}, {Name: "shirley", UID: "456"}})
+		{Name: "bob", UID: "123", APIVersion: "v1", Kind: "Pod"},
+		{Name: "shirley", UID: "456", APIVersion: "v1", Kind: "Pod"},
+	})
 	ownerA := v1.Pod{}
 	ownerA.SetUID("123")
 	ownerA.Name = "bob"
@@ -491,93 +1424,17 @@ func TestHasStaleOwnerRef(t *testing.T) {
 	ownerC := v1.Pod{}
 	ownerC.Name = "yvonne"
 	ownerC.SetUID("345")
-	if hasStaleOwnerRef(&target, &ownerA) {
+	if hasStaleOwnerRef(&target, &ownerA, podKind) {
 		t.Error("ownerA should not be stale")
 	}
-	if !hasStaleOwnerRef(&target, &ownerB) {
+	if !hasStaleOwnerRef(&target, &ownerB, podKind) {
 		t.Error("ownerB should be stale")
 	}
-	if hasStaleOwnerRef(&target, &ownerC) {
+	if hasStaleOwnerRef(&target, &ownerC, podKind) {
 		t.Error("ownerC should not be stale")
 	}
 }
 
-func TestSetOwnerRef(t *testing.T) {
-	target := v1.Pod{}
-	ownerA := v1.Pod{}
-	ownerA.Name = "A"
-	ownerA.GetObjectMeta().SetUID("ABC")
-	if setOwnerRef(&target, &ownerA, &ownerA.TypeMeta) != true {
-		t.Errorf("Unexpected lack of update")
-	}
-	ownerRefs := target.GetObjectMeta().GetOwnerReferences()
-	if len(ownerRefs) != 1 {
-		t.Errorf("Unexpected owner ref count: %d", len(ownerRefs))
-	}
-	if ownerRefs[0].UID != "ABC" {
-		t.Errorf("Unexpected owner UID %v", ownerRefs[0].UID)
-	}
-	if setOwnerRef(&target, &ownerA, &ownerA.TypeMeta) != false {
-		t.Errorf("Unexpected update")
-	}
-	if len(target.GetObjectMeta().GetOwnerReferences()) != 1 {
-		t.Error("Unexpected duplicate reference")
-	}
-	ownerB := v1.Pod{}
-	ownerB.Name = "B"
-	ownerB.GetObjectMeta().SetUID("BCD")
-	if setOwnerRef(&target, &ownerB, &ownerB.TypeMeta) != true {
-		t.Error("Unexpected lack of second update")
-	}
-	ownerRefs = target.GetObjectMeta().GetOwnerReferences()
-	if len(ownerRefs) != 2 {
-		t.Errorf("Unexpected owner ref count: %d", len(ownerRefs))
-	}
-	if ownerRefs[0].UID != "ABC" || ownerRefs[1].UID != "BCD" {
-		t.Errorf("Bad second ownerRefs: %v", ownerRefs)
-	}
-}
-
-func TestRemoveOwnerRef(t *testing.T) {
-	target := v1.Pod{}
-	ownerA := v1.Pod{}
-	ownerA.Name = "A"
-	ownerA.GetObjectMeta().SetUID("ABC")
-	if removeOwnerRef(&target, &ownerA) != false {
-		t.Error("Unexpected update on empty remove")
-	}
-	setOwnerRef(&target, &ownerA, &ownerA.TypeMeta)
-	if removeOwnerRef(&target, &ownerA) != true {
-		t.Error("Unexpected lack of update")
-	}
-	if len(target.GetObjectMeta().GetOwnerReferences()) != 0 {
-		t.Error("Unexpected owner reference remains")
-	}
-
-	ownerB := v1.Pod{}
-	ownerB.Name = "B"
-	ownerB.GetObjectMeta().SetUID("BCD")
-
-	setOwnerRef(&target, &ownerA, &ownerA.TypeMeta)
-	if removeOwnerRef(&target, &ownerB) != false {
-		t.Error("Unexpected update for mismatched owner")
-	}
-	if len(target.GetObjectMeta().GetOwnerReferences()) != 1 {
-		t.Error("Missing ref after no-op remove")
-	}
-	setOwnerRef(&target, &ownerB, &ownerB.TypeMeta)
-	if removeOwnerRef(&target, &ownerA) != true {
-		t.Error("Missing update for second remove")
-	}
-	ownerRefs := target.GetObjectMeta().GetOwnerReferences()
-	if len(ownerRefs) != 1 {
-		t.Error("Extra ref after second remove")
-	}
-	if ownerRefs[0].UID != "BCD" {
-		t.Error("Bad UID after second remove")
-	}
-}
-
 func TestIsRunningAndReady(t *testing.T) {
 	set := newStatefulSet(3)
 	pod := newStatefulSetPod(set, 1)
diff --git a/test/e2e/apps/statefulset.go b/test/e2e/apps/statefulset.go
index 91b795cf71e92..a34cb004361ad 100644
--- a/test/e2e/apps/statefulset.go
+++ b/test/e2e/apps/statefulset.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"reflect"
 	"regexp"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -34,6 +35,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	klabels "k8s.io/apimachinery/pkg/labels"
@@ -59,6 +61,7 @@ import (
 	imageutils "k8s.io/kubernetes/test/utils/image"
 	admissionapi "k8s.io/pod-security-admission/api"
 	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -1339,6 +1342,84 @@ var _ = SIGDescribe("StatefulSet", func() {
 			framework.ExpectNoError(err)
 		})
 
+		ginkgo.It("should not delete PVC with OnScaledown policy if another controller owns the PVC", func(ctx context.Context) {
+			e2epv.SkipIfNoDefaultStorageClass(ctx, c)
+			ginkgo.By("Creating statefulset " + ssName + " in namespace " + ns)
+			*(ss.Spec.Replicas) = 3
+			_, err := c.AppsV1().StatefulSets(ns).Create(ctx, ss, metav1.CreateOptions{})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Confirm PVC has been created")
+			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{0, 1, 2})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Create configmap to use as dummy controller")
+			dummyConfigMap, err := c.CoreV1().ConfigMaps(ns).Create(ctx, &v1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					GenerateName: "dummy-controller",
+					Namespace:    ns,
+				},
+			}, metav1.CreateOptions{})
+			framework.ExpectNoError(err)
+			defer func() {
+				// Will be cleaned up with the namespace if this fails.
+				_ = c.CoreV1().ConfigMaps(ns).Delete(ctx, dummyConfigMap.Name, metav1.DeleteOptions{})
+			}()
+
+			ginkgo.By("Update PVC 1 owner ref")
+			pvc1Name := fmt.Sprintf("datadir-%s-1", ss.Name)
+			_, err = updatePVCWithRetries(ctx, c, ns, pvc1Name, func(update *v1.PersistentVolumeClaim) {
+				update.OwnerReferences = []metav1.OwnerReference{
+					{
+						Name:       dummyConfigMap.GetName(),
+						APIVersion: "v1",
+						Kind:       "ConfigMap",
+						UID:        dummyConfigMap.GetUID(),
+						Controller: ptr.To(true),
+					},
+				}
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Update StatefulSet retention policy")
+			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ss.Name, func(update *appsv1.StatefulSet) {
+				update.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{
+					WhenScaled: appsv1.DeletePersistentVolumeClaimRetentionPolicyType,
+				}
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Scale StatefulSet down to 0")
+			_, err = e2estatefulset.Scale(ctx, c, ss, 0)
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Verify PVC 1 still exists")
+			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{1})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Remove PVC 1 owner ref")
+			_, err = updatePVCWithRetries(ctx, c, ns, pvc1Name, func(update *v1.PersistentVolumeClaim) {
+				update.OwnerReferences = nil
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Scale set back up to 2")
+			_, err = e2estatefulset.Scale(ctx, c, ss, 2)
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Confirm PVCs scaled up as well")
+			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{0, 1})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Scale set down to 1")
+			_, err = e2estatefulset.Scale(ctx, c, ss, 1)
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Confirm PVC 1 deleted this time")
+			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{0})
+			framework.ExpectNoError(err)
+		})
+
 		ginkgo.It("should delete PVCs after adopting pod (WhenDeleted)", func(ctx context.Context) {
 			e2epv.SkipIfNoDefaultStorageClass(ctx, c)
 			ginkgo.By("Creating statefulset " + ssName + " in namespace " + ns)
@@ -1400,6 +1481,191 @@ var _ = SIGDescribe("StatefulSet", func() {
 			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{0})
 			framework.ExpectNoError(err)
 		})
+
+		ginkgo.It("should not delete PVCs when there is another controller", func(ctx context.Context) {
+			e2epv.SkipIfNoDefaultStorageClass(ctx, c)
+			ginkgo.By("Creating statefulset " + ssName + " with no retention policy in namespace " + ns)
+
+			*(ss.Spec.Replicas) = 4
+			_, err := c.AppsV1().StatefulSets(ns).Create(ctx, ss, metav1.CreateOptions{})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Confirming all 4 PVCs exist")
+			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{0, 1, 2, 3})
+			framework.ExpectNoError(err)
+
+			claimNames := make([]string, 4)
+			for i := 0; i < 4; i++ {
+				claimNames[i] = fmt.Sprintf("%s-%s-%d", statefulPodMounts[0].Name, ssName, i)
+			}
+
+			ginkgo.By("Create configmap to use as random owner")
+			randomConfigMap, err := c.CoreV1().ConfigMaps(ns).Create(ctx, &v1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					GenerateName: "random-owner",
+					Namespace:    ns,
+				},
+			}, metav1.CreateOptions{})
+			framework.ExpectNoError(err)
+			defer func() {
+				// Will be cleaned up by the namespace delete if this fails
+				_ = c.CoreV1().ConfigMaps(ns).Delete(ctx, randomConfigMap.Name, metav1.DeleteOptions{})
+			}()
+
+			ginkgo.By("Add external owner to PVC 1")
+			expectedExternalRef := []metav1.OwnerReference{
+				{
+					APIVersion: "v1",
+					Kind:       "ConfigMap",
+					Name:       randomConfigMap.GetName(),
+					UID:        randomConfigMap.GetUID(),
+				},
+			}
+			_, err = updatePVCWithRetries(ctx, c, ns, claimNames[1], func(update *v1.PersistentVolumeClaim) {
+				update.SetOwnerReferences(expectedExternalRef)
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Add stale statefulset controller to PVC 3, with finalizer to prevent garbage collection")
+			expectedStaleRef := []metav1.OwnerReference{
+				{
+					APIVersion:         "apps/v1",
+					Kind:               "StatefulSet",
+					Name:               "unknown",
+					UID:                "9d86d6ae-4e06-4ff1-bc55-f77f52e272e9",
+					Controller:         ptr.To(true),
+					BlockOwnerDeletion: ptr.To(true),
+				},
+			}
+			_, err = updatePVCWithRetries(ctx, c, ns, claimNames[3], func(update *v1.PersistentVolumeClaim) {
+				update.SetOwnerReferences(expectedStaleRef)
+				update.SetFinalizers([]string{"keep-with/stale-ref"})
+			})
+			framework.ExpectNoError(err)
+
+			defer func() {
+				if _, err := c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[3], metav1.GetOptions{}); apierrors.IsNotFound(err) {
+					return
+				}
+				_, err := updatePVCWithRetries(ctx, c, ns, claimNames[3], func(update *v1.PersistentVolumeClaim) {
+					update.SetFinalizers([]string{})
+				})
+				framework.ExpectNoError(err)
+			}()
+
+			ginkgo.By("Check references updated")
+			err = wait.PollUntilContextTimeout(ctx, e2estatefulset.StatefulSetPoll, e2estatefulset.StatefulSetTimeout, true, func(ctx context.Context) (bool, error) {
+				claim, err := c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[1], metav1.GetOptions{})
+				if err != nil {
+					return false, nil // retry
+				}
+				if !reflect.DeepEqual(claim.GetOwnerReferences(), expectedExternalRef) {
+					return false, nil // retry
+				}
+				claim, err = c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[3], metav1.GetOptions{})
+				if err != nil {
+					return false, nil // retry
+				}
+				if !reflect.DeepEqual(claim.GetOwnerReferences(), expectedStaleRef) {
+					return false, nil // retry
+				}
+				return true, nil // found them all!
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Update retention policy to delete to force claims to resync")
+			var ssUID types.UID
+			_, err = updateStatefulSetWithRetries(ctx, c, ns, ssName, func(update *appsv1.StatefulSet) {
+				update.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{
+					WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType,
+				}
+				ssUID = update.GetUID()
+			})
+			framework.ExpectNoError(err)
+
+			expectedOwnerRef := []metav1.OwnerReference{
+				{
+					APIVersion:         "apps/v1",
+					Kind:               "StatefulSet",
+					Name:               ssName,
+					UID:                ssUID,
+					Controller:         ptr.To(true),
+					BlockOwnerDeletion: ptr.To(true),
+				},
+			}
+			ginkgo.By("Expect claims 0, 1 and 2 to have ownerRefs to the statefulset, and 3 to have a stale reference")
+			err = wait.PollUntilContextTimeout(ctx, e2estatefulset.StatefulSetPoll, e2estatefulset.StatefulSetTimeout, true, func(ctx context.Context) (bool, error) {
+				for _, i := range []int{0, 2} {
+					claim, err := c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[i], metav1.GetOptions{})
+					if err != nil {
+						return false, nil // retry
+					}
+					if !reflect.DeepEqual(claim.GetOwnerReferences(), expectedOwnerRef) {
+						return false, nil // retry
+					}
+				}
+				claim, err := c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[1], metav1.GetOptions{})
+				if err != nil {
+					return false, nil // retry
+				}
+				// Claim 1's external owner is neither its pod nor its set, so it should get updated with a controller.
+				if !reflect.DeepEqual(claim.GetOwnerReferences(), slices.Concat(expectedExternalRef, expectedOwnerRef)) {
+					return false, nil // retry
+				}
+				claim, err = c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[3], metav1.GetOptions{})
+				if err != nil {
+					return false, nil // retry
+				}
+				if !reflect.DeepEqual(claim.GetOwnerReferences(), expectedStaleRef) {
+					return false, nil // retry
+				}
+				return true, nil // found them all!
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Remove controller flag from claim 0")
+			_, err = updatePVCWithRetries(ctx, c, ns, claimNames[0], func(update *v1.PersistentVolumeClaim) {
+				update.SetOwnerReferences([]metav1.OwnerReference{
+					{
+						APIVersion:         "apps/v1",
+						Kind:               "StatefulSet",
+						Name:               ssName,
+						UID:                ssUID,
+						Controller:         ptr.To(false),
+						BlockOwnerDeletion: ptr.To(true),
+					},
+				})
+			})
+			framework.ExpectNoError(err)
+
+			ginkgo.By("Update statefulset to provoke a reconcile")
+			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ssName, func(update *appsv1.StatefulSet) {
+				update.Spec.PersistentVolumeClaimRetentionPolicy = &appsv1.StatefulSetPersistentVolumeClaimRetentionPolicy{
+					WhenDeleted: appsv1.DeletePersistentVolumeClaimRetentionPolicyType,
+					WhenScaled:  appsv1.DeletePersistentVolumeClaimRetentionPolicyType,
+				}
+			})
+			framework.ExpectNoError(err)
+			ginkgo.By("Expect controller flag for claim 0 to reconcile back to true")
+			err = wait.PollUntilContextTimeout(ctx, e2estatefulset.StatefulSetPoll, e2estatefulset.StatefulSetTimeout, true, func(ctx context.Context) (bool, error) {
+				claim, err := c.CoreV1().PersistentVolumeClaims(ns).Get(ctx, claimNames[0], metav1.GetOptions{})
+				if err != nil {
+					return false, nil // retry
+				}
+				if reflect.DeepEqual(claim.GetOwnerReferences(), expectedOwnerRef) {
+					return true, nil // success!
+				}
+				return false, nil // retry
+			})
+			framework.ExpectNoError(err)
+
+			// Claim 1 has an external owner, and 3 has a finalizer still, so they will not be deleted.
+			ginkgo.By("Delete the stateful set and wait for claims 0 and 2 but not 1 and 3 to disappear")
+			err = c.AppsV1().StatefulSets(ns).Delete(ctx, ssName, metav1.DeleteOptions{})
+			framework.ExpectNoError(err)
+			err = verifyStatefulSetPVCsExist(ctx, c, ss, []int{1, 3})
+			framework.ExpectNoError(err)
+		})
 	})
 
 	ginkgo.Describe("Automatically recreate PVC for pending pod when PVC is missing", func() {
@@ -1587,7 +1853,7 @@ var _ = SIGDescribe("StatefulSet", func() {
 			framework.ExpectNoError(err)
 
 			ginkgo.By("Increasing .spec.ordinals.start = 4")
-			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ss.Name, func(update *appsv1.StatefulSet) {
+			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ssName, func(update *appsv1.StatefulSet) {
 				update.Spec.Ordinals = &appsv1.StatefulSetOrdinals{
 					Start: 4,
 				}
@@ -1621,7 +1887,7 @@ var _ = SIGDescribe("StatefulSet", func() {
 			framework.ExpectNoError(err)
 
 			ginkgo.By("Decreasing .spec.ordinals.start = 2")
-			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ss.Name, func(update *appsv1.StatefulSet) {
+			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ssName, func(update *appsv1.StatefulSet) {
 				update.Spec.Ordinals = &appsv1.StatefulSetOrdinals{
 					Start: 2,
 				}
@@ -1654,14 +1920,14 @@ var _ = SIGDescribe("StatefulSet", func() {
 			framework.ExpectNoError(err)
 
 			ginkgo.By("Removing .spec.ordinals")
-			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ss.Name, func(update *appsv1.StatefulSet) {
+			ss, err = updateStatefulSetWithRetries(ctx, c, ns, ssName, func(update *appsv1.StatefulSet) {
 				update.Spec.Ordinals = nil
 			})
 			framework.ExpectNoError(err)
 
 			// since we are replacing 2 pods for 2, we need to ensure we wait
 			// for the new ones to show up, not just for any random 2
-			framework.Logf("Confirming 2 replicas, with start ordinal 0")
+			ginkgo.By("Confirming 2 replicas, with start ordinal 0")
 			waitForStatus(ctx, c, ss)
 			waitForPodNames(ctx, c, ss, []string{"ss-0", "ss-1"})
 			e2estatefulset.WaitForStatusReplicas(ctx, c, ss, 2)
@@ -2231,7 +2497,7 @@ type updateStatefulSetFunc func(*appsv1.StatefulSet)
 func updateStatefulSetWithRetries(ctx context.Context, c clientset.Interface, namespace, name string, applyUpdate updateStatefulSetFunc) (statefulSet *appsv1.StatefulSet, err error) {
 	statefulSets := c.AppsV1().StatefulSets(namespace)
 	var updateErr error
-	pollErr := wait.PollWithContext(ctx, 10*time.Millisecond, 1*time.Minute, func(ctx context.Context) (bool, error) {
+	pollErr := wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, 1*time.Minute, true, func(ctx context.Context) (bool, error) {
 		if statefulSet, err = statefulSets.Get(ctx, name, metav1.GetOptions{}); err != nil {
 			return false, err
 		}
@@ -2245,11 +2511,34 @@ func updateStatefulSetWithRetries(ctx context.Context, c clientset.Interface, na
 		return false, nil
 	})
 	if wait.Interrupted(pollErr) {
-		pollErr = fmt.Errorf("couldn't apply the provided updated to stateful set %q: %v", name, updateErr)
+		pollErr = fmt.Errorf("couldn't apply the provided updated to stateful set %q: %w", name, updateErr)
 	}
 	return statefulSet, pollErr
 }
 
+// updatePVCWithRetries updates PVCs with retries.
+func updatePVCWithRetries(ctx context.Context, c clientset.Interface, namespace, name string, applyUpdate func(*v1.PersistentVolumeClaim)) (pvc *v1.PersistentVolumeClaim, err error) {
+	pvcs := c.CoreV1().PersistentVolumeClaims(namespace)
+	var updateErr error
+	pollErr := wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, 1*time.Minute, true, func(ctx context.Context) (bool, error) {
+		if pvc, err = pvcs.Get(ctx, name, metav1.GetOptions{}); err != nil {
+			return false, err
+		}
+		// Apply the update, then attempt to push it to the apiserver.
+		applyUpdate(pvc)
+		if pvc, err = pvcs.Update(ctx, pvc, metav1.UpdateOptions{}); err == nil {
+			framework.Logf("Updating pvc %s", name)
+			return true, nil
+		}
+		updateErr = err
+		return false, nil
+	})
+	if wait.Interrupted(pollErr) {
+		pollErr = fmt.Errorf("couldn't apply the provided updated to stateful set %q: %w", name, updateErr)
+	}
+	return pvc, pollErr
+}
+
 // getStatefulSet gets the StatefulSet named name in namespace.
 func getStatefulSet(ctx context.Context, c clientset.Interface, namespace, name string) *appsv1.StatefulSet {
 	ss, err := c.AppsV1().StatefulSets(namespace).Get(ctx, name, metav1.GetOptions{})