diff --git a/api/v1/utils.go b/api/v1/utils.go
index e994efe7..5d7c7a76 100644
--- a/api/v1/utils.go
+++ b/api/v1/utils.go
@@ -7,9 +7,8 @@ import (
 	"regexp"
 	"strings"
 
-	"k8s.io/apimachinery/pkg/util/sets"
-
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/utils/ptr"
 
 	internalerrors "github.com/aerospike/aerospike-kubernetes-operator/errors"
diff --git a/api/v1beta1/aerospikebackup_webhook.go b/api/v1beta1/aerospikebackup_webhook.go
index acacddeb..25ed9f75 100644
--- a/api/v1beta1/aerospikebackup_webhook.go
+++ b/api/v1beta1/aerospikebackup_webhook.go
@@ -17,15 +17,11 @@ limitations under the License.
 package v1beta1
 
 import (
-	"context"
 	"fmt"
 	"reflect"
 	"strings"
 
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
-	utilRuntime "k8s.io/apimachinery/pkg/util/runtime"
-	clientGoScheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
@@ -33,9 +29,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 	"sigs.k8s.io/yaml"
 
-	"github.com/aerospike/aerospike-backup-service/pkg/model"
-	asdbv1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/validation"
 )
 
 func (r *AerospikeBackup) SetupWebhookWithManager(mgr ctrl.Manager) error {
@@ -65,12 +60,12 @@ func (r *AerospikeBackup) ValidateCreate() (admission.Warnings, error) {
 
 	abLog.Info("Validate create")
 
-	if len(r.Spec.OnDemandBackups) != 0 {
-		return nil, fmt.Errorf("onDemand backups config cannot be specified while creating backup")
+	if err := r.validate(); err != nil {
+		return nil, err
 	}
 
-	if err := r.validateBackupConfig(); err != nil {
-		return nil, err
+	if len(r.Spec.OnDemandBackups) != 0 {
+		return nil, fmt.Errorf("onDemand backups config cannot be specified while creating backup")
 	}
 
 	return nil, nil
@@ -88,7 +83,7 @@ func (r *AerospikeBackup) ValidateUpdate(old runtime.Object) (admission.Warnings
 		return nil, fmt.Errorf("backup service cannot be updated")
 	}
 
-	if err := r.validateBackupConfig(); err != nil {
+	if err := r.validate(); err != nil {
 		return nil, err
 	}
 
@@ -103,6 +98,22 @@ func (r *AerospikeBackup) ValidateUpdate(old runtime.Object) (admission.Warnings
 	return nil, nil
 }
 
+func (r *AerospikeBackup) validate() error {
+	k8sClient, gErr := getK8sClient()
+	if gErr != nil {
+		return gErr
+	}
+
+	if err := validateBackupSvcSupportedVersion(k8sClient,
+		r.Spec.BackupService.Name,
+		r.Spec.BackupService.Namespace,
+	); err != nil {
+		return err
+	}
+
+	return r.validateBackupConfig(k8sClient)
+}
+
 // ValidateDelete implements webhook.Validator so a webhook will be registered for the type
 func (r *AerospikeBackup) ValidateDelete() (admission.Warnings, error) {
 	abLog := logf.Log.WithName(namespacedName(r))
@@ -113,45 +124,32 @@ func (r *AerospikeBackup) ValidateDelete() (admission.Warnings, error) {
 	return nil, nil
 }
 
-func (r *AerospikeBackup) validateBackupConfig() error {
+func (r *AerospikeBackup) validateBackupConfig(k8sClient client.Client) error {
 	backupConfig := make(map[string]interface{})
 
 	if err := yaml.Unmarshal(r.Spec.Config.Raw, &backupConfig); err != nil {
 		return err
 	}
 
-	if _, ok := backupConfig[common.ServiceKey]; ok {
+	if _, ok := backupConfig[ServiceKey]; ok {
 		return fmt.Errorf("service field cannot be specified in backup config")
 	}
 
-	if _, ok := backupConfig[common.BackupPoliciesKey]; ok {
+	if _, ok := backupConfig[BackupPoliciesKey]; ok {
 		return fmt.Errorf("backup-policies field cannot be specified in backup config")
 	}
 
-	if _, ok := backupConfig[common.StorageKey]; ok {
+	if _, ok := backupConfig[StorageKey]; ok {
 		return fmt.Errorf("storage field cannot be specified in backup config")
 	}
 
-	if _, ok := backupConfig[common.SecretAgentsKey]; ok {
+	if _, ok := backupConfig[SecretAgentsKey]; ok {
 		return fmt.Errorf("secret-agent field cannot be specified in backup config")
 	}
 
-	var backupSvc AerospikeBackupService
-
-	cl, gErr := getK8sClient()
-	if gErr != nil {
-		return gErr
-	}
-
-	if err := cl.Get(context.TODO(),
-		types.NamespacedName{Name: r.Spec.BackupService.Name, Namespace: r.Spec.BackupService.Namespace},
-		&backupSvc); err != nil {
-		return err
-	}
-
-	var backupSvcConfig model.Config
-
-	if err := yaml.UnmarshalStrict(backupSvc.Spec.Config.Raw, &backupSvcConfig); err != nil {
+	backupSvcConfig, err := getBackupServiceFullConfig(k8sClient, r.Spec.BackupService.Name,
+		r.Spec.BackupService.Namespace)
+	if err != nil {
 		return err
 	}
 
@@ -165,7 +163,7 @@ func (r *AerospikeBackup) validateBackupConfig() error {
 		return err
 	}
 
-	err = updateValidateBackupSvcConfig(aeroClusters, backupRoutines, &backupSvcConfig)
+	err = updateValidateBackupSvcConfig(aeroClusters, backupRoutines, backupSvcConfig)
 	if err != nil {
 		return err
 	}
@@ -181,32 +179,13 @@ func (r *AerospikeBackup) validateBackupConfig() error {
 	return nil
 }
 
-func getK8sClient() (client.Client, error) {
-	restConfig := ctrl.GetConfigOrDie()
-
-	scheme := runtime.NewScheme()
-
-	utilRuntime.Must(asdbv1.AddToScheme(scheme))
-	utilRuntime.Must(clientGoScheme.AddToScheme(scheme))
-	utilRuntime.Must(AddToScheme(scheme))
-
-	cl, err := client.New(restConfig, client.Options{
-		Scheme: scheme,
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return cl, nil
-}
-
 func (r *AerospikeBackup) getValidatedAerospikeClusters(backupConfig map[string]interface{},
-) (map[string]*model.AerospikeCluster, error) {
-	if _, ok := backupConfig[common.AerospikeClusterKey]; !ok {
+) (map[string]*dto.AerospikeCluster, error) {
+	if _, ok := backupConfig[AerospikeClusterKey]; !ok {
 		return nil, fmt.Errorf("aerospike-cluster field is required field in backup config")
 	}
 
-	cluster, ok := backupConfig[common.AerospikeClusterKey].(map[string]interface{})
+	cluster, ok := backupConfig[AerospikeClusterKey].(map[string]interface{})
 	if !ok {
 		return nil, fmt.Errorf("aerospike-cluster field is not in the right format")
 	}
@@ -216,7 +195,7 @@ func (r *AerospikeBackup) getValidatedAerospikeClusters(backupConfig map[string]
 		return nil, cErr
 	}
 
-	aeroClusters := make(map[string]*model.AerospikeCluster)
+	aeroClusters := make(map[string]*dto.AerospikeCluster)
 
 	if err := yaml.UnmarshalStrict(clusterBytes, &aeroClusters); err != nil {
 		return nil, err
@@ -272,8 +251,8 @@ func (r *AerospikeBackup) validateAerospikeClusterUpdate(oldObj *AerospikeBackup
 		return err
 	}
 
-	oldCluster := oldObjConfig[common.AerospikeClusterKey].(map[string]interface{})
-	newCluster := currentConfig[common.AerospikeClusterKey].(map[string]interface{})
+	oldCluster := oldObjConfig[AerospikeClusterKey].(map[string]interface{})
+	newCluster := currentConfig[AerospikeClusterKey].(map[string]interface{})
 
 	for clusterName := range newCluster {
 		if _, ok := oldCluster[clusterName]; !ok {
@@ -286,13 +265,13 @@ func (r *AerospikeBackup) validateAerospikeClusterUpdate(oldObj *AerospikeBackup
 
 func (r *AerospikeBackup) getValidatedBackupRoutines(
 	backupConfig map[string]interface{},
-	aeroClusters map[string]*model.AerospikeCluster,
-) (map[string]*model.BackupRoutine, error) {
-	if _, ok := backupConfig[common.BackupRoutinesKey]; !ok {
+	aeroClusters map[string]*dto.AerospikeCluster,
+) (map[string]*dto.BackupRoutine, error) {
+	if _, ok := backupConfig[BackupRoutinesKey]; !ok {
 		return nil, fmt.Errorf("backup-routines field is required in backup config")
 	}
 
-	routines, ok := backupConfig[common.BackupRoutinesKey].(map[string]interface{})
+	routines, ok := backupConfig[BackupRoutinesKey].(map[string]interface{})
 	if !ok {
 		return nil, fmt.Errorf("backup-routines field is not in the right format")
 	}
@@ -302,7 +281,7 @@ func (r *AerospikeBackup) getValidatedBackupRoutines(
 		return nil, rErr
 	}
 
-	backupRoutines := make(map[string]*model.BackupRoutine)
+	backupRoutines := make(map[string]*dto.BackupRoutine)
 
 	if err := yaml.UnmarshalStrict(routineBytes, &backupRoutines); err != nil {
 		return nil, err
@@ -329,12 +308,12 @@ func (r *AerospikeBackup) getValidatedBackupRoutines(
 }
 
 func updateValidateBackupSvcConfig(
-	clusters map[string]*model.AerospikeCluster,
-	routines map[string]*model.BackupRoutine,
-	backupSvcConfig *model.Config,
+	clusters map[string]*dto.AerospikeCluster,
+	routines map[string]*dto.BackupRoutine,
+	backupSvcConfig *dto.Config,
 ) error {
 	if len(backupSvcConfig.AerospikeClusters) == 0 {
-		backupSvcConfig.AerospikeClusters = make(map[string]*model.AerospikeCluster)
+		backupSvcConfig.AerospikeClusters = make(map[string]*dto.AerospikeCluster)
 	}
 
 	for name, cluster := range clusters {
@@ -342,7 +321,7 @@ func updateValidateBackupSvcConfig(
 	}
 
 	if len(backupSvcConfig.BackupRoutines) == 0 {
-		backupSvcConfig.BackupRoutines = make(map[string]*model.BackupRoutine)
+		backupSvcConfig.BackupRoutines = make(map[string]*dto.BackupRoutine)
 	}
 
 	for name, routine := range routines {
@@ -350,19 +329,15 @@ func updateValidateBackupSvcConfig(
 	}
 
 	// Add empty placeholders for missing backupSvcConfig sections. This is required for validation to work.
-	if backupSvcConfig.ServiceConfig == nil {
-		backupSvcConfig.ServiceConfig = &model.BackupServiceConfig{}
-	}
-
 	if backupSvcConfig.ServiceConfig.HTTPServer == nil {
-		backupSvcConfig.ServiceConfig.HTTPServer = &model.HTTPServerConfig{}
+		backupSvcConfig.ServiceConfig.HTTPServer = &dto.HTTPServerConfig{}
 	}
 
 	if backupSvcConfig.ServiceConfig.Logger == nil {
-		backupSvcConfig.ServiceConfig.Logger = &model.LoggerConfig{}
+		backupSvcConfig.ServiceConfig.Logger = &dto.LoggerConfig{}
 	}
 
-	return backupSvcConfig.Validate()
+	return validation.ValidateConfiguration(backupSvcConfig)
 }
 
 func (r *AerospikeBackup) NamePrefix() string {
diff --git a/api/v1beta1/aerospikebackupservice_webhook.go b/api/v1beta1/aerospikebackupservice_webhook.go
index 18057d82..a48a0608 100644
--- a/api/v1beta1/aerospikebackupservice_webhook.go
+++ b/api/v1beta1/aerospikebackupservice_webhook.go
@@ -27,9 +27,12 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 	"sigs.k8s.io/yaml"
 
-	"github.com/aerospike/aerospike-backup-service/pkg/model"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/validation"
 )
 
+const minSupportedVersion = "3.0.0"
+
 func (r *AerospikeBackupService) SetupWebhookWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewWebhookManagedBy(mgr).
 		For(r).
@@ -57,15 +60,7 @@ func (r *AerospikeBackupService) ValidateCreate() (admission.Warnings, error) {
 
 	absLog.Info("Validate create")
 
-	if err := r.validateBackupServiceConfig(); err != nil {
-		return nil, err
-	}
-
-	if err := r.validateBackupServiceSecrets(); err != nil {
-		return nil, err
-	}
-
-	return nil, nil
+	return r.validate()
 }
 
 // ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
@@ -74,15 +69,7 @@ func (r *AerospikeBackupService) ValidateUpdate(_ runtime.Object) (admission.War
 
 	absLog.Info("Validate update")
 
-	if err := r.validateBackupServiceConfig(); err != nil {
-		return nil, err
-	}
-
-	if err := r.validateBackupServiceSecrets(); err != nil {
-		return nil, err
-	}
-
-	return nil, nil
+	return r.validate()
 }
 
 // ValidateDelete implements webhook.Validator so a webhook will be registered for the type
@@ -95,8 +82,24 @@ func (r *AerospikeBackupService) ValidateDelete() (admission.Warnings, error) {
 	return nil, nil
 }
 
+func (r *AerospikeBackupService) validate() (admission.Warnings, error) {
+	if err := ValidateBackupSvcVersion(r.Spec.Image); err != nil {
+		return nil, err
+	}
+
+	if err := r.validateBackupServiceConfig(); err != nil {
+		return nil, err
+	}
+
+	if err := r.validateBackupServiceSecrets(); err != nil {
+		return nil, err
+	}
+
+	return nil, nil
+}
+
 func (r *AerospikeBackupService) validateBackupServiceConfig() error {
-	var config model.Config
+	var config dto.Config
 
 	if err := yaml.UnmarshalStrict(r.Spec.Config.Raw, &config); err != nil {
 		return err
@@ -111,19 +114,15 @@ func (r *AerospikeBackupService) validateBackupServiceConfig() error {
 	}
 
 	// Add empty placeholders for missing config sections. This is required for validation to work.
-	if config.ServiceConfig == nil {
-		config.ServiceConfig = &model.BackupServiceConfig{}
-	}
-
 	if config.ServiceConfig.HTTPServer == nil {
-		config.ServiceConfig.HTTPServer = &model.HTTPServerConfig{}
+		config.ServiceConfig.HTTPServer = &dto.HTTPServerConfig{}
 	}
 
 	if config.ServiceConfig.Logger == nil {
-		config.ServiceConfig.Logger = &model.LoggerConfig{}
+		config.ServiceConfig.Logger = &dto.LoggerConfig{}
 	}
 
-	return config.Validate()
+	return validation.ValidateConfiguration(&config)
 }
 
 func (r *AerospikeBackupService) validateBackupServiceSecrets() error {
diff --git a/api/v1beta1/aerospikerestore_webhook.go b/api/v1beta1/aerospikerestore_webhook.go
index 8ac0fff6..e98c9947 100644
--- a/api/v1beta1/aerospikerestore_webhook.go
+++ b/api/v1beta1/aerospikerestore_webhook.go
@@ -23,13 +23,14 @@ import (
 
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 	"sigs.k8s.io/yaml"
 
-	"github.com/aerospike/aerospike-backup-service/pkg/model"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/validation"
 )
 
 const defaultPollingPeriod time.Duration = 60 * time.Second
@@ -67,7 +68,19 @@ func (r *AerospikeRestore) ValidateCreate() (admission.Warnings, error) {
 
 	arLog.Info("Validate create")
 
-	if err := r.validateRestoreConfig(); err != nil {
+	k8sClient, gErr := getK8sClient()
+	if gErr != nil {
+		return nil, gErr
+	}
+
+	if err := validateBackupSvcSupportedVersion(k8sClient,
+		r.Spec.BackupService.Name,
+		r.Spec.BackupService.Namespace,
+	); err != nil {
+		return nil, err
+	}
+
+	if err := r.validateRestoreConfig(k8sClient); err != nil {
 		return nil, err
 	}
 
@@ -99,22 +112,28 @@ func (r *AerospikeRestore) ValidateDelete() (admission.Warnings, error) {
 	return nil, nil
 }
 
-func (r *AerospikeRestore) validateRestoreConfig() error {
+func (r *AerospikeRestore) validateRestoreConfig(k8sClient client.Client) error {
 	restoreConfig := make(map[string]interface{})
 
 	if err := yaml.Unmarshal(r.Spec.Config.Raw, &restoreConfig); err != nil {
 		return err
 	}
 
+	backupSvcConfig, err := getBackupServiceFullConfig(k8sClient, r.Spec.BackupService.Name,
+		r.Spec.BackupService.Namespace)
+	if err != nil {
+		return err
+	}
+
 	switch r.Spec.Type {
 	case Full, Incremental:
-		var restoreRequest model.RestoreRequest
+		var restoreRequest dto.RestoreRequest
 
-		if _, ok := restoreConfig[common.RoutineKey]; ok {
+		if _, ok := restoreConfig[RoutineKey]; ok {
 			return fmt.Errorf("routine field is not allowed in restore config for restore type %s", r.Spec.Type)
 		}
 
-		if _, ok := restoreConfig[common.TimeKey]; ok {
+		if _, ok := restoreConfig[TimeKey]; ok {
 			return fmt.Errorf("time field is not allowed in restore config for restore type %s", r.Spec.Type)
 		}
 
@@ -122,12 +141,12 @@ func (r *AerospikeRestore) validateRestoreConfig() error {
 			return err
 		}
 
-		return restoreRequest.Validate()
+		return validation.ValidateRestoreRequest(&restoreRequest, backupSvcConfig)
 
 	case Timestamp:
-		var restoreRequest model.RestoreTimestampRequest
+		var restoreRequest dto.RestoreTimestampRequest
 
-		if _, ok := restoreConfig[common.SourceKey]; ok {
+		if _, ok := restoreConfig[SourceKey]; ok {
 			return fmt.Errorf("source field is not allowed in restore config for restore type %s", r.Spec.Type)
 		}
 
@@ -135,7 +154,7 @@ func (r *AerospikeRestore) validateRestoreConfig() error {
 			return err
 		}
 
-		return restoreRequest.Validate()
+		return validation.ValidateRestoreTimestampRequest(&restoreRequest, backupSvcConfig)
 
 	default:
 		// Code flow should not come here
diff --git a/internal/controller/common/constant.go b/api/v1beta1/constant.go
similarity index 51%
rename from internal/controller/common/constant.go
rename to api/v1beta1/constant.go
index e863bf3b..f1a6c2c3 100644
--- a/internal/controller/common/constant.go
+++ b/api/v1beta1/constant.go
@@ -1,4 +1,4 @@
-package common
+package v1beta1
 
 // Backup Config relate keys
 const (
@@ -8,19 +8,22 @@ const (
 	StorageKey              = "storage"
 	BackupRoutinesKey       = "backup-routines"
 	BackupPoliciesKey       = "backup-policies"
-	SecretAgentsKey         = "secret-agent"
+	SecretAgentsKey         = "secret-agents"
 	SourceClusterKey        = "source-cluster"
 	BackupServiceConfigYAML = "aerospike-backup-service.yml"
 )
 
 // Restore config fields
 const (
-	RoutineKey = "routine"
-	TimeKey    = "time"
-	SourceKey  = "source"
+	RoutineKey        = "routine"
+	TimeKey           = "time"
+	SourceKey         = "source"
+	BackupDataPathKey = "backup-data-path"
 )
 
 const (
-	HTTPKey                = "http"
-	AerospikeBackupService = "aerospike-backup-service"
+	HTTPKey                   = "http"
+	AerospikeBackupServiceKey = "aerospike-backup-service"
+	ForceRefreshKey           = AerospikeBackupServiceKey + "/force-refresh"
+	RefreshTimeKey            = AerospikeBackupServiceKey + "/last-refresh"
 )
diff --git a/api/v1beta1/utils.go b/api/v1beta1/utils.go
index 2219c1e4..adaede8b 100644
--- a/api/v1beta1/utils.go
+++ b/api/v1beta1/utils.go
@@ -1,8 +1,21 @@
 package v1beta1
 
 import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	utilRuntime "k8s.io/apimachinery/pkg/util/runtime"
+	clientGoScheme "k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/yaml"
+
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
+	asdbv1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1"
+	lib "github.com/aerospike/aerospike-management-lib"
 )
 
 func namespacedName(obj client.Object) string {
@@ -11,3 +24,75 @@ func namespacedName(obj client.Object) string {
 		Name:      obj.GetName(),
 	}.String()
 }
+
+func getK8sClient() (client.Client, error) {
+	restConfig := ctrl.GetConfigOrDie()
+
+	scheme := runtime.NewScheme()
+
+	utilRuntime.Must(asdbv1.AddToScheme(scheme))
+	utilRuntime.Must(clientGoScheme.AddToScheme(scheme))
+	utilRuntime.Must(AddToScheme(scheme))
+
+	cl, err := client.New(restConfig, client.Options{
+		Scheme: scheme,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return cl, nil
+}
+
+func getBackupServiceFullConfig(k8sClient client.Client, name, namespace string) (*dto.Config, error) {
+	var backupSvcConfigMap corev1.ConfigMap
+
+	if err := k8sClient.Get(context.TODO(),
+		types.NamespacedName{Name: name, Namespace: namespace},
+		&backupSvcConfigMap); err != nil {
+		return nil, err
+	}
+
+	var backupSvcConfig dto.Config
+
+	if err := yaml.Unmarshal([]byte(backupSvcConfigMap.Data[BackupServiceConfigYAML]),
+		&backupSvcConfig); err != nil {
+		return nil, err
+	}
+
+	return &backupSvcConfig, nil
+}
+
+func ValidateBackupSvcVersion(image string) error {
+	version, err := asdbv1.GetImageVersion(image)
+	if err != nil {
+		return err
+	}
+
+	val, err := lib.CompareVersions(version, minSupportedVersion)
+	if err != nil {
+		return fmt.Errorf("failed to check backup service image version: %v", err)
+	}
+
+	if val < 0 {
+		return fmt.Errorf("backup service version %s is not supported. Minimum supported version is %s",
+			version, minSupportedVersion)
+	}
+
+	return nil
+}
+
+// validateBackupSvcSupportedVersion validates the supported backup service version.
+// It returns an error if the backup service version is less than 3.0.0.
+func validateBackupSvcSupportedVersion(k8sClient client.Client, name, namespace string) error {
+	var backupSvc AerospikeBackupService
+
+	if err := k8sClient.Get(context.TODO(),
+		types.NamespacedName{Name: name, Namespace: namespace},
+		&backupSvc,
+	); err != nil {
+		return err
+	}
+
+	return ValidateBackupSvcVersion(backupSvc.Spec.Image)
+}
diff --git a/config/samples/aerospikebackup.yaml b/config/samples/aerospikebackup.yaml
index 2bc9392d..046366e9 100644
--- a/config/samples/aerospikebackup.yaml
+++ b/config/samples/aerospikebackup.yaml
@@ -35,4 +35,4 @@ spec:
         incr-interval-cron: "@hourly"
         namespaces: [ "test" ]
         source-cluster: aerospike-aerospikebackup-test-cluster
-        storage: s3Storage
+        storage: s3Storage
\ No newline at end of file
diff --git a/config/samples/aerospikebackupservice.yaml b/config/samples/aerospikebackupservice.yaml
index 893c92d6..0f4a40b7 100644
--- a/config/samples/aerospikebackupservice.yaml
+++ b/config/samples/aerospikebackupservice.yaml
@@ -18,14 +18,13 @@ spec:
         remove-files: KeepAll
     storage:
       local:
-        path: /localStorage
-        type: local
+        local-storage:
+          path: /localStorage
       s3Storage:
-        type: aws-s3
-        path: "s3://aerospike-kubernetes-operator-test"
-        s3-region: us-east-1
-        s3-endpoint-override: ""
-        s3-profile: default
+        s3-storage:
+          bucket: aerospike-kubernetes-operator-test
+          s3-region: us-east-1
+          s3-profile: default
 
   secrets:
     - secretName: aws-secret
@@ -36,4 +35,3 @@ spec:
 
   service:
     type: LoadBalancer
-
diff --git a/config/samples/aerospikerestore.yaml b/config/samples/aerospikerestore.yaml
index c25daded..8e4912f3 100644
--- a/config/samples/aerospikerestore.yaml
+++ b/config/samples/aerospikerestore.yaml
@@ -22,5 +22,6 @@ spec:
       no-generation: true
       no-indexes: true
     source:
-      "path": "/localStorage/aerospike-aerospikebackup-test-routine/backup/1722326391329/data/test"
-      "type": local
+      local-storage:
+        path: /localStorage
+    backup-data-path: aerospike-aerospikebackup-test-routine/backup/1733506015827/data/test
diff --git a/go.mod b/go.mod
index 89f44d26..76265db3 100644
--- a/go.mod
+++ b/go.mod
@@ -3,8 +3,8 @@ module github.com/aerospike/aerospike-kubernetes-operator
 go 1.22
 
 require (
-	github.com/aerospike/aerospike-backup-service v0.0.0-20240822110128-dc2b4811b9d3
-	github.com/aerospike/aerospike-client-go/v7 v7.6.1
+	github.com/aerospike/aerospike-backup-service/v2 v2.0.1-0.20241205081925-ebbb018935bd
+	github.com/aerospike/aerospike-client-go/v7 v7.8.0
 	github.com/aerospike/aerospike-management-lib v1.5.1-0.20250106091653-f0c86baa6cd7
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d
 	github.com/deckarep/golang-set/v2 v2.3.1
@@ -13,7 +13,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.16.0
 	github.com/onsi/gomega v1.32.0
 	github.com/sirupsen/logrus v1.9.1
-	github.com/stretchr/testify v1.9.0
+	github.com/stretchr/testify v1.10.0
 	golang.org/x/crypto v0.31.0
 	golang.org/x/net v0.33.0
 	gomodules.xyz/jsonpatch/v2 v2.4.0
@@ -27,7 +27,8 @@ require (
 )
 
 require (
-	github.com/aws/smithy-go v1.20.4 // indirect
+	github.com/aerospike/backup-go v0.3.1 // indirect
+	github.com/aws/smithy-go v1.22.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
@@ -51,6 +52,7 @@ require (
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/moby/spdystream v0.2.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -59,11 +61,11 @@ require (
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_golang v1.19.1 // indirect
-	github.com/prometheus/client_model v0.5.0 // indirect
-	github.com/prometheus/common v0.48.0 // indirect
-	github.com/prometheus/procfs v0.12.0 // indirect
-	github.com/reugn/go-quartz v0.12.0 // indirect
+	github.com/prometheus/client_golang v1.20.5 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.55.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/reugn/go-quartz v0.13.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
@@ -71,17 +73,17 @@ require (
 	github.com/yuin/gopher-lua v1.1.1 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
-	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
-	golang.org/x/oauth2 v0.20.0 // indirect
+	golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f // indirect
+	golang.org/x/oauth2 v0.24.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/term v0.27.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
-	golang.org/x/time v0.6.0 // indirect
+	golang.org/x/time v0.8.0 // indirect
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d // indirect
-	google.golang.org/grpc v1.65.0 // indirect
-	google.golang.org/protobuf v1.34.2 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576 // indirect
+	google.golang.org/grpc v1.67.3 // indirect
+	google.golang.org/protobuf v1.35.2 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
diff --git a/go.sum b/go.sum
index 6eeb9c24..302f2775 100644
--- a/go.sum
+++ b/go.sum
@@ -1,15 +1,17 @@
-github.com/aerospike/aerospike-backup-service v0.0.0-20240822110128-dc2b4811b9d3 h1:zb4SUg9JZw4tP7kYmBNhUOrg3iMy/2pqlP6SKsrpeiQ=
-github.com/aerospike/aerospike-backup-service v0.0.0-20240822110128-dc2b4811b9d3/go.mod h1:PFWhqxcMsEEyoOZtQ70b+X8xWbbemDYuitT24EPBizk=
-github.com/aerospike/aerospike-client-go/v7 v7.6.1 h1:VZK6S9YKq2w6ptTk3kXXjTxG2U9M9Y7Oi3YQ+3T7wQQ=
-github.com/aerospike/aerospike-client-go/v7 v7.6.1/go.mod h1:uCbSYMpjlRcH/9f26VSF/luzDDXrcDaV8c6/WIcKtT4=
+github.com/aerospike/aerospike-backup-service/v2 v2.0.1-0.20241205081925-ebbb018935bd h1:nnFUezmeTYrVn5tREDXISRc3hpRQPDT0n9FE1ZShn8o=
+github.com/aerospike/aerospike-backup-service/v2 v2.0.1-0.20241205081925-ebbb018935bd/go.mod h1:g3oSLu/Vj1NSZyzE34Ed4KBaE+S0v4GeLn2frZDMJ5U=
+github.com/aerospike/aerospike-client-go/v7 v7.8.0 h1:mKWTf/8sWQkWSYlIR3ZWXZMr9FQQPnIihrA+ujGD+n8=
+github.com/aerospike/aerospike-client-go/v7 v7.8.0/go.mod h1:STlBtOkKT8nmp7iD+sEkr/JGEOu+4e2jGlNN0Jiu2a4=
 github.com/aerospike/aerospike-management-lib v1.5.1-0.20250106091653-f0c86baa6cd7 h1:lhccxfqnvqdSNnpxdM4xjUnp8AzHsUKKqJPXRKIgfsw=
 github.com/aerospike/aerospike-management-lib v1.5.1-0.20250106091653-f0c86baa6cd7/go.mod h1:hsEptY/AmTmHoJnItJNmfJ4yCMG8LIB8YPnIpIyvGXI=
+github.com/aerospike/backup-go v0.3.1 h1:mKEyGl2WrN7UKbDi6ivAEfEdlvh+WS7N0oTYTURxiWs=
+github.com/aerospike/backup-go v0.3.1/go.mod h1:JwrUCJEtsUD0iHAs5yZY8+iF56nfw2m/DKvlBey77XE=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
 github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d h1:Byv0BzEl3/e6D5CLfI0j/7hiIEtvGVFPCZ7Ei2oq8iQ=
 github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
-github.com/aws/smithy-go v1.20.4 h1:2HK1zBdPgRbjFOHlfeQZfpC4r72MOb9bZkiFwggKO+4=
-github.com/aws/smithy-go v1.20.4/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
+github.com/aws/smithy-go v1.22.1 h1:/HPHZQ0g7f4eUeK6HKglFz8uwVfZKgoI25rb/J+dnro=
+github.com/aws/smithy-go v1.22.1/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
@@ -71,6 +73,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -78,6 +82,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/moby/spdystream v0.2.0 h1:cjW1zVyyoiM0T7b6UoySUFqzXMoqRckQtXwGPiBhOM8=
@@ -100,16 +106,16 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prometheus/client_golang v1.19.1 h1:wZWJDwK+NameRJuPGDhlnFgx8e8HN3XHQeLaYJFJBOE=
-github.com/prometheus/client_golang v1.19.1/go.mod h1:mP78NwGzrVks5S2H6ab8+ZZGJLZUq1hoULYBAYBw1Ho=
-github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw=
-github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
-github.com/prometheus/common v0.48.0 h1:QO8U2CdOzSn1BBsmXJXduaaW+dY/5QLjfB8svtSzKKE=
-github.com/prometheus/common v0.48.0/go.mod h1:0/KsvlIEfPQCQ5I2iNSAWKPZziNCvRs5EC6ILDTlAPc=
-github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
-github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
-github.com/reugn/go-quartz v0.12.0 h1:RsrklW++R5Swc7mCPYseXM06PTWN4N7/f1rsYkhHiww=
-github.com/reugn/go-quartz v0.12.0/go.mod h1:no4ktgYbAAuY0E1SchR8cTx1LF4jYIzdgaQhzRPSkpk=
+github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
+github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
+github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
+github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
+github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
+github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
+github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
+github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/reugn/go-quartz v0.13.0 h1:0eMxvj28Qu1npIDdN9Mzg9hwyksGH6XJt4Cz0QB8EUk=
+github.com/reugn/go-quartz v0.13.0/go.mod h1:0ghKksELp8MJ4h84T203aTHRF3Kug5BrxEW3ErBvhzY=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/sirupsen/logrus v1.9.1 h1:Ou41VVR3nMWWmTiEUnj0OlsgOSCUFgsPAOl6jRIcVtQ=
@@ -119,14 +125,16 @@ github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
-github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb h1:zGWFAtiMcyryUHoUjUJX0/lt1H2+i2Ka2n+D3DImSNo=
 github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
@@ -151,8 +159,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
-golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e h1:+WEEuIdZHnUeJJmEUjyYC2gfUMj69yZXw17EnHg/otA=
-golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e/go.mod h1:Kr81I6Kryrl9sr8s2FK3vxD90NdsKWRuOIl2O4CvYbA=
+golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f h1:99ci1mjWVBWwJiEKYY6jWa4d2nTQVIEhZIptnrVb1XY=
+golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f/go.mod h1:/lliqkxwWAhPjf5oSOIJup2XcqJaw8RGS6k3TGEc7GI=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -161,8 +169,8 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
 golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
-golang.org/x/oauth2 v0.20.0 h1:4mQdhULixXKP1rwYBW0vAijoXnkTG0BLCDRzfe1idMo=
-golang.org/x/oauth2 v0.20.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.24.0 h1:KTBBxWqUa0ykRPLtV69rRto9TLXcqYkeswu48x/gvNE=
+golang.org/x/oauth2 v0.24.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -180,8 +188,8 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
 golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
-golang.org/x/time v0.6.0 h1:eTDhh4ZXt5Qf0augr54TN6suAUudPcawVZeIAPU7D4U=
-golang.org/x/time v0.6.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
+golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg=
+golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
@@ -194,12 +202,12 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gomodules.xyz/jsonpatch/v2 v2.4.0 h1:Ci3iUJyx9UeRx7CeFN8ARgGbkESwJK+KB9lLcWxY/Zw=
 gomodules.xyz/jsonpatch/v2 v2.4.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d h1:JU0iKnSg02Gmb5ZdV8nYsKEKsP6o/FGVWTrw4i1DA9A=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20240711142825-46eb208f015d/go.mod h1:Ue6ibwXGpU+dqIcODieyLOcgj7z8+IcskoNIgZxtrFY=
-google.golang.org/grpc v1.65.0 h1:bs/cUb4lp1G5iImFFd3u5ixQzweKizoZJAwBNLR42lc=
-google.golang.org/grpc v1.65.0/go.mod h1:WgYC2ypjlB0EiQi6wdKixMqukr6lBc0Vo+oOgjrM5ZQ=
-google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
-google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576 h1:8ZmaLZE4XWrtU3MyClkYqqtl6Oegr3235h7jxsDyqCY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241209162323-e6fa225c2576/go.mod h1:5uTbfoYQed2U9p3KIj2/Zzm02PYhndfdmML0qC3q3FU=
+google.golang.org/grpc v1.67.3 h1:OgPcDAFKHnH8X3O4WcO4XUc8GRDeKsKReqbQtiCj7N8=
+google.golang.org/grpc v1.67.3/go.mod h1:YGaHCc6Oap+FzBJTZLBzkGSYt/cvGPFTPxkn7QfSU8s=
+google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
+google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
diff --git a/helm-charts/aerospike-backup-service/values.yaml b/helm-charts/aerospike-backup-service/values.yaml
index 577dd6df..af21d28e 100644
--- a/helm-charts/aerospike-backup-service/values.yaml
+++ b/helm-charts/aerospike-backup-service/values.yaml
@@ -35,14 +35,13 @@ backupServiceConfig: {}
 #      remove-files: KeepAll
 #  storage:
 #    local:
-#      path: /localStorage
-#      type: local
+#      local-storage:
+#        path: /localStorage
 #    s3Storage:
-#      type: aws-s3
-#      path: "s3://test-bucket"
-#      s3-region: us-east-1
-#      s3-endpoint-override: ""
-#      s3-profile: default
+#      s3-storage:
+#        bucket: aerospike-kubernetes-operator-test
+#        s3-region: us-east-1
+#        s3-profile: default
 
 ## SecretMounts is the list of secret to be mounted in the backup service.
 secrets: []
diff --git a/helm-charts/aerospike-cluster/values.yaml b/helm-charts/aerospike-cluster/values.yaml
index 4dd1bea8..1112b1b6 100644
--- a/helm-charts/aerospike-cluster/values.yaml
+++ b/helm-charts/aerospike-cluster/values.yaml
@@ -121,10 +121,10 @@ k8sNodeBlockList: []
 paused: false
 
 operations: []
-#  - kind: "WarmRestart"
-#    id: "1"
+#  - kind: WarmRestart
+#    id: warm-restart-1
 #    podList:
-#      - podName: "aerospike-cluster-0"
+#      - aerospike-cluster-0-1
 
 ## Dev Mode
 devMode: false
diff --git a/helm-charts/aerospike-restore/values.yaml b/helm-charts/aerospike-restore/values.yaml
index 83173755..abe38f3a 100644
--- a/helm-charts/aerospike-restore/values.yaml
+++ b/helm-charts/aerospike-restore/values.yaml
@@ -33,8 +33,9 @@ restoreConfig: {}
 #    no-generation: true
 #    no-indexes: true
 #  source:
-#    "path": "/localStorage/aerospike-aerospikebackup-test-routine/backup/1722326391329/data/test"
-#    "type": local
+#    local-storage:
+#      path: /localStorage
+#  backup-data-path: aerospike-aerospikebackup-test-routine/backup/1722326391329/data/test
 
 
 ## Polling period for restore operation status
diff --git a/internal/controller/backup-service/reconciler.go b/internal/controller/backup-service/reconciler.go
index 809f60d8..46ca3289 100644
--- a/internal/controller/backup-service/reconciler.go
+++ b/internal/controller/backup-service/reconciler.go
@@ -10,7 +10,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	k8sRuntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -22,6 +21,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/yaml"
 
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/validation"
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
 	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
 	"github.com/aerospike/aerospike-kubernetes-operator/pkg/utils"
@@ -34,7 +35,7 @@ type serviceConfig struct {
 
 var defaultServiceConfig = serviceConfig{
 	portInfo: map[string]int32{
-		common.HTTPKey: 8080,
+		asdbv1beta1.HTTPKey: 8080,
 	},
 	contextPath: "/",
 }
@@ -62,6 +63,13 @@ func (r *SingleBackupServiceReconciler) Reconcile() (result ctrl.Result, recErr
 		}
 	}()
 
+	// Skip reconcile if the backup service version is less than 3.0.0.
+	// This is to avoid rolling restart of the backup service pods after AKO upgrade
+	if err := asdbv1beta1.ValidateBackupSvcVersion(r.aeroBackupService.Spec.Image); err != nil {
+		r.Log.Info("Skipping reconcile as backup service version is less than 3.0.0")
+		return reconcile.Result{}, nil
+	}
+
 	if !r.aeroBackupService.ObjectMeta.DeletionTimestamp.IsZero() {
 		r.Log.Info("Deleted AerospikeBackupService")
 		r.Recorder.Eventf(
@@ -161,7 +169,7 @@ func (r *SingleBackupServiceReconciler) reconcileConfigMap() error {
 			context.TODO(), cm, common.CreateOption,
 		); err != nil {
 			return fmt.Errorf(
-				"failed to create ConfigMap: %v",
+				"failed to create ConfigMap: %w",
 				err,
 			)
 		}
@@ -186,23 +194,38 @@ func (r *SingleBackupServiceReconciler) reconcileConfigMap() error {
 		return err
 	}
 
-	data := cm.Data[common.BackupServiceConfigYAML]
+	data := cm.Data[asdbv1beta1.BackupServiceConfigYAML]
 
 	if err := yaml.Unmarshal([]byte(data), &currentDataMap); err != nil {
 		return err
 	}
 
-	currentDataMap[common.ServiceKey] = desiredDataMap[common.ServiceKey]
-	currentDataMap[common.BackupPoliciesKey] = desiredDataMap[common.BackupPoliciesKey]
-	currentDataMap[common.StorageKey] = desiredDataMap[common.StorageKey]
-	currentDataMap[common.SecretAgentsKey] = desiredDataMap[common.SecretAgentsKey]
+	// Sync keys
+	keys := []string{
+		asdbv1beta1.ServiceKey,
+		asdbv1beta1.BackupPoliciesKey,
+		asdbv1beta1.StorageKey,
+		asdbv1beta1.SecretAgentsKey,
+	}
+
+	for _, key := range keys {
+		if value, ok := desiredDataMap[key]; ok {
+			currentDataMap[key] = value
+		} else {
+			delete(currentDataMap, key)
+		}
+	}
+
+	// Remove old "secret-agent: null" from configMap
+	// This was added internally in AKO (3.4) during backup service configMap update
+	delete(currentDataMap, "secret-agent")
 
 	updatedConfig, err := yaml.Marshal(currentDataMap)
 	if err != nil {
 		return err
 	}
 
-	cm.Data[common.BackupServiceConfigYAML] = string(updatedConfig)
+	cm.Data[asdbv1beta1.BackupServiceConfigYAML] = string(updatedConfig)
 
 	if err = r.Client.Update(
 		context.TODO(), cm, common.UpdateOption,
@@ -223,7 +246,7 @@ func (r *SingleBackupServiceReconciler) reconcileConfigMap() error {
 
 func (r *SingleBackupServiceReconciler) getConfigMapData() map[string]string {
 	data := make(map[string]string)
-	data[common.BackupServiceConfigYAML] = string(r.aeroBackupService.Spec.Config.Raw)
+	data[asdbv1beta1.BackupServiceConfigYAML] = string(r.aeroBackupService.Spec.Config.Raw)
 
 	return data
 }
@@ -313,45 +336,64 @@ func (r *SingleBackupServiceReconciler) reconcileDeployment() error {
 		return err
 	}
 
-	// If there is a change in config hash, then restart the deployment pod
+	// If there is a change in config hash, then reload the config or restart the deployment pod
 	if desiredHash != currentHash {
-		r.Log.Info("BackupService config is updated, will result in rolling restart")
+		r.Log.Info("BackupService config mismatch, will reload the config")
 
-		podList, err := r.getBackupServicePodList()
-		if err != nil {
+		if err := r.updateBackupSvcConfig(); err != nil {
 			return err
 		}
 
-		for idx := range podList.Items {
-			pod := &podList.Items[idx]
-
-			err = r.Client.Delete(context.TODO(), pod)
-			if err != nil {
-				return err
-			}
-		}
+		r.Log.Info("Reloaded backup service")
 	}
 
-	return r.waitForDeploymentToBeReady()
+	return nil
 }
 
-func getBackupServiceName(aeroBackupService *asdbv1beta1.AerospikeBackupService) types.NamespacedName {
-	return types.NamespacedName{Name: aeroBackupService.Name, Namespace: aeroBackupService.Namespace}
-}
+func (r *SingleBackupServiceReconciler) updateBackupSvcConfig() error {
+	var currentConfig, desiredConfig dto.Config
 
-func (r *SingleBackupServiceReconciler) getBackupServicePodList() (*corev1.PodList, error) {
-	var podList corev1.PodList
+	if err := yaml.Unmarshal(r.aeroBackupService.Status.Config.Raw, &currentConfig); err != nil {
+		return err
+	}
 
-	labelSelector := labels.SelectorFromSet(utils.LabelsForAerospikeBackupService(r.aeroBackupService.Name))
-	listOps := &client.ListOptions{
-		Namespace: r.aeroBackupService.Namespace, LabelSelector: labelSelector,
+	if err := yaml.Unmarshal(r.aeroBackupService.Spec.Config.Raw, &desiredConfig); err != nil {
+		return err
 	}
 
-	if err := r.Client.List(context.TODO(), &podList, listOps); err != nil {
-		return nil, err
+	if err := validation.ValidateStaticFieldChanges(&currentConfig, &desiredConfig); err != nil {
+		r.Log.Info("Static config change detected, will result in rolling restart")
+		// In case of static config change restart the backup service pod
+		return r.restartBackupSvcPod()
 	}
 
-	return &podList, nil
+	return common.ReloadBackupServiceConfigInPods(r.Client, r.Log,
+		&asdbv1beta1.BackupService{
+			Name:      r.aeroBackupService.Name,
+			Namespace: r.aeroBackupService.Namespace},
+	)
+}
+
+func (r *SingleBackupServiceReconciler) restartBackupSvcPod() error {
+	podList, err := common.GetBackupServicePodList(r.Client, r.aeroBackupService.Name, r.aeroBackupService.Namespace)
+	if err != nil {
+		return err
+	}
+
+	for idx := range podList.Items {
+		pod := &podList.Items[idx]
+
+		err = r.Client.Delete(context.TODO(), pod)
+		if err != nil {
+			return err
+		}
+	}
+
+	return r.waitForDeploymentToBeReady()
+}
+
+func getBackupServiceName(aeroBackupService *asdbv1beta1.AerospikeBackupService) types.NamespacedName {
+	return types.NamespacedName{Name: aeroBackupService.Name, Namespace: aeroBackupService.Namespace}
 }
 
 func (r *SingleBackupServiceReconciler) getDeploymentObject() (*app.Deployment, error) {
@@ -395,10 +437,10 @@ func (r *SingleBackupServiceReconciler) getDeploymentObject() (*app.Deployment,
 				},
 				Spec: corev1.PodSpec{
 					// TODO: Finalise on this. Who should create this SA?
-					ServiceAccountName: common.AerospikeBackupService,
+					ServiceAccountName: asdbv1beta1.AerospikeBackupServiceKey,
 					Containers: []corev1.Container{
 						{
-							Name:            common.AerospikeBackupService,
+							Name:            asdbv1beta1.AerospikeBackupServiceKey,
 							Image:           r.aeroBackupService.Spec.Image,
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							VolumeMounts:    volumeMounts,
@@ -406,30 +448,6 @@ func (r *SingleBackupServiceReconciler) getDeploymentObject() (*app.Deployment,
 							Ports:           containerPorts,
 						},
 					},
-					// Init-container is used to copy configMap data to work-dir(emptyDir).
-					// There is a limitation of read-only file-system for mounted configMap volumes
-					// Remove this init-container when backup-service start supporting hot reload
-					InitContainers: []corev1.Container{
-						{
-							Name:  "init-backup-service",
-							Image: "busybox",
-							Command: []string{
-								"sh",
-								"-c",
-								"cp /etc/aerospike-backup-service/aerospike-backup-service.yml /work-dir/aerospike-backup-service.yml",
-							},
-							VolumeMounts: []corev1.VolumeMount{
-								{
-									Name:      "backup-service-config-configmap",
-									MountPath: "/etc/aerospike-backup-service/",
-								},
-								{
-									Name:      "backup-service-config",
-									MountPath: "/work-dir",
-								},
-							},
-						},
-					},
 					Volumes: volumes,
 				},
 			},
@@ -460,13 +478,12 @@ func (r *SingleBackupServiceReconciler) getVolumeAndMounts() ([]corev1.VolumeMou
 	// Backup service configMap mountPath
 	volumeMounts = append(volumeMounts, corev1.VolumeMount{
 		Name:      "backup-service-config",
-		MountPath: fmt.Sprintf("/etc/aerospike-backup-service/%s", common.BackupServiceConfigYAML),
-		SubPath:   common.BackupServiceConfigYAML,
+		MountPath: "/etc/aerospike-backup-service",
 	})
 
 	// Backup service configMap
 	volumes = append(volumes, corev1.Volume{
-		Name: "backup-service-config-configmap",
+		Name: "backup-service-config",
 		VolumeSource: corev1.VolumeSource{
 			ConfigMap: &corev1.ConfigMapVolumeSource{
 				LocalObjectReference: corev1.LocalObjectReference{
@@ -476,15 +493,6 @@ func (r *SingleBackupServiceReconciler) getVolumeAndMounts() ([]corev1.VolumeMou
 		},
 	})
 
-	// EmptyDir for init-container to copy configMap data to work-dir
-	// Remove this volume when backup-service starts supporting hot reload
-	volumes = append(volumes, corev1.Volume{
-		Name: "backup-service-config",
-		VolumeSource: corev1.VolumeSource{
-			EmptyDir: &corev1.EmptyDirVolumeSource{},
-		},
-	})
-
 	return volumeMounts, volumes
 }
 
@@ -594,22 +602,22 @@ func (r *SingleBackupServiceReconciler) getBackupServiceConfig() (*serviceConfig
 		return nil, err
 	}
 
-	if _, ok := config[common.ServiceKey]; !ok {
+	if _, ok := config[asdbv1beta1.ServiceKey]; !ok {
 		r.Log.Info("Service config not found")
 		return &defaultServiceConfig, nil
 	}
 
-	svc, ok := config[common.ServiceKey].(map[string]interface{})
+	svc, ok := config[asdbv1beta1.ServiceKey].(map[string]interface{})
 	if !ok {
 		return nil, fmt.Errorf("service config is not in correct format")
 	}
 
-	if _, ok = svc[common.HTTPKey]; !ok {
+	if _, ok = svc[asdbv1beta1.HTTPKey]; !ok {
 		r.Log.Info("HTTP config not found")
 		return &defaultServiceConfig, nil
 	}
 
-	httpConf, ok := svc[common.HTTPKey].(map[string]interface{})
+	httpConf, ok := svc[asdbv1beta1.HTTPKey].(map[string]interface{})
 	if !ok {
 		return nil, fmt.Errorf("http config is not in correct format")
 	}
@@ -620,7 +628,7 @@ func (r *SingleBackupServiceReconciler) getBackupServiceConfig() (*serviceConfig
 	if !ok {
 		svcConfig.portInfo = defaultServiceConfig.portInfo
 	} else {
-		svcConfig.portInfo = map[string]int32{common.HTTPKey: int32(port.(float64))}
+		svcConfig.portInfo = map[string]int32{asdbv1beta1.HTTPKey: int32(port.(float64))}
 	}
 
 	ctxPath, ok := httpConf["context-path"]
@@ -645,7 +653,7 @@ func (r *SingleBackupServiceReconciler) waitForDeploymentToBeReady() error {
 
 	if err := wait.PollUntilContextTimeout(context.TODO(),
 		podStatusRetryInterval, podStatusTimeout, true, func(ctx context.Context) (done bool, err error) {
-			podList, err := r.getBackupServicePodList()
+			podList, err := common.GetBackupServicePodList(r.Client, r.aeroBackupService.Name, r.aeroBackupService.Namespace)
 			if err != nil {
 				return false, err
 			}
@@ -713,7 +721,7 @@ func (r *SingleBackupServiceReconciler) updateStatus() error {
 
 	status := r.CopySpecToStatus()
 	status.ContextPath = svcConfig.contextPath
-	status.Port = svcConfig.portInfo[common.HTTPKey]
+	status.Port = svcConfig.portInfo[asdbv1beta1.HTTPKey]
 	status.Phase = asdbv1beta1.AerospikeBackupServiceCompleted
 
 	r.aeroBackupService.Status = *status
diff --git a/internal/controller/backup/reconciler.go b/internal/controller/backup/reconciler.go
index 6edeafd6..b76c44e4 100644
--- a/internal/controller/backup/reconciler.go
+++ b/internal/controller/backup/reconciler.go
@@ -39,7 +39,7 @@ func (r *SingleBackupReconciler) Reconcile() (result ctrl.Result, recErr error)
 	if !r.aeroBackup.ObjectMeta.DeletionTimestamp.IsZero() {
 		r.Log.Info("Deleting AerospikeBackup")
 
-		if err := r.removeFinalizer(finalizerName); err != nil {
+		if err := r.cleanUpAndRemoveFinalizer(finalizerName); err != nil {
 			r.Log.Error(err, "Failed to remove finalizer")
 			return reconcile.Result{}, err
 		}
@@ -108,17 +108,18 @@ func (r *SingleBackupReconciler) addFinalizer(finalizerName string) error {
 	return nil
 }
 
-func (r *SingleBackupReconciler) removeFinalizer(finalizerName string) error {
+func (r *SingleBackupReconciler) cleanUpAndRemoveFinalizer(finalizerName string) error {
 	if utils.ContainsString(r.aeroBackup.ObjectMeta.Finalizers, finalizerName) {
+		r.Log.Info("Removing finalizer")
+
 		if err := r.removeBackupInfoFromConfigMap(); err != nil {
 			return err
 		}
 
-		if err := r.unregisterBackup(); err != nil {
+		if err := common.ReloadBackupServiceConfigInPods(r.Client, r.Log, &r.aeroBackup.Spec.BackupService); err != nil {
 			return err
 		}
 
-		r.Log.Info("Removing finalizer")
 		// Remove finalizer from the list
 		r.aeroBackup.ObjectMeta.Finalizers = utils.RemoveString(
 			r.aeroBackup.ObjectMeta.Finalizers, finalizerName,
@@ -127,6 +128,8 @@ func (r *SingleBackupReconciler) removeFinalizer(finalizerName string) error {
 		if err := r.Client.Update(context.TODO(), r.aeroBackup); err != nil {
 			return err
 		}
+
+		r.Log.Info("Removed finalizer")
 	}
 
 	return nil
@@ -150,19 +153,19 @@ func (r *SingleBackupReconciler) reconcileConfigMap() error {
 
 	backupSvcConfig := make(map[string]interface{})
 
-	data := cm.Data[common.BackupServiceConfigYAML]
+	data := cm.Data[asdbv1beta1.BackupServiceConfigYAML]
 
 	err = yaml.Unmarshal([]byte(data), &backupSvcConfig)
 	if err != nil {
 		return err
 	}
 
-	clusterMap, err := common.GetConfigSection(backupSvcConfig, common.AerospikeClustersKey)
+	clusterMap, err := common.GetConfigSection(backupSvcConfig, asdbv1beta1.AerospikeClustersKey)
 	if err != nil {
 		return err
 	}
 
-	cluster := specBackupConfig[common.AerospikeClusterKey].(map[string]interface{})
+	cluster := specBackupConfig[asdbv1beta1.AerospikeClusterKey].(map[string]interface{})
 
 	var clusterName string
 
@@ -175,14 +178,14 @@ func (r *SingleBackupReconciler) reconcileConfigMap() error {
 		clusterMap[name] = clusterInfo
 	}
 
-	backupSvcConfig[common.AerospikeClustersKey] = clusterMap
+	backupSvcConfig[asdbv1beta1.AerospikeClustersKey] = clusterMap
 
-	routineMap, err := common.GetConfigSection(backupSvcConfig, common.BackupRoutinesKey)
+	routineMap, err := common.GetConfigSection(backupSvcConfig, asdbv1beta1.BackupRoutinesKey)
 	if err != nil {
 		return err
 	}
 
-	routines := specBackupConfig[common.BackupRoutinesKey].(map[string]interface{})
+	routines := specBackupConfig[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 
 	// Remove the routines which are not in spec
 	routinesToBeDeleted := r.routinesToDelete(routines, routineMap, clusterName)
@@ -196,14 +199,14 @@ func (r *SingleBackupReconciler) reconcileConfigMap() error {
 		routineMap[name] = routine
 	}
 
-	backupSvcConfig[common.BackupRoutinesKey] = routineMap
+	backupSvcConfig[asdbv1beta1.BackupRoutinesKey] = routineMap
 
 	updatedConfig, err := yaml.Marshal(backupSvcConfig)
 	if err != nil {
 		return err
 	}
 
-	cm.Data[common.BackupServiceConfigYAML] = string(updatedConfig)
+	cm.Data[asdbv1beta1.BackupServiceConfigYAML] = string(updatedConfig)
 
 	if err := r.Client.Update(
 		context.TODO(), cm, common.UpdateOption,
@@ -247,7 +250,7 @@ func (r *SingleBackupReconciler) removeBackupInfoFromConfigMap() error {
 
 	backupSvcConfig := make(map[string]interface{})
 
-	data := cm.Data[common.BackupServiceConfigYAML]
+	data := cm.Data[asdbv1beta1.BackupServiceConfigYAML]
 
 	err = yaml.Unmarshal([]byte(data), &backupSvcConfig)
 	if err != nil {
@@ -256,19 +259,19 @@ func (r *SingleBackupReconciler) removeBackupInfoFromConfigMap() error {
 
 	var clusterName string
 
-	if clusterIface, ok := backupSvcConfig[common.AerospikeClustersKey]; ok {
+	if clusterIface, ok := backupSvcConfig[asdbv1beta1.AerospikeClustersKey]; ok {
 		if clusterMap, ok := clusterIface.(map[string]interface{}); ok {
-			currentCluster := specBackupConfig[common.AerospikeClusterKey].(map[string]interface{})
+			currentCluster := specBackupConfig[asdbv1beta1.AerospikeClusterKey].(map[string]interface{})
 			for name := range currentCluster {
 				clusterName = name
 				delete(clusterMap, name)
 			}
 
-			backupSvcConfig[common.AerospikeClustersKey] = clusterMap
+			backupSvcConfig[asdbv1beta1.AerospikeClustersKey] = clusterMap
 		}
 	}
 
-	if routineIface, ok := backupSvcConfig[common.BackupRoutinesKey]; ok {
+	if routineIface, ok := backupSvcConfig[asdbv1beta1.BackupRoutinesKey]; ok {
 		if routineMap, ok := routineIface.(map[string]interface{}); ok {
 			routinesToBeDelete := r.routinesToDelete(nil, routineMap, clusterName)
 
@@ -276,7 +279,7 @@ func (r *SingleBackupReconciler) removeBackupInfoFromConfigMap() error {
 				delete(routineMap, routinesToBeDelete[idx])
 			}
 
-			backupSvcConfig[common.BackupRoutinesKey] = routineMap
+			backupSvcConfig[asdbv1beta1.BackupRoutinesKey] = routineMap
 		}
 	}
 
@@ -285,7 +288,7 @@ func (r *SingleBackupReconciler) removeBackupInfoFromConfigMap() error {
 		return err
 	}
 
-	cm.Data[common.BackupServiceConfigYAML] = string(updatedConfig)
+	cm.Data[asdbv1beta1.BackupServiceConfigYAML] = string(updatedConfig)
 
 	if err := r.Client.Update(
 		context.TODO(), cm, common.UpdateOption,
@@ -366,82 +369,43 @@ func (r *SingleBackupReconciler) reconcileScheduledBackup() error {
 		return err
 	}
 
-	if specBackupConfig[common.AerospikeClusterKey] != nil {
-		cluster := specBackupConfig[common.AerospikeClusterKey].(map[string]interface{})
+	var (
+		hotReloadRequired bool
+		clusterName       string
+	)
 
-		currentClusters, gErr := common.GetConfigSection(backupSvcConfig, common.AerospikeClustersKey)
-		if gErr != nil {
-			return gErr
-		}
+	if cluster, ok := specBackupConfig[asdbv1beta1.AerospikeClusterKey].(map[string]interface{}); ok {
+		hotReloadRequired = r.checkForConfigUpdate(
+			cluster,
+			asdbv1beta1.AerospikeClustersKey,
+			backupSvcConfig,
+		)
 
-		// TODO: Remove these API calls when hot reload is implemented
-		for name, clusterConfig := range cluster {
-			if _, ok := currentClusters[name]; ok {
-				// Only update if there is any change
-				if !reflect.DeepEqual(currentClusters[name], clusterConfig) {
-					r.Log.Info("Cluster config has been changed, updating it", "cluster", name)
-
-					err = serviceClient.PutCluster(name, clusterConfig)
-					if err != nil {
-						return err
-					}
-				}
-			} else {
-				r.Log.Info("Adding new cluster", "cluster", name)
-
-				err = serviceClient.AddCluster(name, clusterConfig)
-				if err != nil {
-					return err
-				}
-
-				r.Log.Info("Added new cluster", "cluster", name)
-			}
+		for name := range cluster {
+			clusterName = name
 		}
 	}
 
-	if specBackupConfig[common.BackupRoutinesKey] != nil {
-		routines := specBackupConfig[common.BackupRoutinesKey].(map[string]interface{})
+	// Skip further checks if hotReloadRequired is already true
+	if !hotReloadRequired {
+		if routines, ok := specBackupConfig[asdbv1beta1.BackupRoutinesKey].(map[string]interface{}); ok {
+			hotReloadRequired = r.checkForConfigUpdate(
+				routines,
+				asdbv1beta1.BackupRoutinesKey,
+				backupSvcConfig,
+			)
 
-		currentRoutines, gErr := common.GetConfigSection(backupSvcConfig, common.BackupRoutinesKey)
-		if gErr != nil {
-			return gErr
-		}
-
-		// TODO: Remove these API calls when hot reload is implemented
-		for name, routine := range routines {
-			if _, ok := currentRoutines[name]; ok {
-				// Only update if there is any change
-				if !reflect.DeepEqual(currentRoutines[name], routine) {
-					r.Log.Info("Routine config has been changed, updating it", "routine", name)
-
-					err = serviceClient.PutBackupRoutine(name, routine)
-					if err != nil {
-						return err
-					}
-				}
-			} else {
-				r.Log.Info("Adding new backup routine", "routine", name)
-
-				err = serviceClient.AddBackupRoutine(name, routine)
-				if err != nil {
-					return err
-				}
-
-				r.Log.Info("Added new backup routine", "routine", name)
+			if !hotReloadRequired {
+				hotReloadRequired = r.checkForDeletedRoutines(routines, backupSvcConfig, clusterName)
 			}
 		}
 	}
 
-	// If there are routines that are removed, unregister them
-	err = r.deregisterBackupRoutines(serviceClient, backupSvcConfig, specBackupConfig)
-	if err != nil {
-		return err
-	}
-
-	// Apply the updated configuration for the changes to take effect
-	err = serviceClient.ApplyConfig()
-	if err != nil {
-		return err
+	if hotReloadRequired {
+		err = common.ReloadBackupServiceConfigInPods(r.Client, r.Log, &r.aeroBackup.Spec.BackupService)
+		if err != nil {
+			return err
+		}
 	}
 
 	r.Log.Info("Reconciled scheduled backup")
@@ -451,102 +415,66 @@ func (r *SingleBackupReconciler) reconcileScheduledBackup() error {
 	return nil
 }
 
-func (r *SingleBackupReconciler) reconcileOnDemandBackup() error {
-	// Schedule on-demand backup if given
-	if len(r.aeroBackup.Spec.OnDemandBackups) > 0 {
-		if err := r.scheduleOnDemandBackup(); err != nil {
-			r.Log.Error(err, "Failed to schedule backup")
-			return err
-		}
-	}
-
-	return nil
-}
-
-func (r *SingleBackupReconciler) unregisterBackup() error {
-	serviceClient, err := backup_service.GetBackupServiceClient(r.Client, &r.aeroBackup.Spec.BackupService)
-	if err != nil {
-		return err
-	}
-
-	backupSvcConfig, err := serviceClient.GetBackupServiceConfig()
-	if err != nil {
-		return err
-	}
-
-	specBackupConfig, err := r.getBackupConfigInMap()
-	if err != nil {
-		return err
-	}
+func (r *SingleBackupReconciler) checkForConfigUpdate(
+	desiredConfig map[string]interface{},
+	sectionKey string,
+	backupSvcConfig map[string]interface{},
+) bool {
+	updated := false
 
-	err = r.deregisterBackupRoutines(serviceClient, backupSvcConfig, specBackupConfig)
+	currentConfig, err := common.GetConfigSection(backupSvcConfig, sectionKey)
 	if err != nil {
-		return err
+		r.Log.Error(err, "Failed to fetch config section", "section", sectionKey)
+		return false
 	}
 
-	if specBackupConfig[common.AerospikeClusterKey] != nil {
-		cluster := specBackupConfig[common.AerospikeClusterKey].(map[string]interface{})
-
-		currentClusters, gErr := common.GetConfigSection(backupSvcConfig, common.AerospikeClustersKey)
-		if gErr != nil {
-			return gErr
-		}
+	for name, config := range desiredConfig {
+		if existingConfig, exists := currentConfig[name]; exists {
+			if !reflect.DeepEqual(existingConfig, config) {
+				r.Log.Info(
+					fmt.Sprintf("%s config has changed, updating", sectionKey), "name", name,
+				)
 
-		for name := range cluster {
-			if _, ok := currentClusters[name]; ok {
-				err = serviceClient.DeleteCluster(name)
-				if err != nil {
-					return err
-				}
+				updated = true
 			}
-		}
-	}
+		} else {
+			r.Log.Info(
+				fmt.Sprintf("Adding new entry in %s config", sectionKey), "name", name,
+			)
 
-	// Apply the updated configuration for the changes to take effect
-	err = serviceClient.ApplyConfig()
-	if err != nil {
-		return err
+			updated = true
+		}
 	}
 
-	return nil
+	return updated
 }
 
-func (r *SingleBackupReconciler) deregisterBackupRoutines(
-	serviceClient *backup_service.Client,
-	backupSvcConfig,
-	specBackupConfig map[string]interface{},
-) error {
-	allRoutines, err := common.GetConfigSection(backupSvcConfig, common.BackupRoutinesKey)
+func (r *SingleBackupReconciler) checkForDeletedRoutines(
+	desired map[string]interface{},
+	currentConfig map[string]interface{},
+	clusterName string,
+) bool {
+	currentRoutines, err := common.GetConfigSection(currentConfig, asdbv1beta1.BackupRoutinesKey)
 	if err != nil {
-		return err
+		r.Log.Error(err, "Failed to fetch current routines")
+		return false
 	}
 
-	cluster := specBackupConfig[common.AerospikeClusterKey].(map[string]interface{})
-
-	var clusterName string
-
-	// There will always be only one cluster in the backup config
-	for name := range cluster {
-		clusterName = name
-	}
-
-	specRoutines := make(map[string]interface{})
-
-	// Ignore routines from the spec if the backup is being deleted
-	if r.aeroBackup.DeletionTimestamp.IsZero() {
-		specRoutines = specBackupConfig[common.BackupRoutinesKey].(map[string]interface{})
+	toDelete := r.routinesToDelete(desired, currentRoutines, clusterName)
+	if len(toDelete) > 0 {
+		r.Log.Info("Routines to be deleted", "count", len(toDelete))
+		return true
 	}
 
-	routinesToBeDelete := r.routinesToDelete(specRoutines, allRoutines, clusterName)
-
-	for idx := range routinesToBeDelete {
-		r.Log.Info("Unregistering backup routine", "routine", routinesToBeDelete[idx])
-
-		if err := serviceClient.DeleteBackupRoutine(routinesToBeDelete[idx]); err != nil {
+	return false
+}
+func (r *SingleBackupReconciler) reconcileOnDemandBackup() error {
+	// Schedule on-demand backup if given
+	if len(r.aeroBackup.Spec.OnDemandBackups) > 0 {
+		if err := r.scheduleOnDemandBackup(); err != nil {
+			r.Log.Error(err, "Failed to schedule backup")
 			return err
 		}
-
-		r.Log.Info("Unregistered backup routine", "routine", routinesToBeDelete[idx])
 	}
 
 	return nil
@@ -588,7 +516,7 @@ func (r *SingleBackupReconciler) routinesToDelete(
 		// Delete any dangling backup-routines related to this cluster
 		// Strict prefix check might fail for cases where the prefix is same.
 		if strings.HasPrefix(name, r.aeroBackup.NamePrefix()) &&
-			allRoutines[name].(map[string]interface{})[common.SourceClusterKey].(string) == clusterName {
+			allRoutines[name].(map[string]interface{})[asdbv1beta1.SourceClusterKey].(string) == clusterName {
 			routinesTobeDeleted = append(routinesTobeDeleted, name)
 		}
 	}
diff --git a/internal/controller/common/backup_config_util.go b/internal/controller/common/backup_config_util.go
index 6248f08d..b3f1dc54 100644
--- a/internal/controller/common/backup_config_util.go
+++ b/internal/controller/common/backup_config_util.go
@@ -1,6 +1,23 @@
 package common
 
-import "fmt"
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"time"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/retry"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/yaml"
+
+	"github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
+	backup_service "github.com/aerospike/aerospike-kubernetes-operator/pkg/backup-service"
+	"github.com/aerospike/aerospike-kubernetes-operator/pkg/utils"
+)
 
 // GetConfigSection returns the section of the config with the given name.
 func GetConfigSection(config map[string]interface{}, section string) (map[string]interface{}, error) {
@@ -16,3 +33,115 @@ func GetConfigSection(config map[string]interface{}, section string) (map[string
 
 	return sectionMap, nil
 }
+
+func GetBackupServicePodList(k8sClient client.Client, name, namespace string) (*corev1.PodList, error) {
+	var podList corev1.PodList
+
+	labelSelector := labels.SelectorFromSet(utils.LabelsForAerospikeBackupService(name))
+	listOps := &client.ListOptions{
+		Namespace: namespace, LabelSelector: labelSelector,
+	}
+
+	if err := k8sClient.List(context.TODO(), &podList, listOps); err != nil {
+		return nil, err
+	}
+
+	return &podList, nil
+}
+
+func ReloadBackupServiceConfigInPods(
+	k8sClient client.Client,
+	log logr.Logger,
+	backupSvc *v1beta1.BackupService,
+) error {
+	log.Info("Reloading backup service config")
+
+	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		podList, err := GetBackupServicePodList(k8sClient,
+			backupSvc.Name,
+			backupSvc.Namespace,
+		)
+		if err != nil {
+			return fmt.Errorf("failed to get backup service pod list, error: %v", err)
+		}
+
+		for idx := range podList.Items {
+			pod := podList.Items[idx]
+			annotations := pod.Annotations
+
+			if annotations == nil {
+				annotations = make(map[string]string)
+			}
+
+			annotations[v1beta1.RefreshTimeKey] = time.Now().Format(time.RFC3339)
+
+			pod.Annotations = annotations
+
+			if err := k8sClient.Update(context.TODO(), &pod); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	// Waiting for 1 second so that pods get the latest configMap update.
+	time.Sleep(1 * time.Second)
+
+	backupServiceClient, err := backup_service.GetBackupServiceClient(k8sClient, backupSvc)
+	if err != nil {
+		return err
+	}
+
+	err = backupServiceClient.ApplyConfig()
+	if err != nil {
+		return err
+	}
+
+	// TODO:// uncomment this when backup service removes default fields from the GET config API response
+	// return validateBackupSvcConfigReload(k8sClient, backupServiceClient, log, backupSvc)
+	log.Info("Reloaded backup service config")
+
+	return nil
+}
+
+//nolint:unused // for future use
+func validateBackupSvcConfigReload(k8sClient client.Client,
+	backupServiceClient *backup_service.Client,
+	log logr.Logger,
+	backupSvc *v1beta1.BackupService,
+) error {
+	apiBackupSvcConfig, err := backupServiceClient.GetBackupServiceConfig()
+	if err != nil {
+		return err
+	}
+
+	var cm corev1.ConfigMap
+
+	if err := k8sClient.Get(context.TODO(), types.NamespacedName{
+		Namespace: backupSvc.Namespace,
+		Name:      backupSvc.Name,
+	}, &cm); err != nil {
+		return err
+	}
+
+	configMapBackupSvcConfig := make(map[string]interface{})
+
+	data := cm.Data[v1beta1.BackupServiceConfigYAML]
+
+	if err := yaml.Unmarshal([]byte(data), &configMapBackupSvcConfig); err != nil {
+		return err
+	}
+
+	log.Info(fmt.Sprintf("API Backup Service Config: %v", apiBackupSvcConfig))
+	log.Info(fmt.Sprintf("ConfigMap Backup Service Config: %v", configMapBackupSvcConfig))
+
+	if !reflect.DeepEqual(apiBackupSvcConfig, configMapBackupSvcConfig) {
+		log.Info("Backup service config not yet updated in pods, requeue")
+		return fmt.Errorf("backup service config not yet updated in pods")
+	}
+
+	return nil
+}
diff --git a/internal/controller/restore/reconciler.go b/internal/controller/restore/reconciler.go
index 30075efb..b2e5d02d 100644
--- a/internal/controller/restore/reconciler.go
+++ b/internal/controller/restore/reconciler.go
@@ -43,6 +43,12 @@ func (r *SingleRestoreReconciler) Reconcile() (result ctrl.Result, recErr error)
 		return reconcile.Result{}, nil
 	}
 
+	if r.aeroRestore.Status.Phase == asdbv1beta1.AerospikeRestoreCompleted {
+		// Stop reconciliation as the Aerospike restore is already completed
+		r.Log.Info("Restore already completed, skipping reconciliation")
+		return reconcile.Result{}, nil
+	}
+
 	if err := r.setStatusPhase(asdbv1beta1.AerospikeRestoreInProgress); err != nil {
 		return ctrl.Result{}, err
 	}
diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go
index 4420f297..7c06482b 100644
--- a/pkg/utils/utils.go
+++ b/pkg/utils/utils.go
@@ -13,7 +13,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	asdbv1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
+	"github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
 )
 
 const (
@@ -147,7 +147,7 @@ func LabelsForPodAntiAffinity(clName string) map[string]string {
 // belonging to the given AerospikeBackupService CR name.
 func LabelsForAerospikeBackupService(clName string) map[string]string {
 	return map[string]string{
-		asdbv1.AerospikeAppLabel:            common.AerospikeBackupService,
+		asdbv1.AerospikeAppLabel:            v1beta1.AerospikeBackupServiceKey,
 		asdbv1.AerospikeCustomResourceLabel: clName,
 	}
 }
diff --git a/test/backup/backup_test.go b/test/backup/backup_test.go
index bef0b6af..d6d711de 100644
--- a/test/backup/backup_test.go
+++ b/test/backup/backup_test.go
@@ -10,7 +10,6 @@ import (
 	"sigs.k8s.io/yaml"
 
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
 )
 
 var _ = Describe(
@@ -58,9 +57,9 @@ var _ = Describe(
 
 				It("Should fail when un-supported field is given in backup config", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					routines := config[common.BackupRoutinesKey].(map[string]interface{})
+					routines := config[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 					routines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})["unknown"] = "unknown"
-					config[common.BackupRoutinesKey] = routines
+					config[asdbv1beta1.BackupRoutinesKey] = routines
 
 					configBytes, mErr := json.Marshal(config)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -73,9 +72,9 @@ var _ = Describe(
 
 				It("Should fail when more than 1 cluster is given in backup config", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					aeroCluster := config[common.AerospikeClusterKey].(map[string]interface{})
+					aeroCluster := config[asdbv1beta1.AerospikeClusterKey].(map[string]interface{})
 					aeroCluster["cluster-two"] = aeroCluster["test-cluster"]
-					config[common.AerospikeClusterKey] = aeroCluster
+					config[asdbv1beta1.AerospikeClusterKey] = aeroCluster
 
 					configBytes, mErr := json.Marshal(config)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -94,7 +93,7 @@ var _ = Describe(
 					backup.Spec.OnDemandBackups = []asdbv1beta1.OnDemandBackupSpec{
 						{
 							ID:          "on-demand",
-							RoutineName: "test-routine",
+							RoutineName: namePrefix(backupNsNm) + "-" + "test-routine",
 						},
 					}
 
@@ -159,10 +158,10 @@ var _ = Describe(
 
 				It("Should fail when non-existing policy is referred in Backup routine", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					routines := config[common.BackupRoutinesKey].(map[string]interface{})
+					routines := config[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 					routines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})["backup-policy"] =
 						"non-existing-policy"
-					config[common.BackupRoutinesKey] = routines
+					config[asdbv1beta1.BackupRoutinesKey] = routines
 
 					configBytes, mErr := json.Marshal(config)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -174,10 +173,10 @@ var _ = Describe(
 
 				It("Should fail when non-existing cluster is referred in Backup routine", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					routines := config[common.BackupRoutinesKey].(map[string]interface{})
-					routines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})[common.SourceClusterKey] =
+					routines := config[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
+					routines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})[asdbv1beta1.SourceClusterKey] =
 						"non-existing-cluster"
-					config[common.BackupRoutinesKey] = routines
+					config[asdbv1beta1.BackupRoutinesKey] = routines
 
 					configBytes, mErr := json.Marshal(config)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -189,10 +188,10 @@ var _ = Describe(
 
 				It("Should fail when non-existing storage is referred in Backup routine", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					routines := config[common.BackupRoutinesKey].(map[string]interface{})
+					routines := config[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 					routines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})["storage"] =
 						"non-existing-storage"
-					config[common.BackupRoutinesKey] = routines
+					config[asdbv1beta1.BackupRoutinesKey] = routines
 
 					configBytes, mErr := json.Marshal(config)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -211,7 +210,7 @@ var _ = Describe(
 
 				It("Should fail when service field is given in backup config", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					config[common.ServiceKey] = map[string]interface{}{
+					config[asdbv1beta1.ServiceKey] = map[string]interface{}{
 						"http": map[string]interface{}{
 							"port": 8081,
 						},
@@ -229,7 +228,7 @@ var _ = Describe(
 
 				It("Should fail when backup-policies field is given in backup config", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					config[common.BackupPoliciesKey] = map[string]interface{}{
+					config[asdbv1beta1.BackupPoliciesKey] = map[string]interface{}{
 						"test-policy": map[string]interface{}{
 							"parallel":     3,
 							"remove-files": "KeepAll",
@@ -249,10 +248,11 @@ var _ = Describe(
 
 				It("Should fail when storage field is given in backup config", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					config[common.StorageKey] = map[string]interface{}{
+					config[asdbv1beta1.StorageKey] = map[string]interface{}{
 						"local": map[string]interface{}{
-							"path": "/localStorage",
-							"type": "local",
+							"local-storage": map[string]interface{}{
+								"path": "/localStorage",
+							},
 						},
 					}
 
@@ -268,7 +268,7 @@ var _ = Describe(
 
 				It("Should fail when secret-agent is given in backup config", func() {
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					config[common.SecretAgentsKey] = map[string]interface{}{
+					config[asdbv1beta1.SecretAgentsKey] = map[string]interface{}{
 						"test-agent": map[string]interface{}{
 							"address": "localhost",
 							"port":    4000,
@@ -335,11 +335,11 @@ var _ = Describe(
 
 					// change storage to change overall backup config
 					config := getBackupConfigInMap(namePrefix(backupNsNm))
-					backupRoutines := config[common.BackupRoutinesKey].(map[string]interface{})
-					backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})[common.StorageKey] =
+					backupRoutines := config[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
+					backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})[asdbv1beta1.StorageKey] =
 						"s3Storage"
 
-					config[common.BackupRoutinesKey] = backupRoutines
+					config[asdbv1beta1.BackupRoutinesKey] = backupRoutines
 
 					configBytes, mErr := json.Marshal(config)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -368,11 +368,11 @@ var _ = Describe(
 
 			It("Should trigger backup when correct backup config with s3 storage is given", func() {
 				config := getBackupConfigInMap(namePrefix(backupNsNm))
-				backupRoutines := config[common.BackupRoutinesKey].(map[string]interface{})
-				backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})[common.StorageKey] =
+				backupRoutines := config[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
+				backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine"].(map[string]interface{})[asdbv1beta1.StorageKey] =
 					"s3Storage"
 
-				config[common.BackupRoutinesKey] = backupRoutines
+				config[asdbv1beta1.BackupRoutinesKey] = backupRoutines
 
 				configBytes, mErr := json.Marshal(config)
 				Expect(mErr).ToNot(HaveOccurred())
@@ -405,24 +405,24 @@ var _ = Describe(
 				Expect(err).ToNot(HaveOccurred())
 
 				// Add a routine to the configMap
-				data := cm.Data[common.BackupServiceConfigYAML]
+				data := cm.Data[asdbv1beta1.BackupServiceConfigYAML]
 				backupSvcConfig := make(map[string]interface{})
 
 				err = yaml.Unmarshal([]byte(data), &backupSvcConfig)
 				Expect(err).ToNot(HaveOccurred())
 
-				backupRoutines := backupSvcConfig[common.BackupRoutinesKey].(map[string]interface{})
+				backupRoutines := backupSvcConfig[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 				// Add a new routine with a different name
 				newRoutineName := namePrefix(backupNsNm) + "-" + "test-routine1"
 				backupRoutines[newRoutineName] =
 					backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine"]
 
-				backupSvcConfig[common.BackupRoutinesKey] = backupRoutines
+				backupSvcConfig[asdbv1beta1.BackupRoutinesKey] = backupRoutines
 
 				newData, mErr := yaml.Marshal(backupSvcConfig)
 				Expect(mErr).ToNot(HaveOccurred())
 
-				cm.Data[common.BackupServiceConfigYAML] = string(newData)
+				cm.Data[asdbv1beta1.BackupServiceConfigYAML] = string(newData)
 
 				err = k8sClient.Update(testCtx, &cm)
 				Expect(err).ToNot(HaveOccurred())
@@ -447,13 +447,13 @@ var _ = Describe(
 					&cm)
 				Expect(err).ToNot(HaveOccurred())
 
-				data = cm.Data[common.BackupServiceConfigYAML]
+				data = cm.Data[asdbv1beta1.BackupServiceConfigYAML]
 				backupSvcConfig = make(map[string]interface{})
 
 				err = yaml.Unmarshal([]byte(data), &backupSvcConfig)
 				Expect(err).ToNot(HaveOccurred())
 
-				backupRoutines = backupSvcConfig[common.BackupRoutinesKey].(map[string]interface{})
+				backupRoutines = backupSvcConfig[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 				_, ok := backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine1"]
 				Expect(ok).To(BeFalse())
 			})
@@ -483,7 +483,7 @@ var _ = Describe(
 
 			It("Should unregister backup-routines when removed from backup CR", func() {
 				backupConfig := getBackupConfigInMap(namePrefix(backupNsNm))
-				backupRoutines := backupConfig[common.BackupRoutinesKey].(map[string]interface{})
+				backupRoutines := backupConfig[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 				backupRoutines[namePrefix(backupNsNm)+"-"+"test-routine1"] = map[string]interface{}{
 					"backup-policy":      "test-policy1",
 					"interval-cron":      "@daily",
@@ -493,7 +493,7 @@ var _ = Describe(
 					"storage":            "local",
 				}
 
-				backupConfig[common.BackupRoutinesKey] = backupRoutines
+				backupConfig[asdbv1beta1.BackupRoutinesKey] = backupRoutines
 
 				configBytes, err := json.Marshal(backupConfig)
 				Expect(err).ToNot(HaveOccurred())
diff --git a/test/backup/test_utils.go b/test/backup/test_utils.go
index 6efd74d6..886d4d6d 100644
--- a/test/backup/test_utils.go
+++ b/test/backup/test_utils.go
@@ -17,10 +17,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 
-	"github.com/aerospike/aerospike-backup-service/pkg/model"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
 	backup_service "github.com/aerospike/aerospike-kubernetes-operator/pkg/backup-service"
+	backupservice "github.com/aerospike/aerospike-kubernetes-operator/test/backup_service"
 )
 
 const (
@@ -95,7 +95,7 @@ func getBackupConfBytes(prefix string) ([]byte, error) {
 
 func getBackupConfigInMap(prefix string) map[string]interface{} {
 	return map[string]interface{}{
-		common.AerospikeClusterKey: map[string]interface{}{
+		asdbv1beta1.AerospikeClusterKey: map[string]interface{}{
 			fmt.Sprintf("%s-%s", prefix, "test-cluster"): map[string]interface{}{
 				"credentials": map[string]interface{}{
 					"password": "admin123",
@@ -111,7 +111,7 @@ func getBackupConfigInMap(prefix string) map[string]interface{} {
 				},
 			},
 		},
-		common.BackupRoutinesKey: map[string]interface{}{
+		asdbv1beta1.BackupRoutinesKey: map[string]interface{}{
 			fmt.Sprintf("%s-%s", prefix, "test-routine"): map[string]interface{}{
 				"backup-policy":      "test-policy",
 				"interval-cron":      "@daily",
@@ -128,8 +128,8 @@ func getWrongBackupConfBytes(prefix string) ([]byte, error) {
 	backupConfig := getBackupConfigInMap(prefix)
 
 	// change the format from map to list
-	backupConfig[common.BackupRoutinesKey] = []interface{}{
-		backupConfig[common.BackupRoutinesKey],
+	backupConfig[asdbv1beta1.BackupRoutinesKey] = []interface{}{
+		backupConfig[asdbv1beta1.BackupRoutinesKey],
 	}
 
 	configBytes, err := json.Marshal(backupConfig)
@@ -220,29 +220,27 @@ func waitForBackup(cl client.Client, backup *asdbv1beta1.AerospikeBackup,
 
 // validateTriggeredBackup validates if the backup is triggered by checking the current config of backup-service
 func validateTriggeredBackup(k8sClient client.Client, backup *asdbv1beta1.AerospikeBackup) error {
-	var backupK8sService corev1.Service
-
 	validateNewEntries := func(currentConfigInMap map[string]interface{}, desiredConfigInMap map[string]interface{},
 		fieldPath string) error {
-		newCluster := desiredConfigInMap[common.AerospikeClusterKey].(map[string]interface{})
+		newCluster := desiredConfigInMap[asdbv1beta1.AerospikeClusterKey].(map[string]interface{})
 
 		for clusterName := range newCluster {
-			if _, ok := currentConfigInMap[common.AerospikeClustersKey].(map[string]interface{})[clusterName]; !ok {
+			if _, ok := currentConfigInMap[asdbv1beta1.AerospikeClustersKey].(map[string]interface{})[clusterName]; !ok {
 				return fmt.Errorf("cluster %s not found in %s backup config", clusterName, fieldPath)
 			}
 		}
 
 		pkgLog.Info(fmt.Sprintf("Cluster info is found in %s backup config", fieldPath))
 
-		routines := desiredConfigInMap[common.BackupRoutinesKey].(map[string]interface{})
+		routines := desiredConfigInMap[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})
 
 		for routineName := range routines {
-			if _, ok := currentConfigInMap[common.BackupRoutinesKey].(map[string]interface{})[routineName]; !ok {
+			if _, ok := currentConfigInMap[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})[routineName]; !ok {
 				return fmt.Errorf("routine %s not found in %s backup config", routineName, fieldPath)
 			}
 		}
 
-		if len(routines) != len(currentConfigInMap[common.BackupRoutinesKey].(map[string]interface{})) {
+		if len(routines) != len(currentConfigInMap[asdbv1beta1.BackupRoutinesKey].(map[string]interface{})) {
 			return fmt.Errorf("backup routine count mismatch in %s backup config", fieldPath)
 		}
 
@@ -264,7 +262,7 @@ func validateTriggeredBackup(k8sClient client.Client, backup *asdbv1beta1.Aerosp
 
 	backupSvcConfig := make(map[string]interface{})
 
-	if err := yaml.Unmarshal([]byte(configmap.Data[common.BackupServiceConfigYAML]), &backupSvcConfig); err != nil {
+	if err := yaml.Unmarshal([]byte(configmap.Data[asdbv1beta1.BackupServiceConfigYAML]), &backupSvcConfig); err != nil {
 		return err
 	}
 
@@ -278,48 +276,13 @@ func validateTriggeredBackup(k8sClient client.Client, backup *asdbv1beta1.Aerosp
 		return err
 	}
 
-	// Wait for Service LB IP to be populated
-	if err := wait.PollUntilContextTimeout(testCtx, interval, timeout, true,
-		func(ctx context.Context) (bool, error) {
-			if err := k8sClient.Get(ctx,
-				types.NamespacedName{
-					Name:      backup.Spec.BackupService.Name,
-					Namespace: backup.Spec.BackupService.Namespace,
-				},
-				&backupK8sService); err != nil {
-				return false, err
-			}
-
-			if backupK8sService.Status.LoadBalancer.Ingress == nil {
-				return false, nil
-			}
-
-			return true, nil
-		}); err != nil {
-		return err
-	}
-
-	serviceClient := backup_service.Client{
-		Address: backupK8sService.Status.LoadBalancer.Ingress[0].IP,
-		Port:    8081,
-	}
-
-	// Wait for Backup service to be ready
-	if err := wait.PollUntilContextTimeout(testCtx, interval, timeout, true,
-		func(_ context.Context) (bool, error) {
-			config, err := serviceClient.GetBackupServiceConfig()
-			if err != nil {
-				pkgLog.Error(err, "Failed to get backup service config")
-				return false, nil
-			}
-
-			backupSvcConfig = config
-			return true, nil
-		}); err != nil {
+	config, err := backupservice.GetAPIBackupSvcConfig(k8sClient, backup.Spec.BackupService.Name,
+		backup.Spec.BackupService.Namespace)
+	if err != nil {
 		return err
 	}
 
-	return validateNewEntries(backupSvcConfig, desiredConfigInMap, "backup-service API")
+	return validateNewEntries(config, desiredConfigInMap, "backup-service API")
 }
 
 func namePrefix(nsNm types.NamespacedName) string {
@@ -352,7 +315,7 @@ func GetBackupDataPaths(k8sClient client.Client, backup *asdbv1beta1.AerospikeBa
 	}
 
 	var (
-		config          model.Config
+		config          dto.Config
 		backupDataPaths []string
 	)
 
diff --git a/test/backup_service/backup_service_test.go b/test/backup_service/backup_service_test.go
index 3719986d..5081c923 100644
--- a/test/backup_service/backup_service_test.go
+++ b/test/backup_service/backup_service_test.go
@@ -2,7 +2,6 @@ package backupservice
 
 import (
 	"encoding/json"
-	"net/http"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -11,7 +10,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
 )
 
 var _ = Describe(
@@ -60,6 +58,34 @@ var _ = Describe(
 					Expect(err).To(HaveOccurred())
 				})
 
+				It("Should fail when backup service version is less than 3.0.0", func() {
+					backupService, err = NewBackupService()
+					Expect(err).ToNot(HaveOccurred())
+
+					backupService.Spec.Image = BackupServiceVersion2Image
+
+					err = deployBackupServiceWithTO(k8sClient, backupService, 1*time.Minute)
+					Expect(err).To(HaveOccurred())
+					Expect(err.Error()).To(ContainSubstring("Minimum supported version is 3.0.0"))
+				})
+
+				It("Should fail when backup service image version is downgraded to 2.0", func() {
+					backupService, err = NewBackupService()
+					Expect(err).ToNot(HaveOccurred())
+					err = DeployBackupService(k8sClient, backupService)
+					Expect(err).ToNot(HaveOccurred())
+
+					By("Downgrading image version to 2.0")
+					backupService, err = getBackupServiceObj(k8sClient, name, namespace)
+					Expect(err).ToNot(HaveOccurred())
+
+					backupService.Spec.Image = BackupServiceVersion2Image
+
+					err = deployBackupServiceWithTO(k8sClient, backupService, 1*time.Minute)
+					Expect(err).To(HaveOccurred())
+					Expect(err.Error()).To(ContainSubstring("Minimum supported version is 3.0.0"))
+				})
+
 				It("Should fail when duplicate volume names are given in secrets", func() {
 					backupService, err = NewBackupService()
 					Expect(err).ToNot(HaveOccurred())
@@ -73,7 +99,7 @@ var _ = Describe(
 
 				It("Should fail when aerospike-clusters field is given", func() {
 					configMap := getBackupServiceConfMap()
-					configMap[common.AerospikeClustersKey] = map[string]interface{}{
+					configMap[asdbv1beta1.AerospikeClustersKey] = map[string]interface{}{
 						"test-cluster": map[string]interface{}{
 							"credentials": map[string]interface{}{
 								"password": "admin123",
@@ -101,7 +127,7 @@ var _ = Describe(
 
 				It("Should fail when backup-routines field is given", func() {
 					configMap := getBackupServiceConfMap()
-					configMap[common.BackupRoutinesKey] = map[string]interface{}{
+					configMap[asdbv1beta1.BackupRoutinesKey] = map[string]interface{}{
 						"test-routine": map[string]interface{}{
 							"backup-policy":      "test-policy",
 							"interval-cron":      "@daily",
@@ -133,7 +159,8 @@ var _ = Describe(
 				Expect(err).ToNot(HaveOccurred())
 			})
 
-			It("Should restart backup service deployment pod when config is changed", func() {
+			It("Should restart backup service deployment pod when static fields are changed in backup service "+
+				"config", func() {
 				backupService, err = NewBackupService()
 				Expect(err).ToNot(HaveOccurred())
 				err = DeployBackupService(k8sClient, backupService)
@@ -149,7 +176,8 @@ var _ = Describe(
 				backupService, err = getBackupServiceObj(k8sClient, name, namespace)
 				Expect(err).ToNot(HaveOccurred())
 
-				// Change config
+				By("Change static fields")
+				// Changing static field 'port' to 8080 and removing dynamic fields like backup policies and storage
 				backupService.Spec.Config.Raw = []byte(`{"service":{"http":{"port":8080}}}`)
 				err = updateBackupService(k8sClient, backupService)
 				Expect(err).ToNot(HaveOccurred())
@@ -161,6 +189,42 @@ var _ = Describe(
 				Expect(podList.Items[0].ObjectMeta.UID).ToNot(Equal(PodUID))
 			})
 
+			It("Should do hot-reload when dynamic fields are changed in backup service config", func() {
+				backupService, err = NewBackupService()
+				backupService.Spec.Service = &asdbv1beta1.Service{Type: corev1.ServiceTypeLoadBalancer}
+				Expect(err).ToNot(HaveOccurred())
+				err = DeployBackupService(k8sClient, backupService)
+				Expect(err).ToNot(HaveOccurred())
+
+				podList, gErr := getBackupServicePodList(k8sClient, backupService)
+				Expect(gErr).ToNot(HaveOccurred())
+				Expect(len(podList.Items)).To(Equal(1))
+
+				PodUID := podList.Items[0].ObjectMeta.UID
+
+				// Get backup service object
+				backupService, err = getBackupServiceObj(k8sClient, name, namespace)
+				Expect(err).ToNot(HaveOccurred())
+
+				By("Change dynamic fields")
+				// Keeping static field 'port' same and removing dynamic fields like backup policies and storage
+				backupService.Spec.Config.Raw = []byte(`{"service":{"http":{"port":8081}}}`)
+				err = updateBackupService(k8sClient, backupService)
+				Expect(err).ToNot(HaveOccurred())
+
+				podList, err = getBackupServicePodList(k8sClient, backupService)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(len(podList.Items)).To(Equal(1))
+
+				Expect(podList.Items[0].ObjectMeta.UID).To(Equal(PodUID))
+
+				config, gErr := GetAPIBackupSvcConfig(k8sClient, backupService.Name, backupService.Namespace)
+				Expect(gErr).ToNot(HaveOccurred())
+				Expect(config).ToNot(BeNil())
+				Expect(config[asdbv1beta1.BackupRoutinesKey]).To(BeNil())
+				Expect(config[asdbv1beta1.StorageKey]).To(BeNil())
+			})
+
 			It("Should restart backup service deployment pod when pod spec is changed", func() {
 				backupService, err = NewBackupService()
 				Expect(err).ToNot(HaveOccurred())
@@ -180,7 +244,7 @@ var _ = Describe(
 				// Change Pod spec
 				backupService.Spec.Resources = &corev1.ResourceRequirements{
 					Limits: corev1.ResourceList{
-						corev1.ResourceCPU: resource.MustParse("0.5"),
+						corev1.ResourceCPU: resource.MustParse("0.2"),
 					},
 				}
 
@@ -218,29 +282,9 @@ var _ = Describe(
 				Expect(err).ToNot(HaveOccurred())
 				Expect(svc.Spec.Type).To(Equal(corev1.ServiceTypeLoadBalancer))
 
-				Eventually(func() bool {
-					svc, err = getBackupK8sServiceObj(k8sClient, name, namespace)
-					if err != nil {
-						return false
-					}
-					return svc.Status.LoadBalancer.Ingress != nil
-				}, timeout, interval).Should(BeTrue())
-
-				// Check backup service health using LB IP
-				Eventually(func() bool {
-					resp, err := http.Get("http://" + svc.Status.LoadBalancer.Ingress[0].IP + ":8081/health")
-					if err != nil {
-						pkgLog.Error(err, "Failed to get health")
-						return false
-					}
-
-					defer resp.Body.Close()
-
-					return resp.StatusCode == http.StatusOK
-				}, timeout, interval).Should(BeTrue())
-
+				_, err = GetAPIBackupSvcConfig(k8sClient, backupService.Name, backupService.Namespace)
+				Expect(err).ToNot(HaveOccurred())
 			})
-
 		})
 	},
 )
diff --git a/test/backup_service/test_utils.go b/test/backup_service/test_utils.go
index a4a75884..cd36d116 100644
--- a/test/backup_service/test_utils.go
+++ b/test/backup_service/test_utils.go
@@ -18,12 +18,13 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
+	backup_service "github.com/aerospike/aerospike-kubernetes-operator/pkg/backup-service"
 	"github.com/aerospike/aerospike-kubernetes-operator/pkg/utils"
 	"github.com/aerospike/aerospike-kubernetes-operator/test"
 )
 
-const BackupServiceImage = "aerospike/aerospike-backup-service:2.0.0"
+const BackupServiceImage = "abhishekdwivedi3060/aerospike-backup-service:3.0.0.1"
+const BackupServiceVersion2Image = "aerospike/aerospike-backup-service:2.0.0"
 
 const (
 	timeout   = 2 * time.Minute
@@ -201,14 +202,14 @@ func getBackupServiceConfBytes() ([]byte, error) {
 func getWrongBackupServiceConfBytes() ([]byte, error) {
 	config := getBackupServiceConfMap()
 
-	tempList := make([]interface{}, 0, len(config[common.BackupPoliciesKey].(map[string]interface{})))
+	tempList := make([]interface{}, 0, len(config[asdbv1beta1.BackupPoliciesKey].(map[string]interface{})))
 
-	for _, policy := range config[common.BackupPoliciesKey].(map[string]interface{}) {
+	for _, policy := range config[asdbv1beta1.BackupPoliciesKey].(map[string]interface{}) {
 		tempList = append(tempList, policy)
 	}
 
 	// change the format from map to list
-	config[common.BackupPoliciesKey] = tempList
+	config[asdbv1beta1.BackupPoliciesKey] = tempList
 
 	configBytes, err := json.Marshal(config)
 	if err != nil {
@@ -222,12 +223,12 @@ func getWrongBackupServiceConfBytes() ([]byte, error) {
 
 func getBackupServiceConfMap() map[string]interface{} {
 	return map[string]interface{}{
-		common.ServiceKey: map[string]interface{}{
+		asdbv1beta1.ServiceKey: map[string]interface{}{
 			"http": map[string]interface{}{
 				"port": 8081,
 			},
 		},
-		common.BackupPoliciesKey: map[string]interface{}{
+		asdbv1beta1.BackupPoliciesKey: map[string]interface{}{
 			"test-policy": map[string]interface{}{
 				"parallel":     3,
 				"remove-files": "KeepAll",
@@ -237,17 +238,19 @@ func getBackupServiceConfMap() map[string]interface{} {
 				"remove-files": "KeepAll",
 			},
 		},
-		common.StorageKey: map[string]interface{}{
+		asdbv1beta1.StorageKey: map[string]interface{}{
 			"local": map[string]interface{}{
-				"path": "/localStorage",
-				"type": "local",
+				"local-storage": map[string]interface{}{
+					"path": "/localStorage",
+				},
 			},
 			"s3Storage": map[string]interface{}{
-				"type":                 "aws-s3",
-				"path":                 "s3://aerospike-kubernetes-operator-test",
-				"s3-region":            "us-east-1",
-				"s3-endpoint-override": "",
-				"s3-profile":           "default",
+				"s3-storage": map[string]interface{}{
+					"bucket":     "aerospike-kubernetes-operator-test",
+					"path":       "/",
+					"s3-region":  "us-east-1",
+					"s3-profile": "default",
+				},
 			},
 		},
 	}
@@ -298,3 +301,53 @@ func DeleteBackupService(
 
 	return nil
 }
+
+func GetAPIBackupSvcConfig(k8sClient client.Client, backupServiceName, backupServiceNamespace string,
+) (map[string]interface{}, error) {
+	var backupK8sService corev1.Service
+
+	// Wait for Service LB IP to be populated
+	if err := wait.PollUntilContextTimeout(testCtx, interval, timeout, true,
+		func(ctx context.Context) (bool, error) {
+			if err := k8sClient.Get(ctx,
+				types.NamespacedName{
+					Name:      backupServiceName,
+					Namespace: backupServiceNamespace,
+				},
+				&backupK8sService); err != nil {
+				return false, err
+			}
+
+			if backupK8sService.Status.LoadBalancer.Ingress == nil {
+				return false, nil
+			}
+
+			return true, nil
+		}); err != nil {
+		return nil, err
+	}
+
+	serviceClient := backup_service.Client{
+		Address: backupK8sService.Status.LoadBalancer.Ingress[0].IP,
+		Port:    8081,
+	}
+
+	backupSvcConfig := make(map[string]interface{})
+
+	// Wait for Backup service to be ready
+	if err := wait.PollUntilContextTimeout(testCtx, interval, timeout, true,
+		func(_ context.Context) (bool, error) {
+			config, err := serviceClient.GetBackupServiceConfig()
+			if err != nil {
+				pkgLog.Error(err, "Failed to get backup service config")
+				return false, nil
+			}
+
+			backupSvcConfig = config
+			return true, nil
+		}); err != nil {
+		return nil, err
+	}
+
+	return backupSvcConfig, nil
+}
diff --git a/test/cluster/cluster_helper.go b/test/cluster/cluster_helper.go
index 13d13c4d..2ebdb2d1 100644
--- a/test/cluster/cluster_helper.go
+++ b/test/cluster/cluster_helper.go
@@ -23,6 +23,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	as "github.com/aerospike/aerospike-client-go/v7"
 	asdbv1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1"
 	internalerrors "github.com/aerospike/aerospike-kubernetes-operator/errors"
 	"github.com/aerospike/aerospike-kubernetes-operator/pkg/utils"
@@ -55,6 +56,14 @@ const aerospikeConfigSecret string = "aerospike-config-secret" //nolint:gosec //
 const serviceTLSPort = 4333
 const serviceNonTLSPort = 3000
 
+// constants for writing data to aerospike
+const (
+	setName  = "test"
+	key      = "key1"
+	binName  = "testBin"
+	binValue = "binValue"
+)
+
 var aerospikeVolumeInitMethodDeleteFiles = asdbv1.AerospikeVolumeMethodDeleteFiles
 
 var (
@@ -680,6 +689,22 @@ func getCluster(
 	return aeroCluster, nil
 }
 
+// GetCluster is the public variant of getCluster
+// Remove this when getCluster will be made public
+func GetCluster(
+	k8sClient client.Client, ctx goctx.Context,
+	clusterNamespacedName types.NamespacedName,
+) (*asdbv1.AerospikeCluster, error) {
+	aeroCluster := &asdbv1.AerospikeCluster{}
+
+	err := k8sClient.Get(ctx, clusterNamespacedName, aeroCluster)
+	if err != nil {
+		return nil, err
+	}
+
+	return aeroCluster, nil
+}
+
 func getClusterIfExists(
 	k8sClient client.Client, ctx goctx.Context,
 	clusterNamespacedName types.NamespacedName,
@@ -1698,3 +1723,92 @@ func getAeroClusterPVCList(
 
 	return pvcList.Items, nil
 }
+
+func WriteDataToCluster(
+	aeroCluster *asdbv1.AerospikeCluster,
+	k8sClient client.Client,
+	namespaces []string,
+) error {
+	asClient, err := getAerospikeClient(aeroCluster, k8sClient)
+	if err != nil {
+		return err
+	}
+
+	defer asClient.Close()
+
+	pkgLog.Info(
+		"Loading record", "nodes", asClient.GetNodeNames(),
+	)
+
+	wp := as.NewWritePolicy(0, 0)
+
+	for _, ns := range namespaces {
+		newKey, err := as.NewKey(ns, setName, key)
+		if err != nil {
+			return err
+		}
+
+		if err := asClient.Put(
+			wp, newKey, as.BinMap{
+				binName: binValue,
+			},
+		); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func CheckDataInCluster(
+	aeroCluster *asdbv1.AerospikeCluster,
+	k8sClient client.Client,
+	namespaces []string,
+) (map[string]bool, error) {
+	data := make(map[string]bool)
+
+	asClient, err := getAerospikeClient(aeroCluster, k8sClient)
+	if err != nil {
+		return nil, err
+	}
+
+	defer asClient.Close()
+
+	pkgLog.Info(
+		"Loading record", "nodes", asClient.GetNodeNames(),
+	)
+
+	for _, ns := range namespaces {
+		newKey, err := as.NewKey(ns, setName, key)
+		if err != nil {
+			return nil, err
+		}
+
+		record, err := asClient.Get(nil, newKey)
+		if err != nil {
+			return nil, nil
+		}
+
+		if bin, exists := record.Bins[binName]; exists {
+			value, ok := bin.(string)
+
+			if !ok {
+				return nil, fmt.Errorf(
+					"Bin-Name: %s - conversion to bin value failed", binName,
+				)
+			}
+
+			if value == binValue {
+				data[ns] = true
+			} else {
+				return nil, fmt.Errorf(
+					"bin: %s exsists but the value is changed", binName,
+				)
+			}
+		} else {
+			data[ns] = false
+		}
+	}
+
+	return data, nil
+}
diff --git a/test/cluster/sample_files_test.go b/test/cluster/sample_files_test.go
index 6e718340..52648587 100644
--- a/test/cluster/sample_files_test.go
+++ b/test/cluster/sample_files_test.go
@@ -65,7 +65,7 @@ var _ = Describe("Sample files validation", func() {
 			})
 			Expect(err).NotTo(HaveOccurred())
 
-			err = writeDataToCluster(
+			err = WriteDataToCluster(
 				aeroCluster, k8sClient, []string{"test"},
 			)
 			Expect(err).NotTo(HaveOccurred())
@@ -77,7 +77,7 @@ var _ = Describe("Sample files validation", func() {
 			})
 			Expect(err).NotTo(HaveOccurred())
 
-			records, err := checkDataInCluster(
+			records, err := CheckDataInCluster(
 				destCluster, k8sClient, []string{"test"},
 			)
 			Expect(err).ToNot(HaveOccurred())
diff --git a/test/cluster/storage_wipe_test.go b/test/cluster/storage_wipe_test.go
index 7aae50c4..99586293 100644
--- a/test/cluster/storage_wipe_test.go
+++ b/test/cluster/storage_wipe_test.go
@@ -11,20 +11,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/ptr"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	as "github.com/aerospike/aerospike-client-go/v7"
 	asdbv1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1"
 	"github.com/aerospike/aerospike-kubernetes-operator/test"
 )
 
-const (
-	setName  = "test"
-	key      = "key1"
-	binName  = "testBin"
-	binValue = "binValue"
-)
-
 var (
 	namespaces = []string{"test", "test1"}
 )
@@ -99,7 +90,7 @@ var _ = Describe(
 						Expect(err).ToNot(HaveOccurred())
 
 						By("Writing some data to the cluster")
-						err = writeDataToCluster(
+						err = WriteDataToCluster(
 							aeroCluster, k8sClient, namespaces,
 						)
 						Expect(err).ToNot(HaveOccurred())
@@ -135,7 +126,7 @@ var _ = Describe(
 						Expect(err).ToNot(HaveOccurred())
 
 						By("Checking - cluster data should not be wiped")
-						records, err := checkDataInCluster(
+						records, err := CheckDataInCluster(
 							aeroCluster, k8sClient, namespaces,
 						)
 						Expect(err).ToNot(HaveOccurred())
@@ -180,7 +171,7 @@ var _ = Describe(
 						Expect(err).ToNot(HaveOccurred())
 
 						By("Checking - cluster data should be wiped")
-						records, err = checkDataInCluster(
+						records, err = CheckDataInCluster(
 							aeroCluster, k8sClient, namespaces,
 						)
 						Expect(err).ToNot(HaveOccurred())
@@ -206,95 +197,6 @@ var _ = Describe(
 	},
 )
 
-func writeDataToCluster(
-	aeroCluster *asdbv1.AerospikeCluster,
-	k8sClient client.Client,
-	namespaces []string,
-) error {
-	asClient, err := getAerospikeClient(aeroCluster, k8sClient)
-	if err != nil {
-		return err
-	}
-
-	defer asClient.Close()
-
-	pkgLog.Info(
-		"Loading record", "nodes", asClient.GetNodeNames(),
-	)
-
-	wp := as.NewWritePolicy(0, 0)
-
-	for _, ns := range namespaces {
-		newKey, err := as.NewKey(ns, setName, key)
-		if err != nil {
-			return err
-		}
-
-		if err := asClient.Put(
-			wp, newKey, as.BinMap{
-				binName: binValue,
-			},
-		); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-func checkDataInCluster(
-	aeroCluster *asdbv1.AerospikeCluster,
-	k8sClient client.Client,
-	namespaces []string,
-) (map[string]bool, error) {
-	data := make(map[string]bool)
-
-	asClient, err := getAerospikeClient(aeroCluster, k8sClient)
-	if err != nil {
-		return nil, err
-	}
-
-	defer asClient.Close()
-
-	pkgLog.Info(
-		"Loading record", "nodes", asClient.GetNodeNames(),
-	)
-
-	for _, ns := range namespaces {
-		newKey, err := as.NewKey(ns, setName, key)
-		if err != nil {
-			return nil, err
-		}
-
-		record, err := asClient.Get(nil, newKey)
-		if err != nil {
-			return nil, nil
-		}
-
-		if bin, exists := record.Bins[binName]; exists {
-			value, ok := bin.(string)
-
-			if !ok {
-				return nil, fmt.Errorf(
-					"Bin-Name: %s - conversion to bin value failed", binName,
-				)
-			}
-
-			if value == binValue {
-				data[ns] = true
-			} else {
-				return nil, fmt.Errorf(
-					"bin: %s exsists but the value is changed", binName,
-				)
-			}
-		} else {
-			data[ns] = false
-		}
-	}
-
-	return data, nil
-}
-
 func getAerospikeClusterConfig() *asdbv1.AerospikeConfigSpec {
 	return &asdbv1.AerospikeConfigSpec{
 		Value: map[string]interface{}{
diff --git a/test/cluster_prereq.go b/test/cluster_prereq.go
index 692fc920..a801bd05 100644
--- a/test/cluster_prereq.go
+++ b/test/cluster_prereq.go
@@ -12,7 +12,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
+	"github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
 )
 
 const (
@@ -289,7 +289,7 @@ func createAuthSecret(
 
 func setupBackupServicePreReq(k8sClient client.Client, ctx goctx.Context, namespace string) error {
 	// Create SA for aerospike backup service
-	if err := createServiceAccount(k8sClient, goctx.TODO(), common.AerospikeBackupService, namespace); err != nil {
+	if err := createServiceAccount(k8sClient, goctx.TODO(), v1beta1.AerospikeBackupServiceKey, namespace); err != nil {
 		return err
 	}
 
diff --git a/test/restore/restore_suite_test.go b/test/restore/restore_suite_test.go
index e6b52cca..afc0cf7d 100644
--- a/test/restore/restore_suite_test.go
+++ b/test/restore/restore_suite_test.go
@@ -69,6 +69,14 @@ var _ = BeforeSuite(
 		err = cluster.DeployCluster(k8sClient, testCtx, aeroCluster)
 		Expect(err).ToNot(HaveOccurred())
 
+		aeroCluster, err = cluster.GetCluster(k8sClient, testCtx, sourceAerospikeClusterNsNm)
+		Expect(err).ToNot(HaveOccurred())
+
+		err = cluster.WriteDataToCluster(
+			aeroCluster, k8sClient, []string{"test"},
+		)
+		Expect(err).NotTo(HaveOccurred())
+
 		backupObj, err := backup.NewBackup(backupNsNm)
 		Expect(err).ToNot(HaveOccurred())
 
@@ -85,7 +93,7 @@ var _ = BeforeSuite(
 		pkgLog.Info(fmt.Sprintf("BackupDataPaths: %v", backupDataPaths))
 		Expect(backupDataPaths).ToNot(BeEmpty())
 
-		// Example backupDataPath = "/localStorage/test-sample-backup-test-routine/backup/1722353745635/data/test"
+		// Example backupDataPath = "test-sample-backup-test-routine/backup/1722353745635/data/test"
 		backupDataPath = backupDataPaths[0]
 
 		By(fmt.Sprintf("Deploy destination Aerospike Cluster: %s", destinationAerospikeClusterNsNm.String()))
diff --git a/test/restore/restore_test.go b/test/restore/restore_test.go
index ded2f17e..de82a4d0 100644
--- a/test/restore/restore_test.go
+++ b/test/restore/restore_test.go
@@ -11,7 +11,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
-	"github.com/aerospike/aerospike-kubernetes-operator/internal/controller/common"
 )
 
 var _ = Describe(
@@ -91,7 +90,8 @@ var _ = Describe(
 				It("Should fail when routine/time is not given for Timestamp restore type", func() {
 					// getRestoreConfigInMap returns restore config without a routine, time and with source type
 					restoreConfig := getRestoreConfigInMap(backupDataPath)
-					delete(restoreConfig, common.SourceKey)
+					delete(restoreConfig, asdbv1beta1.SourceKey)
+					delete(restoreConfig, asdbv1beta1.BackupDataPathKey)
 
 					configBytes, mErr := json.Marshal(restoreConfig)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -114,7 +114,7 @@ var _ = Describe(
 
 				It("Should fail when routine field is given for Full/Incremental restore type", func() {
 					restoreConfig := getRestoreConfigInMap(backupDataPath)
-					restoreConfig[common.RoutineKey] = "test-routine"
+					restoreConfig[asdbv1beta1.RoutineKey] = "test-routine"
 
 					configBytes, mErr := json.Marshal(restoreConfig)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -128,7 +128,7 @@ var _ = Describe(
 
 				It("Should fail when time field is given for Full/Incremental restore type", func() {
 					restoreConfig := getRestoreConfigInMap(backupDataPath)
-					restoreConfig[common.TimeKey] = 1722408895094
+					restoreConfig[asdbv1beta1.TimeKey] = 1722408895094
 
 					configBytes, mErr := json.Marshal(restoreConfig)
 					Expect(mErr).ToNot(HaveOccurred())
@@ -150,6 +150,10 @@ var _ = Describe(
 
 						err = createRestore(k8sClient, restore)
 						Expect(err).ToNot(HaveOccurred())
+
+						err = validateRestoredData(k8sClient)
+						Expect(err).ToNot(HaveOccurred())
+
 					},
 				)
 
@@ -160,23 +164,27 @@ var _ = Describe(
 
 						err = createRestore(k8sClient, restore)
 						Expect(err).ToNot(HaveOccurred())
+
+						err = validateRestoredData(k8sClient)
+						Expect(err).ToNot(HaveOccurred())
+
 					},
 				)
 
 				It(
 					"Should complete restore for Timestamp restore type", func() {
 						restoreConfig := getRestoreConfigInMap(backupDataPath)
-						delete(restoreConfig, common.SourceKey)
+						delete(restoreConfig, asdbv1beta1.SourceKey)
+						delete(restoreConfig, asdbv1beta1.BackupDataPathKey)
 
 						parts := strings.Split(backupDataPath, "/")
-
 						time := parts[len(parts)-3]
 						timeInt, err := strconv.Atoi(time)
 						Expect(err).ToNot(HaveOccurred())
 
 						// increase time by 1 millisecond to consider the latest backup under time bound
-						restoreConfig[common.TimeKey] = int64(timeInt) + 1
-						restoreConfig[common.RoutineKey] = parts[len(parts)-5]
+						restoreConfig[asdbv1beta1.TimeKey] = int64(timeInt) + 1
+						restoreConfig[asdbv1beta1.RoutineKey] = parts[len(parts)-5]
 
 						configBytes, err := json.Marshal(restoreConfig)
 						Expect(err).ToNot(HaveOccurred())
@@ -185,6 +193,10 @@ var _ = Describe(
 
 						err = createRestore(k8sClient, restore)
 						Expect(err).ToNot(HaveOccurred())
+
+						err = validateRestoredData(k8sClient)
+						Expect(err).ToNot(HaveOccurred())
+
 					},
 				)
 			})
diff --git a/test/restore/test_utils.go b/test/restore/test_utils.go
index 1dc5ca04..3f3c63df 100644
--- a/test/restore/test_utils.go
+++ b/test/restore/test_utils.go
@@ -14,8 +14,9 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/aerospike/aerospike-backup-service/pkg/model"
+	"github.com/aerospike/aerospike-backup-service/v2/pkg/dto"
 	asdbv1beta1 "github.com/aerospike/aerospike-kubernetes-operator/api/v1beta1"
+	"github.com/aerospike/aerospike-kubernetes-operator/test/cluster"
 )
 
 const (
@@ -176,13 +177,13 @@ func waitForRestore(cl client.Client, restore *asdbv1beta1.AerospikeRestore,
 		return fmt.Errorf("restore result is not set")
 	}
 
-	var restoreResult model.RestoreJobStatus
+	var restoreResult dto.RestoreJobStatus
 
 	if err := json.Unmarshal(restore.Status.RestoreResult.Raw, &restoreResult); err != nil {
 		return err
 	}
 
-	if restoreResult.Status != model.JobStatusDone {
+	if restoreResult.Status != dto.JobStatusDone {
 		return fmt.Errorf("restore job status is not done")
 	}
 
@@ -229,8 +230,30 @@ func getRestoreConfigInMap(backupPath string) map[string]interface{} {
 			"no-indexes":    true,
 		},
 		"source": map[string]interface{}{
-			"path": backupPath,
-			"type": "local",
+			"local-storage": map[string]interface{}{
+				"path": "/localStorage",
+			},
 		},
+		"backup-data-path": backupPath,
+	}
+}
+
+func validateRestoredData(k8sClient client.Client) error {
+	aeroCluster, err := cluster.GetCluster(k8sClient, testCtx, destinationAerospikeClusterNsNm)
+	if err != nil {
+		return err
 	}
+
+	records, err := cluster.CheckDataInCluster(aeroCluster, k8sClient, []string{"test"})
+	if err != nil {
+		return err
+	}
+
+	for ns, recordExists := range records {
+		if !recordExists {
+			return fmt.Errorf("namespace: %s - should have records", ns)
+		}
+	}
+
+	return nil
 }