Repository: apecloud/kubeblocks

Issue:
[BUG] InstanceSet's LEADER colume is incorrect
In KB 1.0-beta 32, Create a PG Cluster, it is running  with expected roles.

When get InstanceSet, the 'LEADER' column i blank.


Hints from issue comments:
@shanshanying: LEADER colume will be removed.

Solution patch:
diff --git a/apis/apps/v1/componentdefinition_types.go b/apis/apps/v1/componentdefinition_types.go
index 3dae6e088..9cd0a19af 100644
--- a/apis/apps/v1/componentdefinition_types.go
+++ b/apis/apps/v1/componentdefinition_types.go
@@ -1135,16 +1135,6 @@ type ComponentTemplateSpec struct {
 	DefaultMode *int32 `json:"defaultMode,omitempty" protobuf:"varint,3,opt,name=defaultMode"`
 }
 
-// RerenderResourceType defines the resource requirements for a component.
-// +enum
-// +kubebuilder:validation:Enum={vscale,hscale,tls}
-type RerenderResourceType string
-
-const (
-	ComponentVScaleType RerenderResourceType = "vscale"
-	ComponentHScaleType RerenderResourceType = "hscale"
-)
-
 type LogConfig struct {
 	// Specifies a descriptive label for the log type, such as 'slow' for a MySQL slow log file.
 	// It provides a clear identification of the log's purpose and content.
diff --git a/apis/parameters/v1alpha1/types.go b/apis/parameters/v1alpha1/types.go
index 86a8193cb..7eeddb3fe 100644
--- a/apis/parameters/v1alpha1/types.go
+++ b/apis/parameters/v1alpha1/types.go
@@ -120,13 +120,14 @@ const (
 
 // RerenderResourceType defines the resource requirements for a component.
 // +enum
-// +kubebuilder:validation:Enum={vscale,hscale,tls}
+// +kubebuilder:validation:Enum={vscale,hscale,tls,shardingHVScale}
 type RerenderResourceType string
 
 const (
-	ComponentVScaleType RerenderResourceType = "vscale"
-	ComponentHScaleType RerenderResourceType = "hscale"
-	ComponentTLSType    RerenderResourceType = "tls"
+	ComponentVScaleType          RerenderResourceType = "vscale"
+	ComponentHScaleType          RerenderResourceType = "hscale"
+	ComponentTLSType             RerenderResourceType = "tls"
+	ShardingComponentHVScaleType RerenderResourceType = "shardingHVScale"
 )
 
 // ParameterPhase defines the Configuration FSM phase
diff --git a/apis/workloads/v1/instanceset_types.go b/apis/workloads/v1/instanceset_types.go
index 1d4cb5a8f..3338e3a61 100644
--- a/apis/workloads/v1/instanceset_types.go
+++ b/apis/workloads/v1/instanceset_types.go
@@ -34,9 +34,10 @@ import (
 // +kubebuilder:subresource:scale:specpath=.spec.replicas,statuspath=.status.replicas
 // +kubebuilder:storageversion
 // +kubebuilder:resource:categories={kubeblocks},shortName=its
-// +kubebuilder:printcolumn:name="LEADER",type="string",JSONPath=".status.membersStatus[?(@.role.isLeader==true)].podName",description="leader instance name."
 // +kubebuilder:printcolumn:name="READY",type="string",JSONPath=".status.readyReplicas",description="ready replicas."
-// +kubebuilder:printcolumn:name="REPLICAS",type="string",JSONPath=".status.replicas",description="total replicas."
+// +kubebuilder:printcolumn:name="DESIRED",type="string",JSONPath=".spec.replicas",description="desired replicas."
+// +kubebuilder:printcolumn:name="UP-TO-DATE",type="string",JSONPath=".status.updatedReplicas",description="updated replicas."
+// +kubebuilder:printcolumn:name="AVAILABLE",type="string",JSONPath=".status.availableReplicas",description="available replicas, which are ready for at least minReadySeconds."
 // +kubebuilder:printcolumn:name="AGE",type="date",JSONPath=".metadata.creationTimestamp"
 
 // InstanceSet is the Schema for the instancesets API.
diff --git a/cmd/manager/main.go b/cmd/manager/main.go
index 9344457dc..757863967 100644
--- a/cmd/manager/main.go
+++ b/cmd/manager/main.go
@@ -375,7 +375,7 @@ func main() {
 		}),
 		Client: client.Options{
 			Cache: &client.CacheOptions{
-				DisableFor: intctrlutil.GetUncachedObjects(),
+				DisableFor: append(intctrlutil.GetUncachedObjects(), &parametersv1alpha1.ComponentParameter{}),
 			},
 		},
 	})
diff --git a/config/crd/bases/parameters.kubeblocks.io_paramconfigrenderers.yaml b/config/crd/bases/parameters.kubeblocks.io_paramconfigrenderers.yaml
index e7bd0fba9..b15c96830 100644
--- a/config/crd/bases/parameters.kubeblocks.io_paramconfigrenderers.yaml
+++ b/config/crd/bases/parameters.kubeblocks.io_paramconfigrenderers.yaml
@@ -143,6 +143,7 @@ spec:
                         - vscale
                         - hscale
                         - tls
+                        - shardingHVScale
                         type: string
                       type: array
                       x-kubernetes-list-type: set
diff --git a/config/crd/bases/workloads.kubeblocks.io_instancesets.yaml b/config/crd/bases/workloads.kubeblocks.io_instancesets.yaml
index a0bd7f345..5877f752e 100644
--- a/config/crd/bases/workloads.kubeblocks.io_instancesets.yaml
+++ b/config/crd/bases/workloads.kubeblocks.io_instancesets.yaml
@@ -20,17 +20,21 @@ spec:
   scope: Namespaced
   versions:
   - additionalPrinterColumns:
-    - description: leader instance name.
-      jsonPath: .status.membersStatus[?(@.role.isLeader==true)].podName
-      name: LEADER
-      type: string
     - description: ready replicas.
       jsonPath: .status.readyReplicas
       name: READY
       type: string
-    - description: total replicas.
-      jsonPath: .status.replicas
-      name: REPLICAS
+    - description: desired replicas.
+      jsonPath: .spec.replicas
+      name: DESIRED
+      type: string
+    - description: updated replicas.
+      jsonPath: .status.updatedReplicas
+      name: UP-TO-DATE
+      type: string
+    - description: available replicas, which are ready for at least minReadySeconds.
+      jsonPath: .status.availableReplicas
+      name: AVAILABLE
       type: string
     - jsonPath: .metadata.creationTimestamp
       name: AGE
diff --git a/controllers/apps/cluster/transformer_cluster_component_status.go b/controllers/apps/cluster/transformer_cluster_component_status.go
index 5a1e65d2f..581e155bf 100644
--- a/controllers/apps/cluster/transformer_cluster_component_status.go
+++ b/controllers/apps/cluster/transformer_cluster_component_status.go
@@ -268,7 +268,11 @@ func (t *clusterComponentStatusTransformer) shardingPhaseNMessage(comps []*appsv
 		return "", map[string]string{"reason": "the component objects are not found"}
 	}
 
-	phase := appsv1.ComponentPhase(composeClusterPhase(statusList))
+	composedPhase := composeClusterPhase(statusList)
+	if composedPhase == appsv1.AbnormalClusterPhase {
+		composedPhase = appsv1.FailedClusterPhase
+	}
+	phase := appsv1.ComponentPhase(composedPhase)
 	return phase, phasedMessage[phase]
 }
 
