diff --git a/changelogs/unreleased/5179-lyndon b/changelogs/unreleased/5179-lyndon
new file mode 100644
index 0000000000..65ae6cc00c
--- /dev/null
+++ b/changelogs/unreleased/5179-lyndon
@@ -0,0 +1 @@
+Add changes for Kopia Integration: Unified Repository Provider - method implementation
\ No newline at end of file
diff --git a/pkg/repository/provider/provider.go b/pkg/repository/provider/provider.go
index 36d69a594e..8e6a639a4a 100644
--- a/pkg/repository/provider/provider.go
+++ b/pkg/repository/provider/provider.go
@@ -23,10 +23,9 @@ import (
 )
 
 // RepoParam includes the parameters to manipulate a backup repository
-// SubDir is used to generate the path in the backup storage
 type RepoParam struct {
-	SubDir         string
 	BackupLocation *velerov1api.BackupStorageLocation
+	BackupRepo     *velerov1api.BackupRepository
 }
 
 type Provider interface {
diff --git a/pkg/repository/provider/unified_repo.go b/pkg/repository/provider/unified_repo.go
index 8ddff57999..994d0a5ca2 100644
--- a/pkg/repository/provider/unified_repo.go
+++ b/pkg/repository/provider/unified_repo.go
@@ -30,7 +30,7 @@ import (
 	repoconfig "github.com/vmware-tanzu/velero/pkg/repository/config"
 	repokey "github.com/vmware-tanzu/velero/pkg/repository/keys"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
-	"github.com/vmware-tanzu/velero/pkg/util/ownership"
+	reposervice "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/service"
 )
 
 type unifiedRepoProvider struct {
@@ -49,28 +49,28 @@ var getS3BucketRegion = repoconfig.GetAWSBucketRegion
 var getAzureStorageDomain = repoconfig.GetAzureStorageDomain
 
 type localFuncTable struct {
-	getRepoPassword       func(credentials.SecretStore, RepoParam) (string, error)
 	getStorageVariables   func(*velerov1api.BackupStorageLocation, string) (map[string]string, error)
 	getStorageCredentials func(*velerov1api.BackupStorageLocation, credentials.FileStore) (map[string]string, error)
 }
 
 var funcTable = localFuncTable{
-	getRepoPassword:       getRepoPassword,
 	getStorageVariables:   getStorageVariables,
 	getStorageCredentials: getStorageCredentials,
 }
 
+const (
+	repoOpDescFullMaintain  = "full maintenance"
+	repoOpDescQuickMaintain = "quick maintenance"
+	repoOpDescForget        = "forget"
+)
+
 // NewUnifiedRepoProvider creates the service provider for Unified Repo
-// workPath is the path for Unified Repo to store some local information
-// workPath could be empty, if so, the default path will be used
 func NewUnifiedRepoProvider(
 	credentialGetter credentials.CredentialGetter,
-	workPath string,
 	log logrus.FieldLogger,
 ) (Provider, error) {
 	repo := unifiedRepoProvider{
 		credentialGetter: credentialGetter,
-		workPath:         workPath,
 		log:              log,
 	}
 
@@ -89,12 +89,18 @@ func (urp *unifiedRepoProvider) InitRepo(ctx context.Context, param RepoParam) e
 
 	log.Debug("Start to init repo")
 
-	repoOption, err := urp.getRepoOption(param)
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithStoreOptions(urp, param),
+		udmrepo.WithDescription(repoOpDescFullMaintain),
+	)
+
 	if err != nil {
 		return errors.Wrap(err, "error to get repo options")
 	}
 
-	err = urp.repoService.Init(ctx, repoOption, true)
+	err = urp.repoService.Init(ctx, *repoOption, true)
 	if err != nil {
 		return errors.Wrap(err, "error to init backup repo")
 	}
@@ -105,22 +111,124 @@ func (urp *unifiedRepoProvider) InitRepo(ctx context.Context, param RepoParam) e
 }
 
 func (urp *unifiedRepoProvider) ConnectToRepo(ctx context.Context, param RepoParam) error {
-	///TODO
+	log := urp.log.WithFields(logrus.Fields{
+		"BSL name": param.BackupLocation.Name,
+		"BSL UID":  param.BackupLocation.UID,
+	})
+
+	log.Debug("Start to connect repo")
+
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithStoreOptions(urp, param),
+		udmrepo.WithDescription(repoOpDescFullMaintain),
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "error to get repo options")
+	}
+
+	err = urp.repoService.Init(ctx, *repoOption, false)
+	if err != nil {
+		return errors.Wrap(err, "error to connect backup repo")
+	}
+
+	log.Debug("Connect repo complete")
+
 	return nil
 }
 
 func (urp *unifiedRepoProvider) PrepareRepo(ctx context.Context, param RepoParam) error {
-	///TODO
+	log := urp.log.WithFields(logrus.Fields{
+		"BSL name": param.BackupLocation.Name,
+		"BSL UID":  param.BackupLocation.UID,
+	})
+
+	log.Debug("Start to prepare repo")
+
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithStoreOptions(urp, param),
+		udmrepo.WithDescription(repoOpDescFullMaintain),
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "error to get repo options")
+	}
+
+	err = urp.repoService.Init(ctx, *repoOption, false)
+	if err == nil {
+		log.Debug("Repo has already been initialized remotely")
+		return nil
+	}
+
+	err = urp.repoService.Init(ctx, *repoOption, true)
+	if err != nil {
+		return errors.Wrap(err, "error to init backup repo")
+	}
+
+	log.Debug("Prepare repo complete")
+
 	return nil
 }
 
 func (urp *unifiedRepoProvider) PruneRepo(ctx context.Context, param RepoParam) error {
-	///TODO
+	log := urp.log.WithFields(logrus.Fields{
+		"BSL name": param.BackupLocation.Name,
+		"BSL UID":  param.BackupLocation.UID,
+	})
+
+	log.Debug("Start to prune repo")
+
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithGenOptions(map[string]string{udmrepo.GenOptionMaintainMode: udmrepo.GenOptionMaintainFull}),
+		udmrepo.WithDescription(repoOpDescFullMaintain),
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "error to get repo options")
+	}
+
+	err = urp.repoService.Maintain(ctx, *repoOption)
+	if err != nil {
+		return errors.Wrap(err, "error to prune backup repo")
+	}
+
+	log.Debug("Prune repo complete")
+
 	return nil
 }
 
 func (urp *unifiedRepoProvider) PruneRepoQuick(ctx context.Context, param RepoParam) error {
-	///TODO
+	log := urp.log.WithFields(logrus.Fields{
+		"BSL name": param.BackupLocation.Name,
+		"BSL UID":  param.BackupLocation.UID,
+	})
+
+	log.Debug("Start to prune repo quick")
+
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithGenOptions(map[string]string{udmrepo.GenOptionMaintainMode: udmrepo.GenOptionMaintainQuick}),
+		udmrepo.WithDescription(repoOpDescQuickMaintain),
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "error to get repo options")
+	}
+
+	err = urp.repoService.Maintain(ctx, *repoOption)
+	if err != nil {
+		return errors.Wrap(err, "error to prune backup repo quick")
+	}
+
+	log.Debug("Prune repo quick complete")
+
 	return nil
 }
 
