add incremental aware object writer

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

changelogs/unreleased/9845-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #9823, add incremental ware object writer for block data mover

pkg/repository/udmrepo/kopialib/lib_repo.go

@@ -22,7 +22,6 @@ import (
 	"io"
 	"os"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -84,8 +83,6 @@ type kopiaObjectWriterEx struct {
 	ctx           context.Context
 	rawRepoWriter repo.RepositoryWriter
 	parentEntries []object.IndirectObjectEntry
-	entries       []object.IndirectObjectEntry
-	entryLock     sync.Mutex
 	blockSize     int64
 	description   string
 	compressor    compression.Name
@@ -855,12 +852,12 @@ func (kow *kopiaObjectWriter) Close() error {
 
 // TODO add implementation in following PRs
 func (kow *kopiaObjectWriterEx) Write(p []byte) (int, error) {
-	return 0, nil
+	return 0, errors.New("not implemented")
 }
 
 // TODO add implementation in following PRs
 func (kow *kopiaObjectWriterEx) WriteAt(p []byte, offset int64) (int, error) {
-	return 0, nil
+	return 0, errors.New("not implemented")
 }
 
 func (kow *kopiaObjectWriterEx) Checkpoint() (udmrepo.ID, error) {
@@ -869,12 +866,12 @@ func (kow *kopiaObjectWriterEx) Checkpoint() (udmrepo.ID, error) {
 
 // TODO add implementation in following PRs
 func (kow *kopiaObjectWriterEx) Result() (udmrepo.ID, error) {
-	return udmrepo.ID(""), nil
+	return udmrepo.ID(""), errors.New("not implemented")
 }
 
 // TODO add implementation in following PRs
 func (kow *kopiaObjectWriterEx) Close() error {
-	return nil
+	return errors.New("not implemented")
 }
 
 // getCompressorForObject returns the compressor for an object, at present, we don't support compression

pkg/repository/udmrepo/kopialib/lib_repo_ex_test.go

@@ -133,7 +133,7 @@ func TestGetFlattenedEntries(t *testing.T) {
 	kr := &kopiaRepository{logger: velerotest.NewLogger()}
 	rawID := object.ID{}
 	_, err := kr.getFlattenedEntries(context.Background(), rawID)
-	assert.Error(t, err)
+	require.Error(t, err)
 	assert.Contains(t, err.Error(), "object is not an indirect object")
 }
 
@@ -178,6 +178,14 @@ func TestNewObjectWriterEx(t *testing.T) {
 			rawWriter:   repomocks.NewMockRepositoryWriter(t),
 			expectedErr: "parent object is only supported for block mode",
 		},
+		{
+			name: "block mode success with async writes",
+			opt: udmrepo.ObjectWriteOptions{
+				AccessMode:  udmrepo.ObjectDataAccessModeBlock,
+				AsyncWrites: 4,
+			},
+			rawWriter: repomocks.NewMockRepositoryWriter(t),
+		},
 	}
 
 	for _, tc := range testCases {