diff --git a/controllers/apps/component/transformer_component_deletion.go b/controllers/apps/component/transformer_component_deletion.go
index 9bb578e3e..584188a54 100644
--- a/controllers/apps/component/transformer_component_deletion.go
+++ b/controllers/apps/component/transformer_component_deletion.go
@@ -129,14 +129,15 @@ func (t *componentDeletionTransformer) deleteCompResources(transCtx *componentTr
 			if appsutil.IsOwnedByInstanceSet(object) {
 				continue
 			}
-			skipDeletion, err := handleRBACResourceDeletion(object, transCtx, comp, graphCli, dag)
-			if err != nil {
-				return fmt.Errorf("handle rbac deletion failed: %w", err)
-			}
-			if skipDeletion {
-				continue
+
+			switch object.(type) {
+			case *corev1.ServiceAccount, *rbacv1.Role, *rbacv1.RoleBinding:
+				if err := handleRBACResourceDeletion(object, transCtx, comp, graphCli, dag, matchLabels); err != nil {
+					return fmt.Errorf("handle rbac deletion failed: %w", err)
+				}
+			default:
+				graphCli.Delete(dag, object)
 			}
-			graphCli.Delete(dag, object)
 		}
 		graphCli.Status(dag, comp, transCtx.Component)
 		return intctrlutil.NewRequeueError(time.Second*1, "not all component sub-resources deleted")
@@ -158,61 +159,24 @@ func (t *componentDeletionTransformer) deleteCompResources(transCtx *componentTr
 }
 
 func handleRBACResourceDeletion(obj client.Object, transCtx *componentTransformContext, comp *appsv1.Component,
-	graphCli model.GraphClient, dag *graph.DAG) (skipDeletion bool, err error) {
-	switch v := obj.(type) {
-	case *corev1.ServiceAccount, *rbacv1.Role, *rbacv1.RoleBinding:
-		// list other components that reference the same componentdefinition
-		transCtx.Logger.V(1).Info("handling rbac resources deletion",
-			"comp", comp.Name, "name", klog.KObj(v).String())
-		compDefName := comp.Spec.CompDef
-		compList := &appsv1.ComponentList{}
-		if err := transCtx.Client.List(transCtx.Context, compList, client.InNamespace(comp.Namespace),
-			client.MatchingLabels{constant.ComponentDefinitionLabelKey: compDefName}); err != nil {
-			return false, err
-		}
-		// if any, transfer ownership to any other component
-		for _, otherComp := range compList.Items {
-			// skip current component
-			if otherComp.Name == comp.Name {
-				continue
-			}
-			// skip deleting component
-			if !otherComp.DeletionTimestamp.IsZero() {
-				continue
-			}
+	graphCli model.GraphClient, dag *graph.DAG, matchLabels map[string]string) (err error) {
+	// orphan a rbac resource so that it can be adopted by another component
+	// this means these resources won't get automatically deleted
+	if err := controllerutil.RemoveControllerReference(comp, obj, model.GetScheme()); err != nil {
+		return err
+	}
+	for k := range matchLabels {
+		delete(obj.GetLabels(), k)
+	}
 
-			if err := controllerutil.RemoveControllerReference(comp, v, model.GetScheme()); err != nil {
-				return false, err
-			}
-			if err := controllerutil.SetControllerReference(&otherComp, v, model.GetScheme()); err != nil {
-				return false, err
-			}
-			// component controller selects a comp's subresource by labels, so change them too
-			clusterName, err := component.GetClusterName(&otherComp)
-			if err != nil {
-				return false, err
-			}
-			compShortName, err := component.ShortName(clusterName, otherComp.Name)
-			if err != nil {
-				return false, err
-			}
-			newLabels := constant.GetCompLabels(clusterName, compShortName)
-			for k, val := range newLabels {
-				v.GetLabels()[k] = val
-			}
-			graphCli.Update(dag, nil, v)
-			gvk, err := apiutil.GVKForObject(v, model.GetScheme())
-			if err != nil {
-				return false, err
-			}
-			transCtx.Logger.V(1).Info("rbac resources owner transferred, skip deletion",
-				"fromComp", comp.Name, "toComp", otherComp.Name, "name", klog.KObj(v).String(), "gvk", gvk)
-			return true, nil
-		}
-		return false, nil
-	default:
-		return false, nil
+	graphCli.Update(dag, nil, obj)
+	gvk, err := apiutil.GVKForObject(obj, model.GetScheme())
+	if err != nil {
+		return err
 	}
+	transCtx.Logger.V(1).Info("rbac resources orphaned",
+		"component", comp.Name, "name", klog.KObj(obj).String(), "gvk", gvk)
+	return nil
 }
 
 func notifyDependents4CompDeletion(transCtx *componentTransformContext, dag *graph.DAG) error {
diff --git a/controllers/apps/component/transformer_component_rbac.go b/controllers/apps/component/transformer_component_rbac.go
index 1478e6fed..47771d9dc 100644
--- a/controllers/apps/component/transformer_component_rbac.go
+++ b/controllers/apps/component/transformer_component_rbac.go
@@ -39,6 +39,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/factory"
 	"github.com/apecloud/kubeblocks/pkg/controller/graph"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	"github.com/apecloud/kubeblocks/pkg/generics"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
@@ -147,7 +148,8 @@ func createOrUpdate[T any, PT generics.PObject[T]](transCtx *componentTransformC
 		}
 		return nil, err
 	}
-	if !cmpFn(oldObj, obj) {
+	// adopt any orphaned object
+	if !cmpFn(oldObj, obj) || metav1.GetControllerOf(oldObj) == nil {
 		transCtx.Logger.V(1).Info("updating rbac resources",
 			"name", klog.KObj(obj).String(), "obj", fmt.Sprintf("%#v", obj))
 		graphCli.Update(dag, oldObj, obj, appsutil.InDataContext4G())
@@ -160,7 +162,7 @@ func createOrUpdateServiceAccount(transCtx *componentTransformContext,
 	synthesizedComp := transCtx.SynthesizeComponent
 
 	sa := factory.BuildServiceAccount(synthesizedComp, serviceAccountName)
-	if err := setCompOwnershipNFinalizer(transCtx.Component, sa); err != nil {
+	if err := intctrlutil.SetOwnership(transCtx.Component, sa, model.GetScheme(), ""); err != nil {
 		return nil, err
 	}
 
@@ -177,7 +179,7 @@ func createOrUpdateRole(transCtx *componentTransformContext, graphCli model.Grap
 	if role == nil {
 		return nil, nil
 	}
-	if err := setCompOwnershipNFinalizer(transCtx.Component, role); err != nil {
+	if err := intctrlutil.SetOwnership(transCtx.Component, role, model.GetScheme(), ""); err != nil {
 		return nil, err
 	}
 	return createOrUpdate(transCtx, role, graphCli, dag, func(old, new *rbacv1.Role) bool {
@@ -201,7 +203,7 @@ func createOrUpdateRoleBinding(transCtx *componentTransformContext,
 			Kind:     "Role",
 			Name:     cmpdRole.Name,
 		}, serviceAccountName)
-		if err := setCompOwnershipNFinalizer(transCtx.Component, cmpdRoleBinding); err != nil {
+		if err := intctrlutil.SetOwnership(transCtx.Component, cmpdRoleBinding, model.GetScheme(), ""); err != nil {
 			return nil, err
 		}
 		rb, err := createOrUpdate(transCtx, cmpdRoleBinding, graphCli, dag, cmpRoleBinding)
@@ -222,7 +224,7 @@ func createOrUpdateRoleBinding(transCtx *componentTransformContext,
 			},
 			serviceAccountName,
 		)
-		if err := setCompOwnershipNFinalizer(transCtx.Component, clusterPodRoleBinding); err != nil {
+		if err := intctrlutil.SetOwnership(transCtx.Component, clusterPodRoleBinding, model.GetScheme(), ""); err != nil {
 			return nil, err
 		}
 		rb, err := createOrUpdate(transCtx, clusterPodRoleBinding, graphCli, dag, cmpRoleBinding)
diff --git a/controllers/dataprotection/backup_controller.go b/controllers/dataprotection/backup_controller.go
index 3bebf142e..83e1be67e 100644
--- a/controllers/dataprotection/backup_controller.go
+++ b/controllers/dataprotection/backup_controller.go
@@ -536,7 +536,7 @@ func (r *BackupReconciler) patchBackupStatus(
 		request.Status.BaseBackupName = request.BaseBackup.Name
 	}
 
-	if err = dpbackup.SetExpirationByCreationTime(request.Backup); err != nil {
+	if err = dpbackup.SetExpirationTime(request.Backup); err != nil {
 		return err
 	}
 	return r.Client.Status().Patch(request.Ctx, request.Backup, client.MergeFrom(original))
@@ -598,10 +598,6 @@ func (r *BackupReconciler) handleRunningPhase(
 					existFailedAction = true
 					break actions
 				case dpv1alpha1.ActionPhaseRunning:
-					// update status
-					if err = r.Client.Status().Patch(reqCtx.Ctx, request.Backup, client.MergeFrom(backup)); err != nil {
-						return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
-					}
 					waiting = true
 					break actions
 				}
@@ -609,6 +605,17 @@ func (r *BackupReconciler) handleRunningPhase(
 		}
 	}
 	if waiting {
+		// reset time related fields for continuous backup
+		request.Status.CompletionTimestamp = nil
+		request.Status.Duration = nil
+		err = dpbackup.SetExpirationTime(request.Backup)
+		if err != nil {
+			return r.updateStatusIfFailed(reqCtx, backup, request.Backup, fmt.Errorf("failed to set expiration time, %v", err))
+		}
+		// update status
+		if err = r.Client.Status().Patch(reqCtx.Ctx, request.Backup, client.MergeFrom(backup)); err != nil {
+			return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
+		}
 		return intctrlutil.Reconciled()
 	}
 	if existFailedAction {
@@ -623,17 +630,9 @@ func (r *BackupReconciler) handleRunningPhase(
 		duration := request.Status.CompletionTimestamp.Sub(request.Status.StartTimestamp.Time).Round(time.Second)
 		request.Status.Duration = &metav1.Duration{Duration: duration}
 	}
-	if request.Spec.RetentionPeriod != "" {
-		// set expiration time
-		duration, err := request.Spec.RetentionPeriod.ToDuration()
-		if err != nil {
-			return r.updateStatusIfFailed(reqCtx, backup, request.Backup, fmt.Errorf("failed to parse retention period %s, %v", request.Spec.RetentionPeriod, err))
-		}
-		if duration.Seconds() > 0 {
-			request.Status.Expiration = &metav1.Time{
-				Time: request.Status.CompletionTimestamp.Add(duration),
-			}
-		}
+	err = dpbackup.SetExpirationTime(request.Backup)
+	if err != nil {
+		return r.updateStatusIfFailed(reqCtx, backup, request.Backup, fmt.Errorf("failed to set expiration time, %v", err))
 	}
 	r.Recorder.Event(backup, corev1.EventTypeNormal, "CreatedBackup", "Completed backup")
 	if err = r.Client.Status().Patch(reqCtx.Ctx, request.Backup, client.MergeFrom(backup)); err != nil {
@@ -679,13 +678,13 @@ func (r *BackupReconciler) checkIsCompletedDuringRunning(reqCtx intctrlutil.Requ
 	patch := client.MergeFrom(backup.DeepCopy())
 	backup.Status.Phase = dpv1alpha1.BackupPhaseCompleted
 	backup.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now().UTC()}
-	_ = dpbackup.SetExpirationByCreationTime(backup)
+	// set expiration time
+	_ = dpbackup.SetExpirationTime(backup)
 	if !backup.Status.StartTimestamp.IsZero() {
 		// round the duration to a multiple of seconds.
 		duration := backup.Status.CompletionTimestamp.Sub(backup.Status.StartTimestamp.Time).Round(time.Second)
 		backup.Status.Duration = &metav1.Duration{Duration: duration}
 	}
-
 	for i := range backup.Status.Actions {
 		act := &backup.Status.Actions[i]
 		act.Phase = dpv1alpha1.ActionPhaseCompleted
@@ -722,7 +721,7 @@ func (r *BackupReconciler) updateStatusIfFailed(
 
 	// set expiration time for failed backup, make sure the failed backup will be
 	// deleted after the expiration time.
-	_ = dpbackup.SetExpirationByCreationTime(backup)
+	_ = dpbackup.SetExpirationTime(backup)
 
 	if errUpdate := r.Client.Status().Patch(reqCtx.Ctx, backup, client.MergeFrom(original)); errUpdate != nil {
 		return intctrlutil.CheckedRequeueWithError(errUpdate, reqCtx.Log, "")
diff --git a/controllers/dataprotection/storageprovider_controller.go b/controllers/dataprotection/storageprovider_controller.go
index 5836a6879..a6c09ae76 100644
--- a/controllers/dataprotection/storageprovider_controller.go
+++ b/controllers/dataprotection/storageprovider_controller.go
@@ -21,9 +21,11 @@ package dataprotection
 
 import (
 	"context"
+	"slices"
 	"sync"
 
 	storagev1 "k8s.io/api/storage/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -80,18 +82,23 @@ func (r *StorageProviderReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	// get provider object
 	provider := &dpv1alpha1.StorageProvider{}
 	if err := r.Get(ctx, req.NamespacedName, provider); err != nil {
+		if apierrors.IsNotFound(err) {
+			r.removeDependency(req.Name)
+			return intctrlutil.Reconciled()
+		}
 		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to get StorageProvider")
 	}
 
 	// add dependency to CSIDriver
 	r.ensureDependency(provider)
 
-	// handle finalizer
-	res, err := intctrlutil.HandleCRDeletion(reqCtx, r, provider, dptypes.DataProtectionFinalizerName, func() (*ctrl.Result, error) {
-		return nil, r.deleteExternalResources(reqCtx, provider)
-	})
-	if res != nil {
-		return *res, err
+	// We should not add a finalizer to the StorageProvider resource,
+	// because when KubeBlocks is uninstalled, the controller workload will be stopped first,
+	// so the finalizer will not be processed, preventing those StorageProvider resources
+	// from being deleted.
+	// Remove the finalizer if it's present.
+	if err := r.removeFinalizer(reqCtx, provider); err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "failed to remove finalizer")
 	}
 
 	// check CSI driver if specified
@@ -117,6 +124,19 @@ func (r *StorageProviderReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	return intctrlutil.Reconciled()
 }
 
+func (r *StorageProviderReconciler) removeFinalizer(reqCtx intctrlutil.RequestCtx,
+	provider *dpv1alpha1.StorageProvider) error {
+	pos := slices.Index(provider.Finalizers, dptypes.DataProtectionFinalizerName)
+	if pos < 0 {
+		return nil
+	}
+	provider.Finalizers = slices.Delete(provider.Finalizers, pos, pos+1)
+	if err := r.Update(reqCtx.Ctx, provider); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (r *StorageProviderReconciler) updateStatus(reqCtx intctrlutil.RequestCtx,
 	provider *dpv1alpha1.StorageProvider,
 	checkErr error) error {
@@ -195,28 +215,21 @@ func (r *StorageProviderReconciler) ensureDependency(provider *dpv1alpha1.Storag
 	r.driverDependencies[driverName] = append(list, provider.Name)
 }
 
-func (r *StorageProviderReconciler) removeDependency(provider *dpv1alpha1.StorageProvider) {
-	if provider.Spec.CSIDriverName == "" {
-		return
-	}
+func (r *StorageProviderReconciler) removeDependency(providerName string) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	list := r.driverDependencies[provider.Spec.CSIDriverName]
-	for i, x := range list {
-		if x == provider.Name {
-			list[i] = list[len(list)-1]
-			r.driverDependencies[provider.Spec.CSIDriverName] = list[:len(list)-1]
-			return
+	for driverName := range r.driverDependencies {
+		list := r.driverDependencies[driverName]
+		for i, x := range list {
+			if x == providerName {
+				list[i] = list[len(list)-1]
+				r.driverDependencies[driverName] = list[:len(list)-1]
+				break
+			}
 		}
 	}
 }
 
-func (r *StorageProviderReconciler) deleteExternalResources(
-	reqCtx intctrlutil.RequestCtx, provider *dpv1alpha1.StorageProvider) error {
-	r.removeDependency(provider)
-	return nil
-}
-
 // SetupWithManager sets up the controller with the Manager.
 func (r *StorageProviderReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	b := intctrlutil.NewControllerManagedBy(mgr).
diff --git a/controllers/parameters/componentdrivenparameter_controller.go b/controllers/parameters/componentdrivenparameter_controller.go
index 02001dccc..b956a2f05 100644
--- a/controllers/parameters/componentdrivenparameter_controller.go
+++ b/controllers/parameters/componentdrivenparameter_controller.go
@@ -21,9 +21,12 @@ package parameters
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
+	"slices"
 
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -41,6 +44,7 @@ import (
 	configctrl "github.com/apecloud/kubeblocks/pkg/controller/configuration"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
+	"github.com/apecloud/kubeblocks/pkg/generics"
 )
 
 // ComponentDrivenParameterReconciler reconciles a Parameter object
@@ -192,6 +196,14 @@ func buildComponentParameter(reqCtx intctrlutil.RequestCtx, reader client.Reader
 	if err != nil {
 		return nil, err
 	}
+	initParameters, err := resolveInitParameters(reqCtx, reader, comp)
+	if err != nil {
+		return nil, err
+	}
+	parameterSpecs := configctrl.ClassifyParamsFromConfigTemplate(initParameters, cmpd, paramsDefs, tpls)
+	if err = handleCustomParameterTemplate(reqCtx.Ctx, reader, comp.Spec.Annotations, parameterSpecs); err != nil {
+		return nil, err
+	}
 
 	clusterName, _ := component.GetClusterName(comp)
 	componentName, _ := component.ShortName(clusterName, comp.Name)
@@ -200,17 +212,118 @@ func buildComponentParameter(reqCtx intctrlutil.RequestCtx, reader client.Reader
 		AddLabelsInMap(constant.GetCompLabelsWithDef(clusterName, componentName, cmpd.Name)).
 		ClusterRef(clusterName).
 		Component(componentName).
-		SetConfigurationItem(configctrl.ClassifyParamsFromConfigTemplate(nil, cmpd, paramsDefs, tpls)).
+		SetConfigurationItem(parameterSpecs).
 		GetObject()
 	if err = intctrlutil.SetOwnerReference(comp, parameterObj); err != nil {
 		return nil, err
 	}
+	sharding, err := configctrl.ResolveShardingReference(reqCtx.Ctx, reader, comp)
+	if err != nil {
+		return nil, err
+	}
 	if configRender != nil {
-		err = configctrl.UpdateConfigPayload(&parameterObj.Spec, &comp.Spec, &configRender.Spec)
+		err = configctrl.UpdateConfigPayload(&parameterObj.Spec, &comp.Spec, &configRender.Spec, sharding)
 	}
 	return parameterObj, err
 }
 
+func handleCustomParameterTemplate(ctx context.Context, reader client.Reader, annotations map[string]string, specs []parametersv1alpha1.ConfigTemplateItemDetail) error {
+	if len(annotations) == 0 {
+		return nil
+	}
+	customParamsTpl := annotations[constant.CustomParameterTemplateAnnotationKey]
+	if customParamsTpl == "" {
+		return nil
+	}
+
+	var customTemplates map[string]parametersv1alpha1.ConfigTemplateExtension
+	if err := json.Unmarshal([]byte(customParamsTpl), &customTemplates); err != nil {
+		return errors.Wrap(err, "failed to unmarshal custom parameter template")
+	}
+	if err := validateCustomTemplate(ctx, reader, customTemplates); err != nil {
+		return errors.Wrap(err, "failed to validate custom parameter template")
+	}
+
+	for tplName, tpl := range customTemplates {
+		match := func(spec parametersv1alpha1.ConfigTemplateItemDetail) bool {
+			return spec.Name == tplName
+		}
+		index := generics.FindFirstFunc(specs, match)
+		if index < 0 {
+			return fmt.Errorf("custom template[%s] not found in component definition", tplName)
+		}
+		specs[index].CustomTemplates = tpl.DeepCopy()
+	}
+	return nil
+}
+
+func resolveInitParameters(reqCtx intctrlutil.RequestCtx, reader client.Reader, comp *appsv1.Component) (parametersv1alpha1.ComponentParameters, error) {
+	resolveShardingName := func(comp *appsv1.Component) string {
+		if len(comp.Labels) == 0 {
+			return ""
+		}
+		return comp.Labels[constant.KBAppShardingNameLabelKey]
+	}
+	componentMatcher := func(clusterName, compName string) func(parametersv1alpha1.Parameter) bool {
+		return func(pcr parametersv1alpha1.Parameter) bool {
+			if model.IsObjectDeleting(&pcr) {
+				return false
+			}
+			if pcr.Spec.ClusterName != clusterName {
+				return false
+			}
+			for _, parameter := range pcr.Spec.ComponentParameters {
+				if parameter.ComponentName == compName {
+					return true
+				}
+			}
+			return false
+		}
+	}
+
+	parameters := &parametersv1alpha1.ParameterList{}
+	clusterName, _ := component.GetClusterName(comp)
+	listOptions := []client.ListOption{
+		client.MatchingLabels{
+			constant.AppInstanceLabelKey:    clusterName,
+			constant.ParametersInitLabelKey: "true",
+		},
+		client.InNamespace(comp.Namespace),
+	}
+	if err := reader.List(reqCtx.Ctx, parameters, listOptions...); err != nil {
+		return nil, errors.Wrapf(err, "failed to list init parameters: %v", client.ObjectKeyFromObject(comp))
+	}
+
+	compRealName, _ := component.ShortName(clusterName, comp.Name)
+	if shardingName := resolveShardingName(comp); shardingName != "" {
+		compRealName = shardingName
+	}
+
+	parameterCRs := generics.FindFunc(parameters.Items, componentMatcher(clusterName, compRealName))
+	if len(parameterCRs) == 0 {
+		return nil, nil
+	}
+	return resolveInitParametersFromParameterCR(parameterCRs, compRealName)
+}
+
+func resolveInitParametersFromParameterCR(parameterCRs []parametersv1alpha1.Parameter, compName string) (parametersv1alpha1.ComponentParameters, error) {
+	var initParams parametersv1alpha1.ComponentParameters
+
+	slices.SortStableFunc(parameterCRs, func(a, b parametersv1alpha1.Parameter) int {
+		return a.CreationTimestamp.Compare(b.CreationTimestamp.Time)
+	})
+	for _, pcr := range parameterCRs {
+		for _, compParams := range pcr.Spec.ComponentParameters {
+			if compParams.ComponentName == compName {
+				if err := mergeWithOverride(&initParams, compParams.Parameters); err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+	return initParams, nil
+}
+
 func resolveComponentTemplate(ctx context.Context, reader client.Reader, cmpd *appsv1.ComponentDefinition) (map[string]*corev1.ConfigMap, error) {
 	tpls := make(map[string]*corev1.ConfigMap, len(cmpd.Spec.Configs))
 	for _, config := range cmpd.Spec.Configs {
@@ -228,6 +341,9 @@ func (r *ComponentDrivenParameterReconciler) mergeComponentParameter(expected *p
 		if len(dest.ConfigFileParams) == 0 && len(expected.ConfigFileParams) != 0 {
 			dest.ConfigFileParams = expected.ConfigFileParams
 		}
+		if expected.CustomTemplates != nil {
+			dest.CustomTemplates = expected.CustomTemplates
+		}
 		dest.Payload = expected.Payload
 	})
 }
diff --git a/controllers/parameters/componentparameter_controller.go b/controllers/parameters/componentparameter_controller.go
index 1b58e007a..7b1adb2c4 100644
--- a/controllers/parameters/componentparameter_controller.go
+++ b/controllers/parameters/componentparameter_controller.go
@@ -39,6 +39,7 @@ import (
 
 	parametersv1alpha1 "github.com/apecloud/kubeblocks/apis/parameters/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
+	"github.com/apecloud/kubeblocks/pkg/controller/model"
 	"github.com/apecloud/kubeblocks/pkg/controller/multicluster"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
@@ -99,14 +100,15 @@ func (r *ComponentParameterReconciler) SetupWithManager(mgr ctrl.Manager, multiC
 func (r *ComponentParameterReconciler) reconcile(reqCtx intctrlutil.RequestCtx, componentParameter *parametersv1alpha1.ComponentParameter) (ctrl.Result, error) {
 	tasks := generateReconcileTasks(reqCtx, componentParameter)
 	if len(tasks) == 0 {
+		reqCtx.Log.Info("nothing to reconcile")
 		return intctrlutil.Reconciled()
 	}
 
 	fetcherTask, err := prepareReconcileTask(reqCtx, r.Client, componentParameter)
 	if err != nil {
-		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, errors.Wrap(err, "failed to get related object").Error())
+		return intctrlutil.RequeueWithError(err, reqCtx.Log, errors.Wrap(err, "failed to get related object").Error())
 	}
-	if !fetcherTask.ClusterObj.GetDeletionTimestamp().IsZero() {
+	if model.IsObjectDeleting(fetcherTask.ComponentObj) {
 		reqCtx.Log.Info("cluster is deleting, skip reconcile")
 		return intctrlutil.Reconciled()
 	}
diff --git a/controllers/parameters/parallel_upgrade_policy.go b/controllers/parameters/parallel_upgrade_policy.go
index f7049199d..2025849ce 100644
--- a/controllers/parameters/parallel_upgrade_policy.go
+++ b/controllers/parameters/parallel_upgrade_policy.go
@@ -49,7 +49,7 @@ func (p *restartContainerUpgradePolicy) GetPolicyName() string {
 }
 
 func (p *restartContainerUpgradePolicy) restartPods(rctx reconfigureContext, pods []corev1.Pod, funcs RollingUpgradeFuncs) (ReturnedStatus, error) {
-	var configKey = rctx.getConfigKey()
+	var configKey = rctx.generateConfigIdentifier()
 	var configVersion = rctx.getTargetVersionHash()
 
 	for _, pod := range pods {
diff --git a/controllers/parameters/parameter_controller.go b/controllers/parameters/parameter_controller.go
index c889aba26..649bfc508 100644
--- a/controllers/parameters/parameter_controller.go
+++ b/controllers/parameters/parameter_controller.go
@@ -29,8 +29,11 @@ import (
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	parametersv1alpha1 "github.com/apecloud/kubeblocks/apis/parameters/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/render"
@@ -79,6 +82,7 @@ func (r *ParameterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 func (r *ParameterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&parametersv1alpha1.Parameter{}).
+		Watches(&appsv1.Cluster{}, handler.EnqueueRequestsFromMapFunc(r.filterParametersResources)).
 		Complete(r)
 }
 
@@ -106,6 +110,13 @@ func (r *ParameterReconciler) handleComponent(rctx *ReconcileContext, updatedPar
 }
 
 func (r *ParameterReconciler) reconcile(reqCtx intctrlutil.RequestCtx, parameter *parametersv1alpha1.Parameter) (ctrl.Result, error) {
+	res, err := handleClusterDeleted(reqCtx, r.Client, parameter)
+	if res != nil {
+		return *res, err
+	}
+	if err != nil {
+		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
+	}
 	if intctrlutil.ParametersTerminalPhases(parameter.Status, parameter.Generation) {
 		return intctrlutil.Reconciled()
 	}
@@ -164,20 +175,47 @@ func (r *ParameterReconciler) failWithTerminalReconcile(reqCtx intctrlutil.Reque
 	return updateParameterStatus(reqCtx, r.Client, parameter, patch, true)
 }
 
+func (r *ParameterReconciler) failWithRetryReconcile(reqCtx intctrlutil.RequestCtx, parameter *parametersv1alpha1.Parameter, err error) (ctrl.Result, error) {
+	patch := parameter.DeepCopy()
+	parameter.Status.Phase = parametersv1alpha1.CRunningPhase
+	parameter.Status.Message = err.Error()
+	return updateParameterStatus(reqCtx, r.Client, parameter, patch, false)
+}
+
 func (r *ParameterReconciler) fail(reqCtx intctrlutil.RequestCtx, parameter *parametersv1alpha1.Parameter, err error) (ctrl.Result, error) {
 	if intctrlutil.IsTargetError(err, intctrlutil.ErrorTypeFatal) {
 		return r.failWithTerminalReconcile(reqCtx, parameter, err)
 	}
-	return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
+	return r.failWithRetryReconcile(reqCtx, parameter, err)
+}
+
+func (r *ParameterReconciler) filterParametersResources(ctx context.Context, object client.Object) []reconcile.Request {
+	listOpts := []client.ListOption{
+		client.InNamespace(object.GetNamespace()),
+		client.MatchingLabels{constant.AppInstanceLabelKey: object.GetName()},
+	}
+	parameters := &parametersv1alpha1.ParameterList{}
+	if err := r.Client.List(ctx, parameters, listOpts...); err != nil {
+		log.FromContext(ctx).Error(err, "failed to list parameters", "clusterKey", client.ObjectKeyFromObject(object))
+		return nil
+	}
+	requests := make([]reconcile.Request, 0, len(parameters.Items))
+	for _, parameter := range parameters.Items {
+		requests = append(requests, reconcile.Request{
+			NamespacedName: client.ObjectKeyFromObject(&parameter),
+		})
+	}
+	return requests
 }
 
-func validateCustomTemplate(ctx context.Context, cli client.Client, templates map[string]parametersv1alpha1.ConfigTemplateExtension) error {
+func validateCustomTemplate(ctx context.Context, cli client.Reader, templates map[string]parametersv1alpha1.ConfigTemplateExtension) error {
 	for configSpec, custom := range templates {
 		cm := &corev1.ConfigMap{}
 		err := cli.Get(ctx, types.NamespacedName{Name: custom.TemplateRef, Namespace: custom.Namespace}, cm)
 		if err != nil {
 			if apierrors.IsNotFound(err) {
-				return intctrlutil.NewErrorf(intctrlutil.ErrorTypeFatal, "not found configmap[%s] for custom template: %s", custom.TemplateRef, configSpec)
+				return intctrlutil.NewErrorf(intctrlutil.ErrorTypeFatal, "not found configmap[%s/%s] for custom template: %s",
+					custom.Namespace, custom.TemplateRef, configSpec)
 			}
 			return err
 		}
diff --git a/controllers/parameters/parameter_util.go b/controllers/parameters/parameter_util.go
index b48fd0bd9..859e8554a 100644
--- a/controllers/parameters/parameter_util.go
+++ b/controllers/parameters/parameter_util.go
@@ -25,11 +25,14 @@ import (
 
 	"github.com/imdario/mergo"
 	corev1 "k8s.io/api/core/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	parametersv1alpha1 "github.com/apecloud/kubeblocks/apis/parameters/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	configctrl "github.com/apecloud/kubeblocks/pkg/controller/configuration"
+	"github.com/apecloud/kubeblocks/pkg/controller/model"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -257,3 +260,45 @@ func syncComponentParameterStatus(rctx *ReconcileContext, parameter *parametersv
 	}
 	return nil
 }
+
+func handleClusterDeleted(reqCtx intctrlutil.RequestCtx, cli client.Client, parameter *parametersv1alpha1.Parameter) (*ctrl.Result, error) {
+	var cluster appsv1.Cluster
+
+	clusterKey := client.ObjectKey{
+		Namespace: parameter.GetNamespace(),
+		Name:      parameter.Spec.ClusterName,
+	}
+	if err := cli.Get(reqCtx.Ctx, clusterKey, &cluster); err != nil {
+		return intctrlutil.ResultToP(intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, ""))
+	}
+	if !cluster.IsDeleting() {
+		return updateOwnerReference(reqCtx, cli, &cluster, parameter)
+	}
+	reqCtx.Log.Info("cluster is deleting, delete parameter", "parameters", client.ObjectKeyFromObject(parameter))
+	if err := cli.Delete(reqCtx.Ctx, parameter); err != nil {
+		return intctrlutil.ResultToP(intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, ""))
+	}
+	return intctrlutil.ResultToP(intctrlutil.Reconciled())
+}
+
+func updateOwnerReference(reqCtx intctrlutil.RequestCtx, cli client.Client, cluster *appsv1.Cluster, parameter *parametersv1alpha1.Parameter) (*ctrl.Result, error) {
+	clusterName := parameter.Labels[constant.AppInstanceLabelKey]
+	if clusterName == parameter.Spec.ClusterName && model.IsOwnerOf(cluster, parameter) {
+		return nil, nil
+	}
+
+	patch := client.MergeFrom(parameter.DeepCopy())
+	if parameter.Labels == nil {
+		parameter.Labels = make(map[string]string)
+	}
+	if !model.IsOwnerOf(cluster, parameter) {
+		if err := intctrlutil.SetOwnerReference(cluster, parameter); err != nil {
+			return intctrlutil.ResultToP(intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, ""))
+		}
+	}
+	parameter.Labels[constant.AppInstanceLabelKey] = parameter.Spec.ClusterName
+	if err := cli.Patch(reqCtx.Ctx, parameter, patch); err != nil {
+		return intctrlutil.ResultToP(intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, ""))
+	}
+	return intctrlutil.ResultToP(intctrlutil.Reconciled())
+}
diff --git a/controllers/parameters/reconfigure_policy.go b/controllers/parameters/reconfigure_policy.go
index 6d6684ede..cf640f91f 100644
--- a/controllers/parameters/reconfigure_policy.go
+++ b/controllers/parameters/reconfigure_policy.go
@@ -21,6 +21,7 @@ package parameters
 
 import (
 	"math"
+	"strings"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -129,13 +130,13 @@ func GetClientFactory() createReconfigureClient {
 	return newGRPCClient
 }
 
-func (param *reconfigureContext) getConfigKey() string {
+func (param *reconfigureContext) generateConfigIdentifier() string {
 	key := param.ConfigTemplate.Name
 	if param.ConfigDescription != nil && param.ConfigDescription.Name != "" {
 		hash, _ := util.ComputeHash(param.ConfigDescription.Name)
-		key = key + "/" + hash
+		key = key + "-" + hash
 	}
-	return key
+	return strings.ReplaceAll(key, "_", "-")
 }
 
 func (param *reconfigureContext) getTargetVersionHash() string {
diff --git a/controllers/parameters/restart_policy.go b/controllers/parameters/restart_policy.go
index 05d9a46c0..30d4683a8 100644
--- a/controllers/parameters/restart_policy.go
+++ b/controllers/parameters/restart_policy.go
@@ -49,7 +49,7 @@ func (s *restartPolicy) GetPolicyName() string {
 func restartAndVerifyComponent(rctx reconfigureContext, funcs RollingUpgradeFuncs, objs []client.Object) (ReturnedStatus, error) {
 	var (
 		newVersion = rctx.getTargetVersionHash()
-		configKey  = rctx.getConfigKey()
+		configKey  = rctx.generateConfigIdentifier()
 
 		retStatus = ESRetry
 		progress  = core.NotStarted
diff --git a/controllers/parameters/rolling_upgrade_policy.go b/controllers/parameters/rolling_upgrade_policy.go
index 92acea8b2..70afaa8bf 100644
--- a/controllers/parameters/rolling_upgrade_policy.go
+++ b/controllers/parameters/rolling_upgrade_policy.go
@@ -69,7 +69,7 @@ func performRollingUpgrade(rctx reconfigureContext, funcs RollingUpgradeFuncs) (
 
 	var (
 		rollingReplicas = rctx.maxRollingReplicas()
-		configKey       = rctx.getConfigKey()
+		configKey       = rctx.generateConfigIdentifier()
 		configVersion   = rctx.getTargetVersionHash()
 	)
 
diff --git a/controllers/parameters/sync_upgrade_policy.go b/controllers/parameters/sync_upgrade_policy.go
index 175e3319a..f71a708e5 100644
--- a/controllers/parameters/sync_upgrade_policy.go
+++ b/controllers/parameters/sync_upgrade_policy.go
@@ -81,7 +81,7 @@ func sync(rctx reconfigureContext, updatedParameters map[string]string, pods []c
 
 		err         error
 		ctx         = rctx.Ctx
-		configKey   = rctx.getConfigKey()
+		configKey   = rctx.generateConfigIdentifier()
 		versionHash = rctx.getTargetVersionHash()
 		selector    = intctrlutil.GetPodSelector(rctx.ParametersDef)
 		fileName    string
diff --git a/deploy/helm/crds/parameters.kubeblocks.io_paramconfigrenderers.yaml b/deploy/helm/crds/parameters.kubeblocks.io_paramconfigrenderers.yaml
index e7bd0fba9..b15c96830 100644
--- a/deploy/helm/crds/parameters.kubeblocks.io_paramconfigrenderers.yaml
+++ b/deploy/helm/crds/parameters.kubeblocks.io_paramconfigrenderers.yaml
@@ -143,6 +143,7 @@ spec:
                         - vscale
                         - hscale
                         - tls
+                        - shardingHVScale
                         type: string
                       type: array
                       x-kubernetes-list-type: set
diff --git a/deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml b/deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml
index a0bd7f345..5877f752e 100644
--- a/deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml
+++ b/deploy/helm/crds/workloads.kubeblocks.io_instancesets.yaml
@@ -20,17 +20,21 @@ spec:
   scope: Namespaced
   versions:
   - additionalPrinterColumns:
-    - description: leader instance name.
-      jsonPath: .status.membersStatus[?(@.role.isLeader==true)].podName
-      name: LEADER
-      type: string
     - description: ready replicas.
       jsonPath: .status.readyReplicas
       name: READY
       type: string
-    - description: total replicas.
-      jsonPath: .status.replicas
-      name: REPLICAS
+    - description: desired replicas.
+      jsonPath: .spec.replicas
+      name: DESIRED
+      type: string
+    - description: updated replicas.
+      jsonPath: .status.updatedReplicas
+      name: UP-TO-DATE
+      type: string
+    - description: available replicas, which are ready for at least minReadySeconds.
+      jsonPath: .status.availableReplicas
+      name: AVAILABLE
       type: string
     - jsonPath: .metadata.creationTimestamp
       name: AGE
diff --git a/docs/developer_docs/api-reference/cluster.md b/docs/developer_docs/api-reference/cluster.md
index 3e96fc62e..c769fd217 100644
--- a/docs/developer_docs/api-reference/cluster.md
+++ b/docs/developer_docs/api-reference/cluster.md
@@ -9247,24 +9247,6 @@ int32
 </tr>
 </tbody>
 </table>
-<h3 id="apps.kubeblocks.io/v1.RerenderResourceType">RerenderResourceType
-(<code>string</code> alias)</h3>
-<div>
-<p>RerenderResourceType defines the resource requirements for a component.</p>
-</div>
-<table>
-<thead>
-<tr>
-<th>Value</th>
-<th>Description</th>
-</tr>
-</thead>
-<tbody><tr><td><p>&#34;hscale&#34;</p></td>
-<td></td>
-</tr><tr><td><p>&#34;vscale&#34;</p></td>
-<td></td>
-</tr></tbody>
-</table>
 <h3 id="apps.kubeblocks.io/v1.RetryPolicy">RetryPolicy
 </h3>
 <p>
diff --git a/pkg/constant/config.go b/pkg/constant/config.go
index 350044628..8b0160ec0 100644
--- a/pkg/constant/config.go
+++ b/pkg/constant/config.go
@@ -47,6 +47,9 @@ const (
 	CMInsCurrentConfigurationHashLabelKey  = "config.kubeblocks.io/update-config-hash"
 	CMConfigurationConstraintsNameLabelKey = "config.kubeblocks.io/config-constraints-name"
 	CMConfigurationTemplateVersion         = "config.kubeblocks.io/config-template-version"
+
+	ParametersInitLabelKey               = "config.kubeblocks.io/init-parameters"
+	CustomParameterTemplateAnnotationKey = "config.kubeblocks.io/custom-template"
 )
 
 const (
diff --git a/pkg/constant/payload.go b/pkg/constant/payload.go
index 6e0e85fd8..195108a0e 100644
--- a/pkg/constant/payload.go
+++ b/pkg/constant/payload.go
@@ -21,7 +21,8 @@ package constant
 
 const (
 	TLSPayload               = "tls"
-	ComponentResourcePayload = "component-resource"
+	ComponentResourcePayload = "componentResource"
 	ReplicasPayload          = "replicas"
-	BinaryVersionPayload     = "binary-version"
+	BinaryVersionPayload     = "binaryVersion"
+	ShardingPayload          = "sharding"
 )
diff --git a/pkg/controller/configuration/config_utils.go b/pkg/controller/configuration/config_utils.go
index 2ba923e91..834792084 100644
--- a/pkg/controller/configuration/config_utils.go
+++ b/pkg/controller/configuration/config_utils.go
@@ -23,9 +23,11 @@ import (
 	"context"
 	"fmt"
 	"slices"
+	"strconv"
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -248,7 +250,7 @@ func findPortByPortName(container corev1.Container) (int32, bool) {
 }
 
 // UpdateConfigPayload updates the configuration payload
-func UpdateConfigPayload(config *parametersv1alpha1.ComponentParameterSpec, component *appsv1.ComponentSpec, configRender *parametersv1alpha1.ParamConfigRendererSpec) error {
+func UpdateConfigPayload(config *parametersv1alpha1.ComponentParameterSpec, component *appsv1.ComponentSpec, configRender *parametersv1alpha1.ParamConfigRendererSpec, sharding *appsv1.ClusterSharding) error {
 	if len(configRender.Configs) == 0 {
 		return nil
 	}
@@ -278,6 +280,12 @@ func UpdateConfigPayload(config *parametersv1alpha1.ComponentParameterSpec, comp
 				return err
 			}
 		}
+		// check sharding h-scale operation
+		if sharding != nil && enableShardingHVScaleTrigger(configDescs) {
+			if _, err := intctrlutil.CheckAndPatchPayload(configSpec, constant.ShardingPayload, resolveShardingResource(sharding)); err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }
@@ -291,6 +299,13 @@ func rerenderConfigEnabled(configDescs []parametersv1alpha1.ComponentConfigDescr
 	return false
 }
 
+func resolveShardingResource(sharding *appsv1.ClusterSharding) map[string]string {
+	return map[string]string{
+		"shards":   strconv.Itoa(int(sharding.Shards)),
+		"replicas": strconv.Itoa(int(sharding.Template.Replicas)),
+	}
+}
+
 func enableHScaleTrigger(configDescs []parametersv1alpha1.ComponentConfigDescription) bool {
 	return rerenderConfigEnabled(configDescs, parametersv1alpha1.ComponentHScaleType)
 }
@@ -303,6 +318,10 @@ func enableTLSTrigger(configDescs []parametersv1alpha1.ComponentConfigDescriptio
 	return rerenderConfigEnabled(configDescs, parametersv1alpha1.ComponentTLSType)
 }
 
+func enableShardingHVScaleTrigger(configDescs []parametersv1alpha1.ComponentConfigDescription) bool {
+	return rerenderConfigEnabled(configDescs, parametersv1alpha1.ShardingComponentHVScaleType)
+}
+
 func ResolveComponentTemplate(ctx context.Context, reader client.Reader, cmpd *appsv1.ComponentDefinition) (map[string]*corev1.ConfigMap, error) {
 	tpls := make(map[string]*corev1.ConfigMap, len(cmpd.Spec.Configs))
 	for _, config := range cmpd.Spec.Configs {
@@ -314,3 +333,33 @@ func ResolveComponentTemplate(ctx context.Context, reader client.Reader, cmpd *a
 	}
 	return tpls, nil
 }
+
+func ResolveShardingReference(ctx context.Context, reader client.Reader, comp *appsv1.Component) (*appsv1.ClusterSharding, error) {
+	resolveShardingName := func(comp *appsv1.Component) string {
+		if len(comp.Labels) == 0 {
+			return ""
+		}
+		return comp.Labels[constant.KBAppShardingNameLabelKey]
+	}
+
+	shardingName := resolveShardingName(comp)
+	if shardingName == "" {
+		return nil, nil
+	}
+
+	cluster := appsv1.Cluster{}
+	clusterName, _ := component.GetClusterName(comp)
+	clusterKey := types.NamespacedName{
+		Name:      clusterName,
+		Namespace: comp.Namespace,
+	}
+	if err := reader.Get(ctx, clusterKey, &cluster); err != nil {
+		return nil, err
+	}
+	for i, sharding := range cluster.Spec.Shardings {
+		if sharding.Name == shardingName {
+			return &cluster.Spec.Shardings[i], nil
+		}
+	}
+	return nil, nil
+}
diff --git a/pkg/dataprotection/backup/utils.go b/pkg/dataprotection/backup/utils.go
index acc7c0bd3..1d4310af7 100644
--- a/pkg/dataprotection/backup/utils.go
+++ b/pkg/dataprotection/backup/utils.go
@@ -258,10 +258,28 @@ func GetSchedulePolicyByMethod(backupSchedule *dpv1alpha1.BackupSchedule, method
 	return nil
 }
 
-func SetExpirationByCreationTime(backup *dpv1alpha1.Backup) error {
-	// if expiration is already set, do not update it.
-	if backup.Status.Expiration != nil {
-		return nil
+func SetExpirationTime(backup *dpv1alpha1.Backup) error {
+	resetExpiration := func() {
+		backup.Status.Expiration = nil
+	}
+
+	phase := backup.Status.Phase
+	backupType := backup.Labels[types.BackupTypeLabelKey]
+	timePoint := backup.GetStartTime()
+	if phase == dpv1alpha1.BackupPhaseCompleted {
+		// if backup is completed, set expiration time based on completion time.
+		timePoint = backup.Status.CompletionTimestamp
+	} else {
+		if backupType == string(dpv1alpha1.BackupTypeContinuous) {
+			if phase == dpv1alpha1.BackupPhaseRunning {
+				// do not set expiration for running continuous backup.
+				resetExpiration()
+				return nil
+			}
+		} else if backup.Status.Expiration != nil {
+			// if expiration is already set, do not update it except for continuous backup.
+			return nil
+		}
 	}
 
 	duration, err := backup.Spec.RetentionPeriod.ToDuration()
@@ -272,20 +290,16 @@ func SetExpirationByCreationTime(backup *dpv1alpha1.Backup) error {
 	// if duration is zero, the backup will be kept forever.
 	// Do not set expiration time for it.
 	if duration.Seconds() == 0 {
+		resetExpiration()
 		return nil
 	}
 
-	var expiration *metav1.Time
-	if backup.Status.StartTimestamp != nil {
-		expiration = &metav1.Time{
-			Time: backup.Status.StartTimestamp.Add(duration),
-		}
-	} else {
-		expiration = &metav1.Time{
-			Time: backup.CreationTimestamp.Add(duration),
-		}
+	if timePoint == nil {
+		timePoint = &backup.CreationTimestamp
+	}
+	backup.Status.Expiration = &metav1.Time{
+		Time: timePoint.Add(duration),
 	}
-	backup.Status.Expiration = expiration
 	return nil
 }
 



Test patch:
diff --git a/controllers/apps/component/component_controller_test.go b/controllers/apps/component/component_controller_test.go
index e2a089c96..987976244 100644
--- a/controllers/apps/component/component_controller_test.go
+++ b/controllers/apps/component/component_controller_test.go
@@ -1152,6 +1152,31 @@ var _ = Describe("Component Controller", func() {
 		Eventually(testapps.CheckObjExists(&testCtx, rbKey, &rbacv1.RoleBinding{}, expectExisted)).Should(Succeed())
 	}
 
+	checkRBACResourceOrphaned := func(saName, rbName string, orphaned bool) {
+		saKey := types.NamespacedName{
+			Namespace: compObj.Namespace,
+			Name:      saName,
+		}
+		rbKey := types.NamespacedName{
+			Namespace: compObj.Namespace,
+			Name:      rbName,
+		}
+		Eventually(testapps.CheckObj(&testCtx, saKey, func(g Gomega, sa *corev1.ServiceAccount) {
+			if orphaned {
+				g.Expect(metav1.GetControllerOf(sa)).Should(BeNil())
+			} else {
+				g.Expect(metav1.GetControllerOf(sa)).ShouldNot(BeNil())
+			}
+		})).Should(Succeed())
+		Eventually(testapps.CheckObj(&testCtx, rbKey, func(g Gomega, rb *rbacv1.RoleBinding) {
+			if orphaned {
+				g.Expect(metav1.GetControllerOf(rb)).Should(BeNil())
+			} else {
+				g.Expect(metav1.GetControllerOf(rb)).ShouldNot(BeNil())
+			}
+		})).Should(Succeed())
+	}
+
 	testCompRBAC := func(compName, compDefName, saName string) {
 		By("creating a component with target service account name")
 		if len(saName) == 0 {
@@ -1182,9 +1207,9 @@ var _ = Describe("Component Controller", func() {
 		testapps.DeleteObject(&testCtx, compKey, &kbappsv1.Component{})
 		Eventually(testapps.CheckObjExists(&testCtx, compKey, &kbappsv1.Component{}, false)).Should(Succeed())
 
-		By("check the RBAC resources deleted")
+		By("check the RBAC resources orphaned")
 		saName := constant.GenerateDefaultServiceAccountName(compDefName)
-		checkRBACResourcesExistence(saName, fmt.Sprintf("%v-pod", saName), false)
+		checkRBACResourceOrphaned(saName, fmt.Sprintf("%v-pod", saName), true)
 	}
 
 	testRecreateCompWithRBACCreateByKubeBlocks := func(compName, compDefName string) {
@@ -1196,10 +1221,11 @@ var _ = Describe("Component Controller", func() {
 
 		By("check the RBAC resources deleted")
 		saName := constant.GenerateDefaultServiceAccountName(compDefName)
-		checkRBACResourcesExistence(saName, fmt.Sprintf("%v-pod", saName), false)
+		checkRBACResourceOrphaned(saName, fmt.Sprintf("%v-pod", saName), true)
 
 		By("re-create component with same name")
 		testCompRBAC(compName, compDefName, "")
+		checkRBACResourceOrphaned(saName, fmt.Sprintf("%v-pod", saName), false)
 	}
 
 	testSharedRBACResourceDeletion := func(compNamePrefix, compDefName string) {
@@ -1236,8 +1262,17 @@ var _ = Describe("Component Controller", func() {
 		By("delete first component")
 		testapps.DeleteObject(&testCtx, comp1Key, &kbappsv1.Component{})
 		Eventually(testapps.CheckObjExists(&testCtx, comp1Key, &kbappsv1.Component{}, false)).Should(Succeed())
+		checkRBACResourceOrphaned(saName, fmt.Sprintf("%v-pod", saName), true)
 
-		By("check rbac resources owner transferred")
+		By("trigger reconcile")
+		Expect(testapps.GetAndChangeObj(&testCtx, comp2Key, func(comp *kbappsv1.Component) {
+			if comp.Annotations == nil {
+				comp.Annotations = map[string]string{}
+			}
+			comp.Annotations["reconcile"] = time.Now().String()
+		})()).Should(Succeed())
+		By("check rbac resources adopted")
+		checkRBACResourceOrphaned(saName, fmt.Sprintf("%v-pod", saName), false)
 		Eventually(testapps.CheckObj(&testCtx, saKey, func(g Gomega, sa *corev1.ServiceAccount) {
 			refs := sa.GetOwnerReferences()
 			g.Expect(refs).Should(HaveLen(1))
@@ -1883,7 +1918,7 @@ var _ = Describe("Component Controller", func() {
 			testRecreateCompWithRBACCreateByKubeBlocks(defaultCompName, compDefObj.Name)
 		})
 
-		It("transfers rbac resources' ownership when multiple components share them", func() {
+		It("adopts an orphaned rbac resource", func() {
 			testSharedRBACResourceDeletion(defaultCompName, compDefObj.Name)
 		})
 
diff --git a/controllers/dataprotection/storageprovider_controller_test.go b/controllers/dataprotection/storageprovider_controller_test.go
index 25cb579a1..3eead14ba 100644
--- a/controllers/dataprotection/storageprovider_controller_test.go
+++ b/controllers/dataprotection/storageprovider_controller_test.go
@@ -33,7 +33,6 @@ import (
 
 	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
-	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/generics"
 	testapps "github.com/apecloud/kubeblocks/pkg/testutil/apps"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
@@ -195,12 +194,6 @@ var _ = Describe("StorageProvider controller", func() {
 			By("creating a StorageProvider with csi3")
 			createStorageProviderSpec("csi3")
 
-			By("checking StorageProvider object")
-			Eventually(testapps.CheckObj(&testCtx, key, func(g Gomega, provider *dpv1alpha1.StorageProvider) {
-				reconcileNoError(g)
-				g.Expect(provider.GetFinalizers()).To(ContainElement(dptypes.DataProtectionFinalizerName))
-			})).Should(Succeed())
-
 			By("deleting StorageProvider object")
 			Eventually(func(g Gomega) {
 				reconcileNoError(g)
@@ -210,7 +203,7 @@ var _ = Describe("StorageProvider controller", func() {
 					return
 				}
 				g.Expect(err).ToNot(HaveOccurred())
-				Expect(testCtx.Cli.Delete(testCtx.Ctx, provider)).ToNot(HaveOccurred())
+				g.Expect(testCtx.Cli.Delete(testCtx.Ctx, provider)).ToNot(HaveOccurred())
 			}).Should(Succeed())
 		})
 	})
diff --git a/controllers/parameters/componentdrivenparameter_controller_test.go b/controllers/parameters/componentdrivenparameter_controller_test.go
index 65529be7a..48c542d1a 100644
--- a/controllers/parameters/componentdrivenparameter_controller_test.go
+++ b/controllers/parameters/componentdrivenparameter_controller_test.go
@@ -20,12 +20,15 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package parameters
 
 import (
+	"encoding/json"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
@@ -82,6 +85,31 @@ var _ = Describe("ComponentParameterGenerator Controller", func() {
 			obj.Status.Phase = parametersv1alpha1.PDAvailablePhase
 		})()).Should(Succeed())
 
+		By("Create init parameters")
+		key := testapps.GetRandomizedKey(testCtx.DefaultNamespace, defaultCompName)
+		testparameters.NewParameterFactory(key.Name, key.Namespace, clusterName, defaultCompName).
+			AddParameters("innodb-buffer-pool-size", "1024M").
+			AddParameters("max_connections", "100").
+			AddLabels(constant.AppInstanceLabelKey, clusterName).
+			AddLabels(constant.ParametersInitLabelKey, "true").
+			Create(&testCtx)
+
+		By("Create a custom template cm")
+		tplKey := testapps.GetRandomizedKey(testCtx.DefaultNamespace, "custom-tpl")
+		tpl := testparameters.NewComponentTemplateFactory(tplKey.Name, testCtx.DefaultNamespace).
+			AddConfigFile(testparameters.MysqlConfigFile, "abcde=1234").
+			Create(&testCtx).
+			GetObject()
+
+		customTemplate := parametersv1alpha1.ConfigTemplateExtension{
+			TemplateRef: tpl.Name,
+			Namespace:   tpl.Namespace,
+			Policy:      parametersv1alpha1.ReplacePolicy,
+		}
+		annotationValue, _ := json.Marshal(map[string]parametersv1alpha1.ConfigTemplateExtension{
+			configSpecName: customTemplate,
+		})
+
 		By("Create a component obj")
 		fullCompName := constant.GenerateClusterComponentName(clusterName, defaultCompName)
 		compObj := testapps.NewComponentFactory(testCtx.DefaultNamespace, fullCompName, compDefObj.Name).
@@ -89,9 +117,9 @@ var _ = Describe("ComponentParameterGenerator Controller", func() {
 			SetUID(types.UID("test-uid")).
 			SetReplicas(1).
 			SetResources(corev1.ResourceRequirements{Limits: corev1.ResourceList{"memory": resource.MustParse("2Gi")}}).
+			SetAnnotations(map[string]string{constant.CustomParameterTemplateAnnotationKey: string(annotationValue)}).
 			Create(&testCtx).
 			GetObject()
-
 		return compObj
 	}
 
@@ -108,6 +136,9 @@ var _ = Describe("ComponentParameterGenerator Controller", func() {
 				g.Expect(item).ShouldNot(BeNil())
 				g.Expect(item.Payload).Should(HaveKey(constant.ReplicasPayload))
 				g.Expect(item.Payload).Should(HaveKey(constant.ComponentResourcePayload))
+				g.Expect(item.ConfigFileParams).Should(HaveKey(testparameters.MysqlConfigFile))
+				g.Expect(item.ConfigFileParams[testparameters.MysqlConfigFile].Parameters).Should(HaveKeyWithValue("innodb-buffer-pool-size", pointer.String("1024M")))
+				g.Expect(item.ConfigFileParams[testparameters.MysqlConfigFile].Parameters).Should(HaveKeyWithValue("max_connections", pointer.String("100")))
 			})).Should(Succeed())
 		})
 	})
diff --git a/controllers/parameters/restart_policy_test.go b/controllers/parameters/restart_policy_test.go
index c7b4da424..af3c8b699 100644
--- a/controllers/parameters/restart_policy_test.go
+++ b/controllers/parameters/restart_policy_test.go
@@ -76,12 +76,12 @@ var _ = Describe("Reconfigure restartPolicy", func() {
 					fromPodObjectList(newMockPodsWithInstanceSet(&mockParam.InstanceSetUnits[0], 2, withReadyPod(0, 2), func(pod *corev1.Pod, index int) {
 						// mock pod-1 restart
 						if index == 1 {
-							updatePodCfgVersion(pod, mockParam.getConfigKey(), mockParam.getTargetVersionHash())
+							updatePodCfgVersion(pod, mockParam.generateConfigIdentifier(), mockParam.getTargetVersionHash())
 						}
 					})),
 					fromPodObjectList(newMockPodsWithInstanceSet(&mockParam.InstanceSetUnits[0], 2, withReadyPod(0, 2), func(pod *corev1.Pod, index int) {
 						// mock all pod restart
-						updatePodCfgVersion(pod, mockParam.getConfigKey(), mockParam.getTargetVersionHash())
+						updatePodCfgVersion(pod, mockParam.generateConfigIdentifier(), mockParam.getTargetVersionHash())
 					})),
 				}),
 				testutil.WithTimes(3),
@@ -129,7 +129,7 @@ var _ = Describe("Reconfigure restartPolicy", func() {
 					fromPodObjectList(newMockPodsWithInstanceSet(&mockParam.InstanceSetUnits[0], 2)),
 					fromPodObjectList(newMockPodsWithInstanceSet(&mockParam.InstanceSetUnits[0], 2,
 						withReadyPod(0, 2), func(pod *corev1.Pod, _ int) {
-							updatePodCfgVersion(pod, mockParam.getConfigKey(), mockParam.getTargetVersionHash())
+							updatePodCfgVersion(pod, mockParam.generateConfigIdentifier(), mockParam.getTargetVersionHash())
 						})),
 				}),
 				testutil.WithAnyTimes(),
@@ -166,12 +166,12 @@ var _ = Describe("Reconfigure restartPolicy", func() {
 					fromPodObjectList(newMockPodsWithInstanceSet(&mockParam.InstanceSetUnits[0], 2, withReadyPod(0, 2), func(pod *corev1.Pod, index int) {
 						// mock pod-1 restart
 						if index == 1 {
-							updatePodCfgVersion(pod, mockParam.getConfigKey(), mockParam.getTargetVersionHash())
+							updatePodCfgVersion(pod, mockParam.generateConfigIdentifier(), mockParam.getTargetVersionHash())
 						}
 					})),
 					fromPodObjectList(newMockPodsWithInstanceSet(&mockParam.InstanceSetUnits[0], 2, withReadyPod(0, 2), func(pod *corev1.Pod, index int) {
 						// mock all pod restart
-						updatePodCfgVersion(pod, mockParam.getConfigKey(), mockParam.getTargetVersionHash())
+						updatePodCfgVersion(pod, mockParam.generateConfigIdentifier(), mockParam.getTargetVersionHash())
 					})),
 				}),
 				testutil.WithTimes(3),
diff --git a/controllers/parameters/rolling_upgrade_policy_test.go b/controllers/parameters/rolling_upgrade_policy_test.go
index 7ee441fce..2174ff06d 100644
--- a/controllers/parameters/rolling_upgrade_policy_test.go
+++ b/controllers/parameters/rolling_upgrade_policy_test.go
@@ -146,22 +146,22 @@ var _ = Describe("Reconfigure RollingPolicy", func() {
 			status, err = rollingPolicy.Upgrade(mockParam)
 			Expect(err).Should(Succeed())
 			Expect(status.Status).Should(BeEquivalentTo(ESRetry))
-			Expect(mockPods[acc][2].Labels[mockParam.getConfigKey()]).Should(BeEquivalentTo(mockParam.getTargetVersionHash()))
-			Expect(mockPods[acc][1].Labels[mockParam.getConfigKey()]).ShouldNot(BeEquivalentTo(mockParam.getTargetVersionHash()))
-			Expect(mockPods[acc][0].Labels[mockParam.getConfigKey()]).ShouldNot(BeEquivalentTo(mockParam.getTargetVersionHash()))
+			Expect(mockPods[acc][2].Labels[mockParam.generateConfigIdentifier()]).Should(BeEquivalentTo(mockParam.getTargetVersionHash()))
+			Expect(mockPods[acc][1].Labels[mockParam.generateConfigIdentifier()]).ShouldNot(BeEquivalentTo(mockParam.getTargetVersionHash()))
+			Expect(mockPods[acc][0].Labels[mockParam.generateConfigIdentifier()]).ShouldNot(BeEquivalentTo(mockParam.getTargetVersionHash()))
 
 			// upgrade pod-2
 			status, err = rollingPolicy.Upgrade(mockParam)
 			Expect(err).Should(Succeed())
 			Expect(status.Status).Should(BeEquivalentTo(ESRetry))
-			Expect(mockPods[acc][0].Labels[mockParam.getConfigKey()]).Should(BeEquivalentTo(mockParam.getTargetVersionHash()))
-			Expect(mockPods[acc][1].Labels[mockParam.getConfigKey()]).ShouldNot(BeEquivalentTo(mockParam.getTargetVersionHash()))
+			Expect(mockPods[acc][0].Labels[mockParam.generateConfigIdentifier()]).Should(BeEquivalentTo(mockParam.getTargetVersionHash()))
+			Expect(mockPods[acc][1].Labels[mockParam.generateConfigIdentifier()]).ShouldNot(BeEquivalentTo(mockParam.getTargetVersionHash()))
 
 			// upgrade pod-1
 			status, err = rollingPolicy.Upgrade(mockParam)
 			Expect(err).Should(Succeed())
 			Expect(status.Status).Should(BeEquivalentTo(ESRetry))
-			Expect(mockPods[acc][1].Labels[mockParam.getConfigKey()]).Should(BeEquivalentTo(mockParam.getTargetVersionHash()))
+			Expect(mockPods[acc][1].Labels[mockParam.generateConfigIdentifier()]).Should(BeEquivalentTo(mockParam.getTargetVersionHash()))
 
 			// finish check, not upgrade
 			status, err = rollingPolicy.Upgrade(mockParam)
diff --git a/controllers/parameters/suite_test.go b/controllers/parameters/suite_test.go
index 713baf854..51285e80b 100644
--- a/controllers/parameters/suite_test.go
+++ b/controllers/parameters/suite_test.go
@@ -90,13 +90,14 @@ var _ = BeforeSuite(func() {
 
 	err = parametersv1alpha1.AddToScheme(scheme.Scheme)
 	Expect(err).NotTo(HaveOccurred())
+	model.AddScheme(parametersv1alpha1.AddToScheme)
 	err = appsv1alpha1.AddToScheme(scheme.Scheme)
 	Expect(err).NotTo(HaveOccurred())
 	err = appsv1.AddToScheme(scheme.Scheme)
 	Expect(err).NotTo(HaveOccurred())
+	model.AddScheme(appsv1.AddToScheme)
 	err = workloadsv1.AddToScheme(scheme.Scheme)
 	Expect(err).NotTo(HaveOccurred())
-	model.AddScheme(parametersv1alpha1.AddToScheme)
 
 	// +kubebuilder:scaffold:scheme
 
diff --git a/pkg/dataprotection/backup/utils_test.go b/pkg/dataprotection/backup/utils_test.go
index 5f6fe25d2..c863c7af0 100644
--- a/pkg/dataprotection/backup/utils_test.go
+++ b/pkg/dataprotection/backup/utils_test.go
@@ -21,12 +21,16 @@ package backup
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/utils/pointer"
 
+	dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
+	dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
 
@@ -87,3 +91,142 @@ func TestBuildCronJobSchedule(t *testing.T) {
 		})
 	}
 }
+
+func TestSetExpirationTime(t *testing.T) {
+	now := metav1.Now()
+	tests := []struct {
+		name          string
+		backup        *dpv1alpha1.Backup
+		expectedError bool
+		verify        func(*testing.T, *dpv1alpha1.Backup)
+	}{
+		{
+			name: "already has expiration",
+			backup: &dpv1alpha1.Backup{
+				Status: dpv1alpha1.BackupStatus{
+					Expiration: &metav1.Time{Time: now.Add(24 * time.Hour)},
+				},
+			},
+			expectedError: false,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.NotNil(t, b.Status.Expiration)
+			},
+		},
+		{
+			name: "continuous backup type with running phase",
+			backup: &dpv1alpha1.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						dptypes.BackupTypeLabelKey: string(dpv1alpha1.BackupTypeContinuous),
+					},
+				},
+				Spec: dpv1alpha1.BackupSpec{
+					RetentionPeriod: "24h",
+				},
+				Status: dpv1alpha1.BackupStatus{
+					Phase: dpv1alpha1.BackupPhaseRunning,
+				},
+			},
+			expectedError: false,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.Nil(t, b.Status.Expiration)
+			},
+		},
+		{
+			name: "continuous backup type with failed phase",
+			backup: &dpv1alpha1.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						dptypes.BackupTypeLabelKey: string(dpv1alpha1.BackupTypeContinuous),
+					},
+				},
+				Spec: dpv1alpha1.BackupSpec{
+					RetentionPeriod: "24h",
+				},
+				Status: dpv1alpha1.BackupStatus{
+					Phase: dpv1alpha1.BackupPhaseFailed,
+					TimeRange: &dpv1alpha1.BackupTimeRange{
+						Start: &now,
+						End:   func() *metav1.Time { t := metav1.NewTime(now.Add(1 * time.Hour)); return &t }(),
+					},
+				},
+			},
+			expectedError: false,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.Equal(t, now.Add(24*time.Hour).UTC(), b.Status.Expiration.Time.UTC())
+			},
+		},
+		{
+			name: "continuous backup type with completed phase",
+			backup: &dpv1alpha1.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						dptypes.BackupTypeLabelKey: string(dpv1alpha1.BackupTypeContinuous),
+					},
+				},
+				Spec: dpv1alpha1.BackupSpec{
+					RetentionPeriod: "24h",
+				},
+				Status: dpv1alpha1.BackupStatus{
+					Phase:               dpv1alpha1.BackupPhaseCompleted,
+					CompletionTimestamp: &now,
+				},
+			},
+			expectedError: false,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.Equal(t, now.Add(24*time.Hour).UTC(), b.Status.Expiration.Time.UTC())
+			},
+		},
+		{
+			name: "with retention period and start timestamp",
+			backup: &dpv1alpha1.Backup{
+				Spec: dpv1alpha1.BackupSpec{
+					RetentionPeriod: "24h",
+				},
+				Status: dpv1alpha1.BackupStatus{
+					StartTimestamp: &now,
+				},
+			},
+			expectedError: false,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.Equal(t, now.Add(24*time.Hour).UTC(), b.Status.Expiration.Time.UTC())
+			},
+		},
+		{
+			name: "without retention period",
+			backup: &dpv1alpha1.Backup{
+				Status: dpv1alpha1.BackupStatus{
+					StartTimestamp: &now,
+				},
+			},
+			expectedError: false,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.Nil(t, b.Status.Expiration)
+			},
+		},
+		{
+			name: "invalid retention period",
+			backup: &dpv1alpha1.Backup{
+				Spec: dpv1alpha1.BackupSpec{
+					RetentionPeriod: "invalid",
+				},
+			},
+			expectedError: true,
+			verify: func(t *testing.T, b *dpv1alpha1.Backup) {
+				assert.Nil(t, b.Status.Expiration)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := SetExpirationTime(tt.backup)
+			if tt.expectedError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+			tt.verify(t, tt.backup)
+		})
+	}
+}
diff --git a/pkg/testutil/apps/component_factory.go b/pkg/testutil/apps/component_factory.go
index 333271403..7e7a10129 100644
--- a/pkg/testutil/apps/component_factory.go
+++ b/pkg/testutil/apps/component_factory.go
@@ -66,6 +66,11 @@ func (factory *MockComponentFactory) SetResources(resources corev1.ResourceRequi
 	return factory
 }
 
+func (factory *MockComponentFactory) SetAnnotations(annotations map[string]string) *MockComponentFactory {
+	factory.Get().Spec.Annotations = annotations
+	return factory
+}
+
 func (factory *MockComponentFactory) SetTLSConfig(enable bool, issuer *appsv1.Issuer) *MockComponentFactory {
 	if enable {
 		factory.Get().Spec.TLSConfig = &appsv1.TLSConfig{