@@ -129,60 +237,108 @@ func (urp *unifiedRepoProvider) EnsureUnlockRepo(ctx context.Context, param Repo
 }
 
 func (urp *unifiedRepoProvider) Forget(ctx context.Context, snapshotID string, param RepoParam) error {
-	///TODO
+	log := urp.log.WithFields(logrus.Fields{
+		"BSL name":   param.BackupLocation.Name,
+		"BSL UID":    param.BackupLocation.UID,
+		"snapshotID": snapshotID,
+	})
+
+	log.Debug("Start to forget snapshot")
+
+	repoOption, err := udmrepo.NewRepoOptions(
+		udmrepo.WithPassword(urp, param),
+		udmrepo.WithConfigFile(urp.workPath, string(param.BackupLocation.UID)),
+		udmrepo.WithDescription(repoOpDescForget),
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "error to get repo options")
+	}
+
+	bkRepo, err := urp.repoService.Open(ctx, *repoOption)
+	if err != nil {
+		return errors.Wrap(err, "error to open backup repo")
+	}
+
+	defer func() {
+		c := bkRepo.Close(ctx)
+		if c != nil {
+			log.WithError(c).Error("Failed to close repo")
+		}
+	}()
+
+	err = bkRepo.DeleteManifest(ctx, udmrepo.ID(snapshotID))
+	if err != nil {
+		return errors.Wrap(err, "error to delete manifest")
+	}
+
+	log.Debug("Forget snapshot complete")
+
 	return nil
 }
 
