diff --git a/pkg/controller/statefulset/stateful_set_utils.go b/pkg/controller/statefulset/stateful_set_utils.go
index d4a80cd01eb..1f35971ae63 100644
--- a/pkg/controller/statefulset/stateful_set_utils.go
+++ b/pkg/controller/statefulset/stateful_set_utils.go
@@ -587,8 +587,9 @@ func inconsistentStatus(set *apps.StatefulSet, status *apps.StatefulSetStatus) b
 // are set to 0.
 func completeRollingUpdate(set *apps.StatefulSet, status *apps.StatefulSetStatus) {
 	if set.Spec.UpdateStrategy.Type == apps.RollingUpdateStatefulSetStrategyType &&
-		status.UpdatedReplicas == status.Replicas &&
-		status.ReadyReplicas == status.Replicas {
+		status.UpdatedReplicas == *set.Spec.Replicas &&
+		status.ReadyReplicas == *set.Spec.Replicas &&
+		status.Replicas == *set.Spec.Replicas {
 		status.CurrentReplicas = status.UpdatedReplicas
 		status.CurrentRevision = status.UpdateRevision
 	}
diff --git a/test/e2e/apps/statefulset.go b/test/e2e/apps/statefulset.go
index baa36de7638..cf61f6b0cbf 100644
--- a/test/e2e/apps/statefulset.go
+++ b/test/e2e/apps/statefulset.go
@@ -77,6 +77,8 @@ const (
 	statefulSetTimeout = 10 * time.Minute
 	// statefulPodTimeout is a timeout for stateful pods to change state
 	statefulPodTimeout = 5 * time.Minute
+
+	testFinalizer = "example.com/test-finalizer"
 )
 
 var httpProbe = &v1.Probe{
@@ -510,6 +512,51 @@ var _ = SIGDescribe("StatefulSet", func() {
 
 		})
 
+		ginkgo.It("should perform canary updates and phased rolling updates of template modifications for partiton1 and delete pod-0 without failing container", func(ctx context.Context) {
+			ginkgo.By("Creating a new StatefulSet without failing container")
+			ss := e2estatefulset.NewStatefulSet("ss2", ns, headlessSvcName, 3, nil, nil, labels)
+			deletingPodForRollingUpdatePartitionTest(ctx, f, c, ns, ss)
+		})
+
+		ginkgo.It("should perform canary updates and phased rolling updates of template modifications for partiton1 and delete pod-0 with failing container", func(ctx context.Context) {
+			ginkgo.By("Creating a new StatefulSet with failing container")
+			ss := e2estatefulset.NewStatefulSet("ss3", ns, headlessSvcName, 3, nil, nil, labels)
+			ss.Spec.Template.Spec.Containers = append(ss.Spec.Template.Spec.Containers, v1.Container{
+				Name:    "sleep-exit-with-1",
+				Image:   imageutils.GetE2EImage(imageutils.BusyBox),
+				Command: []string{"sh", "-c"},
+				Args: []string{`
+						echo "Running in pod $POD_NAME"
+						_term(){
+							echo "Received SIGTERM signal"
+							if [ "${POD_NAME}" = "ss3-0" ]; then
+								exit 1
+							else
+								exit 0
+							fi
+						}
+						trap _term SIGTERM
+						while true; do
+							echo "Running in infinite loop in $POD_NAME"
+							sleep 1
+						done
+						`,
+				},
+				Env: []v1.EnvVar{
+					{
+						Name: "POD_NAME",
+						ValueFrom: &v1.EnvVarSource{
+							FieldRef: &v1.ObjectFieldSelector{
+								APIVersion: "v1",
+								FieldPath:  "metadata.name",
+							},
+						},
+					},
+				},
+			})
+			deletingPodForRollingUpdatePartitionTest(ctx, f, c, ns, ss)
+		})
+
 		// Do not mark this as Conformance.
 		// The legacy OnDelete strategy only exists for backward compatibility with pre-v1 APIs.
 		ginkgo.It("should implement legacy replacement when the update strategy is OnDelete", func(ctx context.Context) {
@@ -1945,6 +1992,144 @@ func rollbackTest(ctx context.Context, c clientset.Interface, ns string, ss *app
 	}
 }
 
+// This function is used canary updates and phased rolling updates of template modifications for partiton1 and delete pod-0
+func deletingPodForRollingUpdatePartitionTest(ctx context.Context, f *framework.Framework, c clientset.Interface, ns string, ss *appsv1.StatefulSet) {
+	setHTTPProbe(ss)
+	ss.Spec.UpdateStrategy = appsv1.StatefulSetUpdateStrategy{
+		Type: appsv1.RollingUpdateStatefulSetStrategyType,
+		RollingUpdate: func() *appsv1.RollingUpdateStatefulSetStrategy {
+			return &appsv1.RollingUpdateStatefulSetStrategy{
+				Partition: pointer.Int32(1),
+			}
+		}(),
+	}
+	ss, err := c.AppsV1().StatefulSets(ns).Create(ctx, ss, metav1.CreateOptions{})
+	framework.ExpectNoError(err)
+	e2estatefulset.WaitForRunningAndReady(ctx, c, *ss.Spec.Replicas, ss)
+	ss = waitForStatus(ctx, c, ss)
+	currentRevision, updateRevision := ss.Status.CurrentRevision, ss.Status.UpdateRevision
+	gomega.Expect(currentRevision).To(gomega.Equal(updateRevision), fmt.Sprintf("StatefulSet %s/%s created with update revision %s not equal to current revision %s",
+		ss.Namespace, ss.Name, updateRevision, currentRevision))
+	pods := e2estatefulset.GetPodList(ctx, c, ss)
+	for i := range pods.Items {
+		gomega.Expect(pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel]).To(gomega.Equal(currentRevision), fmt.Sprintf("Pod %s/%s revision %s is not equal to currentRevision %s",
+			pods.Items[i].Namespace,
+			pods.Items[i].Name,
+			pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel],
+			currentRevision))
+	}
+
+	ginkgo.By("Adding finalizer for pod-0")
+	pod0name := getStatefulSetPodNameAtIndex(0, ss)
+	pod0, err := c.CoreV1().Pods(ns).Get(ctx, pod0name, metav1.GetOptions{})
+	framework.ExpectNoError(err)
+	pod0.Finalizers = append(pod0.Finalizers, testFinalizer)
+	pod0, err = c.CoreV1().Pods(ss.Namespace).Update(ctx, pod0, metav1.UpdateOptions{})
+	framework.ExpectNoError(err)
+	pods.Items[0] = *pod0
+	defer e2epod.NewPodClient(f).RemoveFinalizer(ctx, pod0.Name, testFinalizer)
+
+	ginkgo.By("Updating image on StatefulSet")
+	newImage := NewWebserverImage
+	oldImage := ss.Spec.Template.Spec.Containers[0].Image
+	ginkgo.By(fmt.Sprintf("Updating stateful set template: update image from %s to %s", oldImage, newImage))
+	gomega.Expect(oldImage).ToNot(gomega.Equal(newImage), "Incorrect test setup: should update to a different image")
+	ss, err = updateStatefulSetWithRetries(ctx, c, ns, ss.Name, func(update *appsv1.StatefulSet) {
+		update.Spec.Template.Spec.Containers[0].Image = newImage
+	})
+	framework.ExpectNoError(err)
+
+	ginkgo.By("Creating a new revision")
+	ss = waitForStatus(ctx, c, ss)
+	currentRevision, updateRevision = ss.Status.CurrentRevision, ss.Status.UpdateRevision
+	gomega.Expect(currentRevision).ToNot(gomega.Equal(updateRevision), "Current revision should not equal update revision during rolling update")
+
+	ginkgo.By("Await for all replicas running, all are updated but pod-0")
+	e2estatefulset.WaitForState(ctx, c, ss, func(set2 *appsv1.StatefulSet, pods2 *v1.PodList) (bool, error) {
+		ss = set2
+		pods = pods2
+		if ss.Status.UpdatedReplicas == *ss.Spec.Replicas-1 && ss.Status.Replicas == *ss.Spec.Replicas && ss.Status.ReadyReplicas == *ss.Spec.Replicas {
+			// rolling updated is not completed, because replica 0 isn't ready
+			return true, nil
+		}
+		return false, nil
+	})
+
+	ginkgo.By("Verify pod images before pod-0 deletion and recreation")
+	for i := range pods.Items {
+		if i < int(*ss.Spec.UpdateStrategy.RollingUpdate.Partition) {
+			gomega.Expect(pods.Items[i].Spec.Containers[0].Image).To(gomega.Equal(oldImage), fmt.Sprintf("Pod %s/%s has image %s not equal to oldimage image %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Spec.Containers[0].Image,
+				oldImage))
+			gomega.Expect(pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel]).To(gomega.Equal(currentRevision), fmt.Sprintf("Pod %s/%s has revision %s not equal to current revision %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel],
+				currentRevision))
+		} else {
+			gomega.Expect(pods.Items[i].Spec.Containers[0].Image).To(gomega.Equal(newImage), fmt.Sprintf("Pod %s/%s has image %s not equal to new image  %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Spec.Containers[0].Image,
+				newImage))
+			gomega.Expect(pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel]).To(gomega.Equal(updateRevision), fmt.Sprintf("Pod %s/%s has revision %s not equal to new revision %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel],
+				updateRevision))
+		}
+	}
+
+	ginkgo.By("Deleting the pod-0 so that kubelet terminates it and StatefulSet controller recreates it")
+	deleteStatefulPodAtIndex(ctx, c, 0, ss)
+	ginkgo.By("Await for two replicas to be updated, while the pod-0 is not running")
+	e2estatefulset.WaitForState(ctx, c, ss, func(set2 *appsv1.StatefulSet, pods2 *v1.PodList) (bool, error) {
+		ss = set2
+		pods = pods2
+		return ss.Status.ReadyReplicas == *ss.Spec.Replicas-1, nil
+	})
+
+	ginkgo.By(fmt.Sprintf("Removing finalizer from pod-0 (%v/%v) to allow recreation", pod0.Namespace, pod0.Name))
+	e2epod.NewPodClient(f).RemoveFinalizer(ctx, pod0.Name, testFinalizer)
+
+	ginkgo.By("Await for recreation of pod-0, so that all replicas are running")
+	e2estatefulset.WaitForState(ctx, c, ss, func(set2 *appsv1.StatefulSet, pods2 *v1.PodList) (bool, error) {
+		ss = set2
+		pods = pods2
+		return ss.Status.ReadyReplicas == *ss.Spec.Replicas, nil
+	})
+
+	ginkgo.By("Verify pod images after pod-0 deletion and recreation")
+	pods = e2estatefulset.GetPodList(ctx, c, ss)
+	for i := range pods.Items {
+		if i < int(*ss.Spec.UpdateStrategy.RollingUpdate.Partition) {
+			gomega.Expect(pods.Items[i].Spec.Containers[0].Image).To(gomega.Equal(oldImage), fmt.Sprintf("Pod %s/%s has image %s not equal to current image %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Spec.Containers[0].Image,
+				oldImage))
+			gomega.Expect(pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel]).To(gomega.Equal(currentRevision), fmt.Sprintf("Pod %s/%s has revision %s not equal to current revision %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel],
+				currentRevision))
+		} else {
+			gomega.Expect(pods.Items[i].Spec.Containers[0].Image).To(gomega.Equal(newImage), fmt.Sprintf("Pod %s/%s has image %s not equal to new image  %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Spec.Containers[0].Image,
+				newImage))
+			gomega.Expect(pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel]).To(gomega.Equal(updateRevision), fmt.Sprintf("Pod %s/%s has revision %s not equal to new revision %s",
+				pods.Items[i].Namespace,
+				pods.Items[i].Name,
+				pods.Items[i].Labels[appsv1.StatefulSetRevisionLabel],
+				updateRevision))
+		}
+	}
+}
+
 // confirmStatefulPodCount asserts that the current number of Pods in ss is count, waiting up to timeout for ss to
 // scale to count.
 func confirmStatefulPodCount(ctx context.Context, c clientset.Interface, count int, ss *appsv1.StatefulSet, timeout time.Duration, hard bool) {
diff --git a/test/integration/statefulset/statefulset_test.go b/test/integration/statefulset/statefulset_test.go
index 065115ed828..4bd42ff0f12 100644
--- a/test/integration/statefulset/statefulset_test.go
+++ b/test/integration/statefulset/statefulset_test.go
@@ -43,6 +43,7 @@ import (
 	"k8s.io/kubernetes/pkg/controlplane"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/test/integration/framework"
+	"k8s.io/utils/ptr"
 )
 
 const (
@@ -489,6 +490,146 @@ func TestAutodeleteOwnerRefs(t *testing.T) {
 	}
 }
 
+func TestDeletingPodForRollingUpdatePartition(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
+	closeFn, rm, informers, c := scSetup(ctx, t)
+	defer closeFn()
+	ns := framework.CreateNamespaceOrDie(c, "test-deleting-pod-for-rolling-update-partition", t)
+	defer framework.DeleteNamespaceOrDie(c, ns, t)
+	cancel := runControllerAndInformers(rm, informers)
+	defer cancel()
+
+	labelMap := labelMap()
+	sts := newSTS("sts", ns.Name, 2)
+	sts.Spec.UpdateStrategy = appsv1.StatefulSetUpdateStrategy{
+		Type: appsv1.RollingUpdateStatefulSetStrategyType,
+		RollingUpdate: func() *appsv1.RollingUpdateStatefulSetStrategy {
+			return &appsv1.RollingUpdateStatefulSetStrategy{
+				Partition: ptr.To[int32](1),
+			}
+		}(),
+	}
+	stss, _ := createSTSsPods(t, c, []*appsv1.StatefulSet{sts}, []*v1.Pod{})
+	sts = stss[0]
+	waitSTSStable(t, c, sts)
+
+	// Verify STS creates 2 pods
+	podClient := c.CoreV1().Pods(ns.Name)
+	pods := getPods(t, podClient, labelMap)
+	if len(pods.Items) != 2 {
+		t.Fatalf("len(pods) = %d, want 2", len(pods.Items))
+	}
+	// Setting all pods in Running, Ready, and Available
+	setPodsReadyCondition(t, c, &v1.PodList{Items: pods.Items}, v1.ConditionTrue, time.Now())
+
+	// 1. Roll out a new image.
+	oldImage := sts.Spec.Template.Spec.Containers[0].Image
+	newImage := "new-image"
+	if oldImage == newImage {
+		t.Fatalf("bad test setup, statefulSet %s roll out with the same image", sts.Name)
+	}
+	// Set finalizers for the pod-0 to trigger pod recreation failure while the status UpdateRevision is bumped
+	pod0 := &pods.Items[0]
+	updatePod(t, podClient, pod0.Name, func(pod *v1.Pod) {
+		pod.Finalizers = []string{"fake.example.com/blockDeletion"}
+	})
+
+	stsClient := c.AppsV1().StatefulSets(ns.Name)
+	_ = updateSTS(t, stsClient, sts.Name, func(sts *appsv1.StatefulSet) {
+		sts.Spec.Template.Spec.Containers[0].Image = newImage
+	})
+
+	// Await for the pod-1 to be recreated, while pod-0 remains running
+	if err := wait.PollUntilContextTimeout(ctx, pollInterval, pollTimeout, false, func(ctx context.Context) (bool, error) {
+		ss, err := stsClient.Get(ctx, sts.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		pods := getPods(t, podClient, labelMap)
+		recreatedPods := v1.PodList{}
+		for _, pod := range pods.Items {
+			if pod.Status.Phase == v1.PodPending {
+				recreatedPods.Items = append(recreatedPods.Items, pod)
+			}
+		}
+		setPodsReadyCondition(t, c, &v1.PodList{Items: recreatedPods.Items}, v1.ConditionTrue, time.Now())
+		return ss.Status.UpdatedReplicas == *ss.Spec.Replicas-*sts.Spec.UpdateStrategy.RollingUpdate.Partition && ss.Status.Replicas == *ss.Spec.Replicas && ss.Status.ReadyReplicas == *ss.Spec.Replicas, nil
+	}); err != nil {
+		t.Fatalf("failed to await for pod-1 to be recreated by sts %s: %v", sts.Name, err)
+	}
+
+	// Mark pod-0 as terminal and not ready
+	updatePodStatus(t, podClient, pod0.Name, func(pod *v1.Pod) {
+		pod.Status.Phase = v1.PodFailed
+	})
+
+	// Make sure pod-0 gets deletion timestamp so that it is recreated
+	if err := c.CoreV1().Pods(ns.Name).Delete(context.TODO(), pod0.Name, metav1.DeleteOptions{}); err != nil {
+		t.Fatalf("error deleting pod %s: %v", pod0.Name, err)
+	}
+
+	// Await for pod-0 to be not ready
+	if err := wait.PollUntilContextTimeout(ctx, pollInterval, pollTimeout, false, func(ctx context.Context) (bool, error) {
+		ss, err := stsClient.Get(ctx, sts.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		return ss.Status.ReadyReplicas == *ss.Spec.Replicas-1, nil
+	}); err != nil {
+		t.Fatalf("failed to await for pod-0 to be not counted as ready in status of sts %s: %v", sts.Name, err)
+	}
+
+	// Remove the finalizer to allow recreation
+	updatePod(t, podClient, pod0.Name, func(pod *v1.Pod) {
+		pod.Finalizers = []string{}
+	})
+
+	// Await for pod-0 to be recreated and make it running
+	if err := wait.PollUntilContextTimeout(ctx, pollInterval, pollTimeout, false, func(ctx context.Context) (bool, error) {
+		pods := getPods(t, podClient, labelMap)
+		recreatedPods := v1.PodList{}
+		for _, pod := range pods.Items {
+			if pod.Status.Phase == v1.PodPending {
+				recreatedPods.Items = append(recreatedPods.Items, pod)
+			}
+		}
+		setPodsReadyCondition(t, c, &v1.PodList{Items: recreatedPods.Items}, v1.ConditionTrue, time.Now().Add(-120*time.Minute))
+		return len(recreatedPods.Items) > 0, nil
+	}); err != nil {
+		t.Fatalf("failed to await for pod-0 to be recreated by sts %s: %v", sts.Name, err)
+	}
+
+	// Await for all stateful set status to record all replicas as ready
+	if err := wait.PollUntilContextTimeout(ctx, pollInterval, pollTimeout, false, func(ctx context.Context) (bool, error) {
+		ss, err := stsClient.Get(ctx, sts.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		return ss.Status.ReadyReplicas == *ss.Spec.Replicas, nil
+	}); err != nil {
+		t.Fatalf("failed to verify .Spec.Template.Spec.Containers[0].Image is updated for sts %s: %v", sts.Name, err)
+	}
+
+	// Verify 3 pods exist
+	pods = getPods(t, podClient, labelMap)
+	if len(pods.Items) != int(*sts.Spec.Replicas) {
+		t.Fatalf("Unexpected number of pods")
+	}
+
+	// Verify pod images
+	for i := range pods.Items {
+		if i < int(*sts.Spec.UpdateStrategy.RollingUpdate.Partition) {
+			if pods.Items[i].Spec.Containers[0].Image != oldImage {
+				t.Fatalf("Pod %s has image %s not equal to old image %s", pods.Items[i].Name, pods.Items[i].Spec.Containers[0].Image, oldImage)
+			}
+		} else {
+			if pods.Items[i].Spec.Containers[0].Image != newImage {
+				t.Fatalf("Pod %s has image %s not equal to new image %s", pods.Items[i].Name, pods.Items[i].Spec.Containers[0].Image, newImage)
+			}
+		}
+	}
+}
+
 func TestStatefulSetStartOrdinal(t *testing.T) {
 	tests := []struct {
 		ordinals         *appsv1.StatefulSetOrdinals