-func getRepoPassword(secretStore credentials.SecretStore, param RepoParam) (string, error) {
-	if secretStore == nil {
-		return "", errors.New("invalid credentials interface")
+func (urp *unifiedRepoProvider) GetPassword(param interface{}) (string, error) {
+	repoParam, ok := param.(RepoParam)
+	if !ok {
+		return "", errors.New("invalid parameter")
 	}
 
-	buf, err := secretStore.Get(repokey.RepoKeySelector())
+	repoPassword, err := getRepoPassword(urp.credentialGetter.FromSecret, repoParam)
 	if err != nil {
-		return "", errors.Wrap(err, "error to get password buffer")
+		return "", errors.Wrap(err, "error to get repo password")
 	}
 
-	return strings.TrimSpace(string(buf)), nil
+	return repoPassword, nil
 }
 
-func (urp *unifiedRepoProvider) getRepoOption(param RepoParam) (udmrepo.RepoOptions, error) {
-	repoPassword, err := funcTable.getRepoPassword(urp.credentialGetter.FromSecret, param)
-	if err != nil {
-		return udmrepo.RepoOptions{}, errors.Wrap(err, "error to get repo password")
+func (urp *unifiedRepoProvider) GetStoreType(param interface{}) (string, error) {
+	repoParam, ok := param.(RepoParam)
+	if !ok {
+		return "", errors.New("invalid parameter")
 	}
 
-	storeVar, err := funcTable.getStorageVariables(param.BackupLocation, param.SubDir)
-	if err != nil {
-		return udmrepo.RepoOptions{}, errors.Wrap(err, "error to get storage variables")
+	return getStorageType(repoParam.BackupLocation), nil
+}
+
+func (urp *unifiedRepoProvider) GetStoreOptions(param interface{}) (map[string]string, error) {
+	repoParam, ok := param.(RepoParam)
+	if !ok {
+		return map[string]string{}, errors.New("invalid parameter")
 	}
 
-	storeCred, err := funcTable.getStorageCredentials(param.BackupLocation, urp.credentialGetter.FromFile)
+	storeVar, err := funcTable.getStorageVariables(repoParam.BackupLocation, repoParam.BackupRepo.Spec.VolumeNamespace)
 	if err != nil {
-		return udmrepo.RepoOptions{}, errors.Wrap(err, "error to get repo credentials")
+		return map[string]string{}, errors.Wrap(err, "error to get storage variables")
 	}
 
-	repoOption := udmrepo.RepoOptions{
-		StorageType:    getStorageType(param.BackupLocation),
-		RepoPassword:   repoPassword,
-		ConfigFilePath: getRepoConfigFile(urp.workPath, string(param.BackupLocation.UID)),
-		Ownership: udmrepo.OwnershipOptions{
-			Username:   ownership.GetRepositoryOwner().Username,
-			DomainName: ownership.GetRepositoryOwner().DomainName,
-		},
-		StorageOptions: make(map[string]string),
-		GeneralOptions: make(map[string]string),
+	storeCred, err := funcTable.getStorageCredentials(repoParam.BackupLocation, urp.credentialGetter.FromFile)
+	if err != nil {
+		return map[string]string{}, errors.Wrap(err, "error to get repo credentials")
 	}
 
+	storeOptions := make(map[string]string)
 	for k, v := range storeVar {
-		repoOption.StorageOptions[k] = v
+		storeOptions[k] = v
 	}
 
 	for k, v := range storeCred {
-		repoOption.StorageOptions[k] = v
+		storeOptions[k] = v
 	}
 
-	return repoOption, nil
+	return storeOptions, nil
+}
+
+func getRepoPassword(secretStore credentials.SecretStore, param RepoParam) (string, error) {
+	if secretStore == nil {
+		return "", errors.New("invalid credentials interface")
+	}
+
+	rawPass, err := secretStore.Get(repokey.RepoKeySelector())
+	if err != nil {
+		return "", errors.Wrap(err, "error to get password")
+	}
+
+	return strings.TrimSpace(rawPass), nil
 }
 
 func getStorageType(backupLocation *velerov1api.BackupStorageLocation) string {
@@ -304,12 +460,6 @@ func getStorageVariables(backupLocation *velerov1api.BackupStorageLocation, repo
 	return result, nil
 }
 
-func getRepoConfigFile(workPath string, repoID string) string {
-	///TODO: call udmrepo to get config file
-	return ""
-}
-
 func createRepoService(log logrus.FieldLogger) udmrepo.BackupRepoService {
-	///TODO: call udmrepo create repo service
-	return nil
+	return reposervice.Create(log)
 }
diff --git a/pkg/repository/provider/unified_repo_test.go b/pkg/repository/provider/unified_repo_test.go
index f2cccb8e5d..8e41b9b41b 100644
--- a/pkg/repository/provider/unified_repo_test.go
+++ b/pkg/repository/provider/unified_repo_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package provider
 
 import (
+	"context"
 	"errors"
 	"testing"
 
@@ -31,6 +32,8 @@ import (
 	credmock "github.com/vmware-tanzu/velero/internal/credentials/mocks"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+	reposervicenmocks "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/mocks"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )
 
 func TestGetStorageCredentials(t *testing.T) {
@@ -451,7 +454,7 @@ func TestGetRepoPassword(t *testing.T) {
 			name:           "error from secret interface",
 			getter:         new(credmock.SecretStore),
 			credStoreError: errors.New("fake error"),
-			expectedErr:    "error to get password buffer: fake error",
+			expectedErr:    "error to get password: fake error",
 		},
 		{
 			name:            "secret with whitespace",
@@ -488,43 +491,41 @@ func TestGetRepoPassword(t *testing.T) {
 	}
 }
 
-func TestGetRepoOption(t *testing.T) {
+func TestGetStoreOptions(t *testing.T) {
 	testCases := []struct {
-		name                  string
-		funcTable             localFuncTable
-		getRepoPassword       func(velerocredentials.SecretStore, RepoParam) (string, error)
-		getStorageCredentials func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error)
-		getStorageVariables   func(*velerov1api.BackupStorageLocation, string) (map[string]string, error)
-		expected              udmrepo.RepoOptions
-		expectedErr           string
+		name        string
+		funcTable   localFuncTable
+		repoParam   interface{}
+		expected    map[string]string
+		expectedErr string
 	}{
 		{
-			name: "get repo password fail",
-			funcTable: localFuncTable{
-				getRepoPassword: func(velerocredentials.SecretStore, RepoParam) (string, error) {
-					return "", errors.New("fake-error-1")
-				},
-			},
-			expectedErr: "error to get repo password: fake-error-1",
+			name:        "wrong param type",
+			repoParam:   struct{}{},
+			expected:    map[string]string{},
+			expectedErr: "invalid parameter",
 		},
 		{
 			name: "get storage variable fail",
+			repoParam: RepoParam{
+				BackupLocation: &velerov1api.BackupStorageLocation{},
+				BackupRepo:     &velerov1api.BackupRepository{},
+			},
 			funcTable: localFuncTable{
-				getRepoPassword: func(velerocredentials.SecretStore, RepoParam) (string, error) {
-					return "fake-password", nil
-				},
 				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
 					return map[string]string{}, errors.New("fake-error-2")
 				},
 			},
+			expected:    map[string]string{},
 			expectedErr: "error to get storage variables: fake-error-2",
 		},
 		{
 			name: "get storage credentials fail",
+			repoParam: RepoParam{
+				BackupLocation: &velerov1api.BackupStorageLocation{},
+				BackupRepo:     &velerov1api.BackupRepository{},
+			},
 			funcTable: localFuncTable{
-				getRepoPassword: func(velerocredentials.SecretStore, RepoParam) (string, error) {
-					return "fake-password", nil
-				},
 				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
 					return map[string]string{}, nil
 				},
@@ -532,6 +533,7 @@ func TestGetRepoOption(t *testing.T) {
 					return map[string]string{}, errors.New("fake-error-3")
 				},
 			},
+			expected:    map[string]string{},
 			expectedErr: "error to get repo credentials: fake-error-3",
 		},
 	}
@@ -539,11 +541,241 @@ func TestGetRepoOption(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			funcTable = tc.funcTable
+
 			urp := unifiedRepoProvider{}
 
-			password, err := urp.getRepoOption(RepoParam{})
+			options, err := urp.GetStoreOptions(tc.repoParam)
 
-			require.Equal(t, tc.expected, password)
+			require.Equal(t, tc.expected, options)
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}
+
+func TestPrepareRepo(t *testing.T) {
+	testCases := []struct {
+		name            string
+		funcTable       localFuncTable
+		getter          *credmock.SecretStore
+		repoService     *reposervicenmocks.BackupRepoService
+		retFuncInit     func(context.Context, udmrepo.RepoOptions, bool) error
+		credStoreReturn string
+		credStoreError  error
+		expectedErr     string
+	}{
+		{
+			name:        "get repo option fail",
+			repoService: new(reposervicenmocks.BackupRepoService),
+			expectedErr: "error to get repo options: error to get repo password: invalid credentials interface",
+		},
+		{
+			name:           "get repo option fail, get password fail",
+			getter:         new(credmock.SecretStore),
+			repoService:    new(reposervicenmocks.BackupRepoService),
+			credStoreError: errors.New("fake-password-error"),
+			expectedErr:    "error to get repo options: error to get repo password: error to get password: fake-password-error",
+		},
+		{
+			name:            "get repo option fail, get store options fail",
+			getter:          new(credmock.SecretStore),
+			repoService:     new(reposervicenmocks.BackupRepoService),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
+					return map[string]string{}, errors.New("fake-store-option-error")
+				},
+			},
+			expectedErr: "error to get repo options: error to get storage variables: fake-store-option-error",
+		},
+		{
+			name:            "already initialized",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			retFuncInit: func(ctx context.Context, repoOption udmrepo.RepoOptions, createNew bool) error {
+				if !createNew {
+					return nil
+				} else {
+					return errors.New("fake-error")
+				}
+			},
+		},
+		{
+			name:            "initialize fail",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			retFuncInit: func(ctx context.Context, repoOption udmrepo.RepoOptions, createNew bool) error {
+				if !createNew {
+					return errors.New("fake-error-1")
+				} else {
+					return errors.New("fake-error-2")
+				}
+			},
+			expectedErr: "error to init backup repo: fake-error-2",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			funcTable = tc.funcTable
+
+			var secretStore velerocredentials.SecretStore
+			if tc.getter != nil {
+				tc.getter.On("Get", mock.Anything, mock.Anything).Return(tc.credStoreReturn, tc.credStoreError)
+				secretStore = tc.getter
+			}
+
+			urp := unifiedRepoProvider{
+				credentialGetter: velerocredentials.CredentialGetter{
+					FromSecret: secretStore,
+				},
+				repoService: tc.repoService,
+				log:         velerotest.NewLogger(),
+			}
+
+			tc.repoService.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(tc.retFuncInit)
+
+			err := urp.PrepareRepo(context.Background(), RepoParam{
+				BackupLocation: &velerov1api.BackupStorageLocation{},
+				BackupRepo:     &velerov1api.BackupRepository{},
+			})
+
+			if tc.expectedErr == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.EqualError(t, err, tc.expectedErr)
+			}
+		})
+	}
+}
+
+func TestForget(t *testing.T) {
+	var backupRepo *reposervicenmocks.BackupRepo
+
+	testCases := []struct {
+		name            string
+		funcTable       localFuncTable
+		getter          *credmock.SecretStore
+		repoService     *reposervicenmocks.BackupRepoService
+		backupRepo      *reposervicenmocks.BackupRepo
+		retFuncOpen     []interface{}
+		retFuncDelete   interface{}
+		credStoreReturn string
+		credStoreError  error
+		expectedErr     string
+	}{
+		{
+			name:        "get repo option fail",
+			expectedErr: "error to get repo options: error to get repo password: invalid credentials interface",
+		},
+		{
+			name:            "repo open fail",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			retFuncOpen: []interface{}{
+				func(context.Context, udmrepo.RepoOptions) udmrepo.BackupRepo {
+					return backupRepo
+				},
+
+				func(context.Context, udmrepo.RepoOptions) error {
+					return errors.New("fake-error-2")
+				},
+			},
+			expectedErr: "error to open backup repo: fake-error-2",
+		},
+		{
+			name:            "delete fail",
+			getter:          new(credmock.SecretStore),
+			credStoreReturn: "fake-password",
+			funcTable: localFuncTable{
+				getStorageVariables: func(*velerov1api.BackupStorageLocation, string) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+				getStorageCredentials: func(*velerov1api.BackupStorageLocation, velerocredentials.FileStore) (map[string]string, error) {
+					return map[string]string{}, nil
+				},
+			},
+			repoService: new(reposervicenmocks.BackupRepoService),
+			backupRepo:  new(reposervicenmocks.BackupRepo),
+			retFuncOpen: []interface{}{
+				func(context.Context, udmrepo.RepoOptions) udmrepo.BackupRepo {
+					return backupRepo
+				},
+
+				func(context.Context, udmrepo.RepoOptions) error {
+					return nil
+				},
+			},
+			retFuncDelete: func(context.Context, udmrepo.ID) error {
+				return errors.New("fake-error-3")
+			},
+			expectedErr: "error to delete manifest: fake-error-3",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			funcTable = tc.funcTable
+
+			var secretStore velerocredentials.SecretStore
+			if tc.getter != nil {
+				tc.getter.On("Get", mock.Anything, mock.Anything).Return(tc.credStoreReturn, tc.credStoreError)
+				secretStore = tc.getter
+			}
+
+			urp := unifiedRepoProvider{
+				credentialGetter: velerocredentials.CredentialGetter{
+					FromSecret: secretStore,
+				},
+				repoService: tc.repoService,
+				log:         velerotest.NewLogger(),
+			}
+
+			backupRepo = tc.backupRepo
+
+			if tc.repoService != nil {
+				tc.repoService.On("Open", mock.Anything, mock.Anything).Return(tc.retFuncOpen[0], tc.retFuncOpen[1])
+			}
+
+			if tc.backupRepo != nil {
+				backupRepo.On("DeleteManifest", mock.Anything, mock.Anything).Return(tc.retFuncDelete)
+				backupRepo.On("Close", mock.Anything).Return(nil)
+			}
+
+			err := urp.Forget(context.Background(), "", RepoParam{
+				BackupLocation: &velerov1api.BackupStorageLocation{},
+			})
 
 			if tc.expectedErr == "" {
 				assert.NoError(t, err)
diff --git a/pkg/repository/udmrepo/mocks/BackupRepo.go b/pkg/repository/udmrepo/mocks/BackupRepo.go
new file mode 100644
index 0000000000..ea8e2ba3c4
--- /dev/null
+++ b/pkg/repository/udmrepo/mocks/BackupRepo.go
@@ -0,0 +1,185 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+	time "time"
+
+	mock "github.com/stretchr/testify/mock"
+
+	udmrepo "github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+// BackupRepo is an autogenerated mock type for the BackupRepo type
+type BackupRepo struct {
+	mock.Mock
+}
+
+// Close provides a mock function with given fields: ctx
+func (_m *BackupRepo) Close(ctx context.Context) error {
+	ret := _m.Called(ctx)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = rf(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// DeleteManifest provides a mock function with given fields: ctx, id
+func (_m *BackupRepo) DeleteManifest(ctx context.Context, id udmrepo.ID) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.ID) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// FindManifests provides a mock function with given fields: ctx, filter
+func (_m *BackupRepo) FindManifests(ctx context.Context, filter udmrepo.ManifestFilter) ([]*udmrepo.ManifestEntryMetadata, error) {
+	ret := _m.Called(ctx, filter)
+
+	var r0 []*udmrepo.ManifestEntryMetadata
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.ManifestFilter) []*udmrepo.ManifestEntryMetadata); ok {
+		r0 = rf(ctx, filter)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*udmrepo.ManifestEntryMetadata)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, udmrepo.ManifestFilter) error); ok {
+		r1 = rf(ctx, filter)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Flush provides a mock function with given fields: ctx
+func (_m *BackupRepo) Flush(ctx context.Context) error {
+	ret := _m.Called(ctx)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = rf(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// GetManifest provides a mock function with given fields: ctx, id, mani
+func (_m *BackupRepo) GetManifest(ctx context.Context, id udmrepo.ID, mani *udmrepo.RepoManifest) error {
+	ret := _m.Called(ctx, id, mani)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.ID, *udmrepo.RepoManifest) error); ok {
+		r0 = rf(ctx, id, mani)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// NewObjectWriter provides a mock function with given fields: ctx, opt
+func (_m *BackupRepo) NewObjectWriter(ctx context.Context, opt udmrepo.ObjectWriteOptions) udmrepo.ObjectWriter {
+	ret := _m.Called(ctx, opt)
+
+	var r0 udmrepo.ObjectWriter
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.ObjectWriteOptions) udmrepo.ObjectWriter); ok {
+		r0 = rf(ctx, opt)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(udmrepo.ObjectWriter)
+		}
+	}
+
+	return r0
+}
+
+// OpenObject provides a mock function with given fields: ctx, id
+func (_m *BackupRepo) OpenObject(ctx context.Context, id udmrepo.ID) (udmrepo.ObjectReader, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 udmrepo.ObjectReader
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.ID) udmrepo.ObjectReader); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(udmrepo.ObjectReader)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, udmrepo.ID) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// PutManifest provides a mock function with given fields: ctx, mani
+func (_m *BackupRepo) PutManifest(ctx context.Context, mani udmrepo.RepoManifest) (udmrepo.ID, error) {
+	ret := _m.Called(ctx, mani)
+
+	var r0 udmrepo.ID
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.RepoManifest) udmrepo.ID); ok {
+		r0 = rf(ctx, mani)
+	} else {
+		r0 = ret.Get(0).(udmrepo.ID)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, udmrepo.RepoManifest) error); ok {
+		r1 = rf(ctx, mani)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Time provides a mock function with given fields:
+func (_m *BackupRepo) Time() time.Time {
+	ret := _m.Called()
+
+	var r0 time.Time
+	if rf, ok := ret.Get(0).(func() time.Time); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(time.Time)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewBackupRepo interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewBackupRepo creates a new instance of BackupRepo. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewBackupRepo(t mockConstructorTestingTNewBackupRepo) *BackupRepo {
+	mock := &BackupRepo{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/pkg/repository/udmrepo/mocks/BackupRepoService.go b/pkg/repository/udmrepo/mocks/BackupRepoService.go
new file mode 100644
index 0000000000..135c0058c4
--- /dev/null
+++ b/pkg/repository/udmrepo/mocks/BackupRepoService.go
@@ -0,0 +1,81 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+	udmrepo "github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+// BackupRepoService is an autogenerated mock type for the BackupRepoService type
+type BackupRepoService struct {
+	mock.Mock
+}
+
+// Init provides a mock function with given fields: ctx, repoOption, createNew
+func (_m *BackupRepoService) Init(ctx context.Context, repoOption udmrepo.RepoOptions, createNew bool) error {
+	ret := _m.Called(ctx, repoOption, createNew)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.RepoOptions, bool) error); ok {
+		r0 = rf(ctx, repoOption, createNew)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Maintain provides a mock function with given fields: ctx, repoOption
+func (_m *BackupRepoService) Maintain(ctx context.Context, repoOption udmrepo.RepoOptions) error {
+	ret := _m.Called(ctx, repoOption)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.RepoOptions) error); ok {
+		r0 = rf(ctx, repoOption)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Open provides a mock function with given fields: ctx, repoOption
+func (_m *BackupRepoService) Open(ctx context.Context, repoOption udmrepo.RepoOptions) (udmrepo.BackupRepo, error) {
+	ret := _m.Called(ctx, repoOption)
+
+	var r0 udmrepo.BackupRepo
+	if rf, ok := ret.Get(0).(func(context.Context, udmrepo.RepoOptions) udmrepo.BackupRepo); ok {
+		r0 = rf(ctx, repoOption)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(udmrepo.BackupRepo)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, udmrepo.RepoOptions) error); ok {
+		r1 = rf(ctx, repoOption)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+type mockConstructorTestingTNewBackupRepoService interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewBackupRepoService creates a new instance of BackupRepoService. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewBackupRepoService(t mockConstructorTestingTNewBackupRepoService) *BackupRepoService {
+	mock := &BackupRepoService{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/pkg/repository/udmrepo/mocks/ObjectReader.go b/pkg/repository/udmrepo/mocks/ObjectReader.go
new file mode 100644
index 0000000000..2410acd6e6
--- /dev/null
+++ b/pkg/repository/udmrepo/mocks/ObjectReader.go
@@ -0,0 +1,95 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+
+// ObjectReader is an autogenerated mock type for the ObjectReader type
+type ObjectReader struct {
+	mock.Mock
+}
+
+// Close provides a mock function with given fields:
+func (_m *ObjectReader) Close() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Length provides a mock function with given fields:
+func (_m *ObjectReader) Length() int64 {
+	ret := _m.Called()
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func() int64); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	return r0
+}
+
+// Read provides a mock function with given fields: p
+func (_m *ObjectReader) Read(p []byte) (int, error) {
+	ret := _m.Called(p)
+
+	var r0 int
+	if rf, ok := ret.Get(0).(func([]byte) int); ok {
+		r0 = rf(p)
+	} else {
+		r0 = ret.Get(0).(int)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func([]byte) error); ok {
+		r1 = rf(p)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Seek provides a mock function with given fields: offset, whence
+func (_m *ObjectReader) Seek(offset int64, whence int) (int64, error) {
+	ret := _m.Called(offset, whence)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64, int) int64); ok {
+		r0 = rf(offset, whence)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int64, int) error); ok {
+		r1 = rf(offset, whence)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+type mockConstructorTestingTNewObjectReader interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewObjectReader creates a new instance of ObjectReader. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewObjectReader(t mockConstructorTestingTNewObjectReader) *ObjectReader {
+	mock := &ObjectReader{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/pkg/repository/udmrepo/mocks/ObjectWriter.go b/pkg/repository/udmrepo/mocks/ObjectWriter.go
new file mode 100644
index 0000000000..277a0ed4a3
--- /dev/null
+++ b/pkg/repository/udmrepo/mocks/ObjectWriter.go
@@ -0,0 +1,126 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package mocks
+
+import (
+	mock "github.com/stretchr/testify/mock"
+	udmrepo "github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+// ObjectWriter is an autogenerated mock type for the ObjectWriter type
+type ObjectWriter struct {
+	mock.Mock
+}
+
+// Checkpoint provides a mock function with given fields:
+func (_m *ObjectWriter) Checkpoint() (udmrepo.ID, error) {
+	ret := _m.Called()
+
+	var r0 udmrepo.ID
+	if rf, ok := ret.Get(0).(func() udmrepo.ID); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(udmrepo.ID)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Close provides a mock function with given fields:
+func (_m *ObjectWriter) Close() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Result provides a mock function with given fields:
+func (_m *ObjectWriter) Result() (udmrepo.ID, error) {
+	ret := _m.Called()
+
+	var r0 udmrepo.ID
+	if rf, ok := ret.Get(0).(func() udmrepo.ID); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(udmrepo.ID)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Seek provides a mock function with given fields: offset, whence
+func (_m *ObjectWriter) Seek(offset int64, whence int) (int64, error) {
+	ret := _m.Called(offset, whence)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(int64, int) int64); ok {
+		r0 = rf(offset, whence)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int64, int) error); ok {
+		r1 = rf(offset, whence)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Write provides a mock function with given fields: p
+func (_m *ObjectWriter) Write(p []byte) (int, error) {
+	ret := _m.Called(p)
+
+	var r0 int
+	if rf, ok := ret.Get(0).(func([]byte) int); ok {
+		r0 = rf(p)
+	} else {
+		r0 = ret.Get(0).(int)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func([]byte) error); ok {
+		r1 = rf(p)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+type mockConstructorTestingTNewObjectWriter interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewObjectWriter creates a new instance of ObjectWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewObjectWriter(t mockConstructorTestingTNewObjectWriter) *ObjectWriter {
+	mock := &ObjectWriter{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/pkg/repository/udmrepo/repo-option-consts.go b/pkg/repository/udmrepo/repo-option-consts.go
deleted file mode 100644
index 7cf55d017c..0000000000
--- a/pkg/repository/udmrepo/repo-option-consts.go
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
-Copyright the Velero contributors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package udmrepo
-
-const (
-	StorageTypeS3    = "s3"
-	StorageTypeAzure = "azure"
-	StorageTypeFs    = "filesystem"
-	StorageTypeGcs   = "gcs"
-
-	GenOptionMaintainMode  = "mode"
-	GenOptionMaintainFull  = "full"
-	GenOptionMaintainQuick = "quick"
-
-	StoreOptionS3KeyId            = "accessKeyID"
-	StoreOptionS3Provider         = "providerName"
-	StoreOptionS3SecretKey        = "secretAccessKey"
-	StoreOptionS3Token            = "sessionToken"
-	StoreOptionS3Endpoint         = "endpoint"
-	StoreOptionS3DisableTls       = "doNotUseTLS"
-	StoreOptionS3DisableTlsVerify = "skipTLSVerify"
-
-	StoreOptionAzureKey            = "storageKey"
-	StoreOptionAzureDomain         = "storageDomain"
-	StoreOptionAzureStorageAccount = "storageAccount"
-	StoreOptionAzureToken          = "sasToken"
-
-	StoreOptionFsPath = "fspath"
-
-	StoreOptionGcsReadonly = "readonly"
-
-	StoreOptionOssBucket = "bucket"
-	StoreOptionOssRegion = "region"
-
-	StoreOptionCredentialFile = "credFile"
-	StoreOptionPrefix         = "prefix"
-	StoreOptionPrefixName     = "unified-repo"
-
-	ThrottleOptionReadOps       = "readOPS"
-	ThrottleOptionWriteOps      = "writeOPS"
-	ThrottleOptionListOps       = "listOPS"
-	ThrottleOptionUploadBytes   = "uploadBytes"
-	ThrottleOptionDownloadBytes = "downloadBytes"
-)
diff --git a/pkg/repository/udmrepo/repo-options.go b/pkg/repository/udmrepo/repo-options.go
new file mode 100644
index 0000000000..b5fd48cf79
--- /dev/null
+++ b/pkg/repository/udmrepo/repo-options.go
@@ -0,0 +1,171 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package udmrepo
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+const (
+	StorageTypeS3    = "s3"
+	StorageTypeAzure = "azure"
+	StorageTypeFs    = "filesystem"
+	StorageTypeGcs   = "gcs"
+
+	GenOptionMaintainMode  = "mode"
+	GenOptionMaintainFull  = "full"
+	GenOptionMaintainQuick = "quick"
+
+	StoreOptionS3KeyId            = "accessKeyID"
+	StoreOptionS3Provider         = "providerName"
+	StoreOptionS3SecretKey        = "secretAccessKey"
+	StoreOptionS3Token            = "sessionToken"
+	StoreOptionS3Endpoint         = "endpoint"
+	StoreOptionS3DisableTls       = "doNotUseTLS"
+	StoreOptionS3DisableTlsVerify = "skipTLSVerify"
+
+	StoreOptionAzureKey            = "storageKey"
+	StoreOptionAzureDomain         = "storageDomain"
+	StoreOptionAzureStorageAccount = "storageAccount"
+	StoreOptionAzureToken          = "sasToken"
+
+	StoreOptionFsPath = "fspath"
+
+	StoreOptionGcsReadonly = "readonly"
+
+	StoreOptionOssBucket = "bucket"
+	StoreOptionOssRegion = "region"
+
+	StoreOptionCredentialFile = "credFile"
+	StoreOptionPrefix         = "prefix"
+	StoreOptionPrefixName     = "unified-repo"
+
+	ThrottleOptionReadOps       = "readOPS"
+	ThrottleOptionWriteOps      = "writeOPS"
+	ThrottleOptionListOps       = "listOPS"
+	ThrottleOptionUploadBytes   = "uploadBytes"
+	ThrottleOptionDownloadBytes = "downloadBytes"
+)
+
+type RepoOptions struct {
+	// StorageType is a repository specific string to identify a backup storage, i.e., "s3", "filesystem"
+	StorageType string
+	// RepoPassword is the backup repository's password, if any
+	RepoPassword string
+	// ConfigFilePath is a custom path to save the repository's configuration, if any
+	ConfigFilePath string
+	// GeneralOptions takes other repository specific options
+	GeneralOptions map[string]string
+	// StorageOptions takes storage specific options
+	StorageOptions map[string]string
+
+	// Description is a description of the backup repository/backup repository operation.
+	// It is for logging/debugging purpose only and doesn't control any behavior of the backup repository.
+	Description string
+}
+
+type PasswordGetter interface {
+	GetPassword(param interface{}) (string, error)
+}
+
+type StoreOptionsGetter interface {
+	GetStoreType(param interface{}) (string, error)
+	GetStoreOptions(param interface{}) (map[string]string, error)
+}
+
+func NewRepoOptions(options ...func(*RepoOptions) error) (*RepoOptions, error) {
+	ro := &RepoOptions{}
+	for _, o := range options {
+		err := o(ro)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return ro, nil
+}
+
+func WithPassword(getter PasswordGetter, param interface{}) func(*RepoOptions) error {
+	return func(ro *RepoOptions) error {
+		password, err := getter.GetPassword(param)
+		if err != nil {
+			return err
+		}
+
+		ro.RepoPassword = password
+
+		return nil
+	}
+}
+
+func WithConfigFile(workPath string, repoID string) func(*RepoOptions) error {
+	return func(ro *RepoOptions) error {
+		ro.ConfigFilePath = getRepoConfigFile(workPath, repoID)
+		return nil
+	}
+}
+
+func WithGenOptions(genOptions map[string]string) func(*RepoOptions) error {
+	return func(ro *RepoOptions) error {
+		for k, v := range genOptions {
+			ro.GeneralOptions[k] = v
+		}
+
+		return nil
+	}
+}
+
+func WithStoreOptions(getter StoreOptionsGetter, param interface{}) func(*RepoOptions) error {
+	return func(ro *RepoOptions) error {
+		storeType, err := getter.GetStoreType(param)
+		if err != nil {
+			return err
+		}
+
+		storeOptions, err := getter.GetStoreOptions(param)
+		if err != nil {
+			return err
+		}
+
+		ro.StorageType = storeType
+
+		for k, v := range storeOptions {
+			ro.StorageOptions[k] = v
+		}
+
+		return nil
+	}
+}
+
+func WithDescription(desc string) func(*RepoOptions) error {
+	return func(ro *RepoOptions) error {
+		ro.Description = desc
+		return nil
+	}
+}
+
+func getRepoConfigFile(workPath string, repoID string) string {
+	if workPath == "" {
+		workPath = filepath.Join(os.Getenv("HOME"), "udmrepo")
+	}
+
+	name := "repo-" + strings.ToLower(repoID) + ".conf"
+
+	return filepath.Join(workPath, name)
+}
diff --git a/pkg/repository/udmrepo/repo.go b/pkg/repository/udmrepo/repo.go
index be18a6d176..01d434fdad 100644
--- a/pkg/repository/udmrepo/repo.go
+++ b/pkg/repository/udmrepo/repo.go
@@ -70,79 +70,51 @@ type ObjectWriteOptions struct {
 	BackupMode  int    // OBJECT_DATA_BACKUP_*
 }
 
-// OwnershipOptions is used to add some access control to the unified repository.
-// For example, some privileged operations of the unified repository can be done by the
-// repository owner only; the data of a backup may be manipulated by the backup owner
-// who created it only. It is optional for a backup repository to support this ownership control.
-type OwnershipOptions struct {
-	Username      string
-	DomainName    string
-	FullQualified string
-}
-
-type RepoOptions struct {
-	// A repository specific string to identify a backup storage, i.e., "s3", "filesystem"
-	StorageType string
-	// Backup repository password, if any
-	RepoPassword string
-	// A custom path to save the repository's configuration, if any
-	ConfigFilePath string
-	// The ownership for the current repository operation
-	Ownership OwnershipOptions
-	// Other repository specific options
-	GeneralOptions map[string]string
-	// Storage specific options
-	StorageOptions map[string]string
-
-	// Description of the backup repository
-	Description string
-}
-
 // BackupRepoService is used to initialize, open or maintain a backup repository
 type BackupRepoService interface {
-	// Create a backup repository or connect to an existing backup repository.
+	// Init creates a backup repository or connect to an existing backup repository.
 	// repoOption: option to the backup repository and the underlying backup storage.
 	// createNew: indicates whether to create a new or connect to an existing backup repository.
 	Init(ctx context.Context, repoOption RepoOptions, createNew bool) error
 
-	// Open an backup repository that has been created/connected.
+	// Open opens an backup repository that has been created/connected.
 	// repoOption: options to open the backup repository and the underlying storage.
 	Open(ctx context.Context, repoOption RepoOptions) (BackupRepo, error)
 
-	// Periodically called to maintain the backup repository to eliminate redundant data and improve performance.
+	// Maintain is periodically called to maintain the backup repository to eliminate redundant data.
 	// repoOption: options to maintain the backup repository.
 	Maintain(ctx context.Context, repoOption RepoOptions) error
 }
 
 // BackupRepo provides the access to the backup repository
 type BackupRepo interface {
-	// Open an existing object for read.
+	// OpenObject opens an existing object for read.
 	// id: the object's unified identifier.
 	OpenObject(ctx context.Context, id ID) (ObjectReader, error)
 
-	// Get a manifest data.
+	// GetManifest gets a manifest data from the backup repository.
 	GetManifest(ctx context.Context, id ID, mani *RepoManifest) error
 
-	// Get one or more manifest data that match the given labels
+	// FindManifests gets one or more manifest data that match the given labels
 	FindManifests(ctx context.Context, filter ManifestFilter) ([]*ManifestEntryMetadata, error)
 
-	// Create a new object and return the object's writer interface.
+	// NewObjectWriter creates a new object and return the object's writer interface.
 	// return: A unified identifier of the object on success.
 	NewObjectWriter(ctx context.Context, opt ObjectWriteOptions) ObjectWriter
 
-	// Save a manifest object
+	// PutManifest saves a manifest object into the backup repository.
 	PutManifest(ctx context.Context, mani RepoManifest) (ID, error)
 
-	// Delete a manifest object
+	// DeleteManifest deletes a manifest object from the backup repository.
 	DeleteManifest(ctx context.Context, id ID) error
 
-	// Flush all the backup repository data
+	// Flush flushes all the backup repository data
 	Flush(ctx context.Context) error
 
-	// Get the local time of the backup repository. It may be different from the time of the caller
+	// Time returns the local time of the backup repository. It may be different from the time of the caller
 	Time() time.Time
 
-	// Close the backup repository
+	// Close closes the backup repository
 	Close(ctx context.Context) error
 }
 
@@ -157,15 +129,15 @@ type ObjectReader interface {
 type ObjectWriter interface {
 	io.WriteCloser
 
-	// For some cases, i.e. block incremental, the object is not written sequentially
+	// Seeker is used in the cases that the object is not written sequentially
 	io.Seeker
 
-	// Periodically called to preserve the state of data written to the repo so far.
-	// Return a unified identifier that represent the current state.
+	// Checkpoint is periodically called to preserve the state of data written to the repo so far.
+	// Checkpoint returns a unified identifier that represent the current state.
 	// An empty ID could be returned on success if the backup repository doesn't support this.
 	Checkpoint() (ID, error)
 
-	// Wait for the completion of the object write.
+	// Result waits for the completion of the object write.
 	// Result returns the object's unified identifier after the write completes.
 	Result() (ID, error)
 }
diff --git a/pkg/repository/udmrepo/service/service.go b/pkg/repository/udmrepo/service/service.go
new file mode 100644
index 0000000000..55fbb03c8f
--- /dev/null
+++ b/pkg/repository/udmrepo/service/service.go
@@ -0,0 +1,37 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package service
+
+import (
+	"github.com/sirupsen/logrus"
+
+	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
+)
+
+const (
+	defaultUsername = "default"
+	defaultDomain   = "default"
+)
+
+func Create(logger logrus.FieldLogger) udmrepo.BackupRepoService {
+	///TODO: create from kopiaLib
+	return nil
+}
+
+func GetRepoUser() (username, domain string) {
+	return defaultUsername, defaultDomain
+}
diff --git a/pkg/util/ownership/backup_owner.go b/pkg/util/ownership/backup_owner.go
deleted file mode 100644
index 078c799dd9..0000000000
--- a/pkg/util/ownership/backup_owner.go
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
-Copyright the Velero contributors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package ownership
-
-import "github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
-
-const (
-	defaultOwnerUsername = "default"
-	defaultOwnerDomain   = "default"
-)
-
-// GetBackupOwner returns the owner used by uploaders when saving a snapshot or
-// opening the unified repository. At present, use the default owner only
-func GetBackupOwner() udmrepo.OwnershipOptions {
-	return udmrepo.OwnershipOptions{
-		Username:   defaultOwnerUsername,
-		DomainName: defaultOwnerDomain,
-	}
-}
-
-// GetBackupOwner returns the owner used to create/connect the unified repository.
-//At present, use the default owner only
-func GetRepositoryOwner() udmrepo.OwnershipOptions {
-	return udmrepo.OwnershipOptions{
-		Username:   defaultOwnerUsername,
-		DomainName: defaultOwnerDomain,
-	}
-}