Merge branch 'master' of https://github.com/bingoohuang/seaweedfs

19 files changed, 739 insertions, 326 deletions

diff --git a/weed/replication/repl_util/replication_util.go b/weed/replication/repl_util/replication_util.go
new file mode 100644
index 000000000..519a9a201
--- /dev/null
+++ b/weed/replication/repl_util/replication_util.go
@@ -0,0 +1,42 @@
+package repl_util
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/replication/source"
+	"github.com/chrislusf/seaweedfs/weed/util"
+)
+
+func CopyFromChunkViews(chunkViews []*filer.ChunkView, filerSource *source.FilerSource, writeFunc func(data []byte) error) error {
+
+	for _, chunk := range chunkViews {
+
+		fileUrls, err := filerSource.LookupFileId(chunk.FileId)
+		if err != nil {
+			return err
+		}
+
+		var writeErr error
+		var shouldRetry bool
+
+		for _, fileUrl := range fileUrls {
+			shouldRetry, err = util.ReadUrlAsStream(fileUrl, nil, false, chunk.IsFullChunk(), chunk.Offset, int(chunk.Size), func(data []byte) {
+				writeErr = writeFunc(data)
+			})
+			if err != nil {
+				glog.V(1).Infof("read from %s: %v", fileUrl, err)
+			} else if writeErr != nil {
+				glog.V(1).Infof("copy from %s: %v", fileUrl, writeErr)
+			} else {
+				break
+			}
+		}
+		if shouldRetry && err != nil {
+			return err
+		}
+		if writeErr != nil {
+			return writeErr
+		}
+	}
+	return nil
+}
diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go
index 7353cdc91..d7e609c68 100644
--- a/weed/replication/replicator.go
+++ b/weed/replication/replicator.go
@@ -3,8 +3,10 @@ package replication
 import (
 	"context"
 	"fmt"
-	"path/filepath"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"google.golang.org/grpc"
 	"strings"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -18,10 +20,10 @@ type Replicator struct {
 	source *source.FilerSource
 }
 
-func NewReplicator(sourceConfig util.Configuration, dataSink sink.ReplicationSink) *Replicator {
+func NewReplicator(sourceConfig util.Configuration, configPrefix string, dataSink sink.ReplicationSink) *Replicator {
 
 	source := &source.FilerSource{}
-	source.Initialize(sourceConfig)
+	source.Initialize(sourceConfig, configPrefix)
 
 	dataSink.SetSourceFiler(source)
 
@@ -32,37 +34,64 @@ func NewReplicator(sourceConfig util.Configuration, dataSink sink.ReplicationSin
 }
 
 func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_pb.EventNotification) error {
+	if message.IsFromOtherCluster && r.sink.GetName() == "filer" {
+		return nil
+	}
 	if !strings.HasPrefix(key, r.source.Dir) {
 		glog.V(4).Infof("skipping %v outside of %v", key, r.source.Dir)
 		return nil
 	}
-	newKey := filepath.ToSlash(filepath.Join(r.sink.GetSinkToDirectory(), key[len(r.source.Dir):]))
+	var dateKey string
+	if r.sink.IsIncremental() {
+		var mTime int64
+		if message.NewEntry != nil {
+			mTime = message.NewEntry.Attributes.Mtime
+		} else if message.OldEntry != nil {
+			mTime = message.OldEntry.Attributes.Mtime
+		}
+		dateKey = time.Unix(mTime, 0).Format("2006-01-02")
+	}
+	newKey := util.Join(r.sink.GetSinkToDirectory(), dateKey, key[len(r.source.Dir):])
 	glog.V(3).Infof("replicate %s => %s", key, newKey)
 	key = newKey
 	if message.OldEntry != nil && message.NewEntry == nil {
 		glog.V(4).Infof("deleting %v", key)
-		return r.sink.DeleteEntry(ctx, key, message.OldEntry.IsDirectory, message.DeleteChunks)
+		return r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
 	}
 	if message.OldEntry == nil && message.NewEntry != nil {
 		glog.V(4).Infof("creating %v", key)
-		return r.sink.CreateEntry(ctx, key, message.NewEntry)
+		return r.sink.CreateEntry(key, message.NewEntry, message.Signatures)
 	}
 	if message.OldEntry == nil && message.NewEntry == nil {
 		glog.V(0).Infof("weird message %+v", message)
 		return nil
 	}
 
-	foundExisting, err := r.sink.UpdateEntry(ctx, key, message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks)
+	foundExisting, err := r.sink.UpdateEntry(key, message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks, message.Signatures)
 	if foundExisting {
 		glog.V(4).Infof("updated %v", key)
 		return err
 	}
 
-	err = r.sink.DeleteEntry(ctx, key, message.OldEntry.IsDirectory, false)
+	err = r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, false, message.Signatures)
 	if err != nil {
 		return fmt.Errorf("delete old entry %v: %v", key, err)
 	}
 
 	glog.V(4).Infof("creating missing %v", key)
-	return r.sink.CreateEntry(ctx, key, message.NewEntry)
+	return r.sink.CreateEntry(key, message.NewEntry, message.Signatures)
+}
+
+func ReadFilerSignature(grpcDialOption grpc.DialOption, filer string) (filerSignature int32, readErr error) {
+	if readErr = pb.WithFilerClient(filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+		if resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}); err != nil {
+			return fmt.Errorf("GetFilerConfiguration %s: %v", filer, err)
+		} else {
+			filerSignature = resp.Signature
+		}
+		return nil
+	}); readErr != nil {
+		return 0, readErr
+	}
+	return filerSignature, nil
 }
diff --git a/weed/replication/sink/azuresink/azure_sink.go b/weed/replication/sink/azuresink/azure_sink.go
index 6381908a1..d13a1049b 100644
--- a/weed/replication/sink/azuresink/azure_sink.go
+++ b/weed/replication/sink/azuresink/azure_sink.go
@@ -4,11 +4,12 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
 	"net/url"
 	"strings"
 
 	"github.com/Azure/azure-storage-blob-go/azblob"
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/sink"
@@ -17,10 +18,11 @@ import (
 )
 
 type AzureSink struct {
-	containerURL azblob.ContainerURL
-	container    string
-	dir          string
-	filerSource  *source.FilerSource
+	containerURL  azblob.ContainerURL
+	container     string
+	dir           string
+	filerSource   *source.FilerSource
+	isIncremental bool
 }
 
 func init() {
@@ -35,12 +37,17 @@ func (g *AzureSink) GetSinkToDirectory() string {
 	return g.dir
 }
 
-func (g *AzureSink) Initialize(configuration util.Configuration) error {
+func (g *AzureSink) IsIncremental() bool {
+	return g.isIncremental
+}
+
+func (g *AzureSink) Initialize(configuration util.Configuration, prefix string) error {
+	g.isIncremental = configuration.GetBool(prefix + "is_incremental")
 	return g.initialize(
-		configuration.GetString("account_name"),
-		configuration.GetString("account_key"),
-		configuration.GetString("container"),
-		configuration.GetString("directory"),
+		configuration.GetString(prefix+"account_name"),
+		configuration.GetString(prefix+"account_key"),
+		configuration.GetString(prefix+"container"),
+		configuration.GetString(prefix+"directory"),
 	)
 }
 
@@ -70,7 +77,7 @@ func (g *AzureSink) initialize(accountName, accountKey, container, dir string) e
 	return nil
 }
 
-func (g *AzureSink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *AzureSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
 
 	key = cleanKey(key)
 
@@ -78,7 +85,7 @@ func (g *AzureSink) DeleteEntry(ctx context.Context, key string, isDirectory, de
 		key = key + "/"
 	}
 
-	if _, err := g.containerURL.NewBlobURL(key).Delete(ctx,
+	if _, err := g.containerURL.NewBlobURL(key).Delete(context.Background(),
 		azblob.DeleteSnapshotsOptionInclude, azblob.BlobAccessConditions{}); err != nil {
 		return fmt.Errorf("azure delete %s/%s: %v", g.container, key, err)
 	}
@@ -87,7 +94,7 @@ func (g *AzureSink) DeleteEntry(ctx context.Context, key string, isDirectory, de
 
 }
 
-func (g *AzureSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
+func (g *AzureSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 
 	key = cleanKey(key)
 
@@ -95,44 +102,32 @@ func (g *AzureSink) CreateEntry(ctx context.Context, key string, entry *filer_pb
 		return nil
 	}
 
-	totalSize := filer2.TotalSize(entry.Chunks)
-	chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize))
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(g.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
 
 	// Create a URL that references a to-be-created blob in your
 	// Azure Storage account's container.
 	appendBlobURL := g.containerURL.NewAppendBlobURL(key)
 
-	_, err := appendBlobURL.Create(ctx, azblob.BlobHTTPHeaders{}, azblob.Metadata{}, azblob.BlobAccessConditions{})
+	_, err := appendBlobURL.Create(context.Background(), azblob.BlobHTTPHeaders{}, azblob.Metadata{}, azblob.BlobAccessConditions{})
 	if err != nil {
 		return err
 	}
 
-	for _, chunk := range chunkViews {
-
-		fileUrl, err := g.filerSource.LookupFileId(ctx, chunk.FileId)
-		if err != nil {
-			return err
-		}
-
-		var writeErr error
-		_, readErr := util.ReadUrlAsStream(fileUrl, chunk.Offset, int(chunk.Size), func(data []byte) {
-			_, writeErr = appendBlobURL.AppendBlock(ctx, bytes.NewReader(data), azblob.AppendBlobAccessConditions{}, nil)
-		})
-
-		if readErr != nil {
-			return readErr
-		}
-		if writeErr != nil {
-			return writeErr
-		}
+	writeFunc := func(data []byte) error {
+		_, writeErr := appendBlobURL.AppendBlock(context.Background(), bytes.NewReader(data), azblob.AppendBlobAccessConditions{}, nil)
+		return writeErr
+	}
 
+	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
+		return err
 	}
 
 	return nil
 
 }
 
-func (g *AzureSink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *AzureSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 	key = cleanKey(key)
 	// TODO improve efficiency
 	return false, nil
diff --git a/weed/replication/sink/b2sink/b2_sink.go b/weed/replication/sink/b2sink/b2_sink.go
index 35c2230fa..90a0bb2e8 100644
--- a/weed/replication/sink/b2sink/b2_sink.go
+++ b/weed/replication/sink/b2sink/b2_sink.go
@@ -2,9 +2,10 @@ package B2Sink
 
 import (
 	"context"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
 	"strings"
 
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/sink"
 	"github.com/chrislusf/seaweedfs/weed/replication/source"
@@ -13,10 +14,11 @@ import (
 )
 
 type B2Sink struct {
-	client      *b2.Client
-	bucket      string
-	dir         string
-	filerSource *source.FilerSource
+	client        *b2.Client
+	bucket        string
+	dir           string
+	filerSource   *source.FilerSource
+	isIncremental bool
 }
 
 func init() {
@@ -31,12 +33,17 @@ func (g *B2Sink) GetSinkToDirectory() string {
 	return g.dir
 }
 
-func (g *B2Sink) Initialize(configuration util.Configuration) error {
+func (g *B2Sink) IsIncremental() bool {
+	return g.isIncremental
+}
+
+func (g *B2Sink) Initialize(configuration util.Configuration, prefix string) error {
+	g.isIncremental = configuration.GetBool(prefix + "is_incremental")
 	return g.initialize(
-		configuration.GetString("b2_account_id"),
-		configuration.GetString("b2_master_application_key"),
-		configuration.GetString("bucket"),
-		configuration.GetString("directory"),
+		configuration.GetString(prefix+"b2_account_id"),
+		configuration.GetString(prefix+"b2_master_application_key"),
+		configuration.GetString(prefix+"bucket"),
+		configuration.GetString(prefix+"directory"),
 	)
 }
 
@@ -45,8 +52,7 @@ func (g *B2Sink) SetSourceFiler(s *source.FilerSource) {
 }
 
 func (g *B2Sink) initialize(accountId, accountKey, bucket, dir string) error {
-	ctx := context.Background()
-	client, err := b2.NewClient(ctx, accountId, accountKey)
+	client, err := b2.NewClient(context.Background(), accountId, accountKey)
 	if err != nil {
 		return err
 	}
@@ -58,7 +64,7 @@ func (g *B2Sink) initialize(accountId, accountKey, bucket, dir string) error {
 	return nil
 }
 
-func (g *B2Sink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *B2Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
 
 	key = cleanKey(key)
 
@@ -66,18 +72,18 @@ func (g *B2Sink) DeleteEntry(ctx context.Context, key string, isDirectory, delet
 		key = key + "/"
 	}
 
-	bucket, err := g.client.Bucket(ctx, g.bucket)
+	bucket, err := g.client.Bucket(context.Background(), g.bucket)
 	if err != nil {
 		return err
 	}
 
 	targetObject := bucket.Object(key)
 
-	return targetObject.Delete(ctx)
+	return targetObject.Delete(context.Background())
 
 }
 
-func (g *B2Sink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
+func (g *B2Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 
 	key = cleanKey(key)
 
@@ -85,46 +91,33 @@ func (g *B2Sink) CreateEntry(ctx context.Context, key string, entry *filer_pb.En
 		return nil
 	}
 
-	totalSize := filer2.TotalSize(entry.Chunks)
-	chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize))
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(g.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
 
-	bucket, err := g.client.Bucket(ctx, g.bucket)
+	bucket, err := g.client.Bucket(context.Background(), g.bucket)
 	if err != nil {
 		return err
 	}
 
 	targetObject := bucket.Object(key)
-	writer := targetObject.NewWriter(ctx)
-
-	for _, chunk := range chunkViews {
-
-		fileUrl, err := g.filerSource.LookupFileId(ctx, chunk.FileId)
-		if err != nil {
-			return err
-		}
-
-		var writeErr error
-		_, readErr := util.ReadUrlAsStream(fileUrl, chunk.Offset, int(chunk.Size), func(data []byte) {
-			_, err := writer.Write(data)
-			if err != nil {
-				writeErr = err
-			}
-		})
-
-		if readErr != nil {
-			return readErr
-		}
-		if writeErr != nil {
-			return writeErr
-		}
+	writer := targetObject.NewWriter(context.Background())
 
+	writeFunc := func(data []byte) error {
+		_, writeErr := writer.Write(data)
+		return writeErr
 	}
 
-	return writer.Close()
+	defer writer.Close()
+
+	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
+		return err
+	}
+
+	return nil
 
 }
 
-func (g *B2Sink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *B2Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 
 	key = cleanKey(key)
 
diff --git a/weed/replication/sink/filersink/README.txt b/weed/replication/sink/filersink/README.txt
new file mode 100644
index 000000000..4ba0fc752
--- /dev/null
+++ b/weed/replication/sink/filersink/README.txt
@@ -0,0 +1,12 @@
+How replication works
+======
+
+All metadata changes within current cluster would be notified to a message queue.
+
+If the meta data change is from other clusters, this metadata would change would not be notified to the message queue.
+
+So active<=>active replication is possible.
+
+
+All metadata changes would be published as metadata changes.
+So all mounts listening for metadata changes will get updated.
\ No newline at end of file
diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go
index 97e9671a3..a7392d856 100644
--- a/weed/replication/sink/filersink/fetch_write.go
+++ b/weed/replication/sink/filersink/fetch_write.go
@@ -3,41 +3,46 @@ package filersink
 import (
 	"context"
 	"fmt"
-	"google.golang.org/grpc"
-	"strings"
+	"github.com/chrislusf/seaweedfs/weed/util"
 	"sync"
 
+	"google.golang.org/grpc"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
+	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/security"
-	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk) (replicatedChunks []*filer_pb.FileChunk, err error) {
+func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path string) (replicatedChunks []*filer_pb.FileChunk, err error) {
 	if len(sourceChunks) == 0 {
 		return
 	}
+
+	replicatedChunks = make([]*filer_pb.FileChunk, len(sourceChunks))
+
 	var wg sync.WaitGroup
-	for _, sourceChunk := range sourceChunks {
+	for chunkIndex, sourceChunk := range sourceChunks {
 		wg.Add(1)
-		go func(chunk *filer_pb.FileChunk) {
+		go func(chunk *filer_pb.FileChunk, index int) {
 			defer wg.Done()
-			replicatedChunk, e := fs.replicateOneChunk(ctx, chunk)
+			replicatedChunk, e := fs.replicateOneChunk(chunk, path)
 			if e != nil {
 				err = e
+				return
 			}
-			replicatedChunks = append(replicatedChunks, replicatedChunk)
-		}(sourceChunk)
+			replicatedChunks[index] = replicatedChunk
+		}(sourceChunk, chunkIndex)
 	}
 	wg.Wait()
 
 	return
 }
 
-func (fs *FilerSink) replicateOneChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk) (*filer_pb.FileChunk, error) {
+func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string) (*filer_pb.FileChunk, error) {
 
-	fileId, err := fs.fetchAndWrite(ctx, sourceChunk)
+	fileId, err := fs.fetchAndWrite(sourceChunk, path)
 	if err != nil {
 		return nil, fmt.Errorf("copy %s: %v", sourceChunk.GetFileIdString(), err)
 	}
@@ -49,21 +54,23 @@ func (fs *FilerSink) replicateOneChunk(ctx context.Context, sourceChunk *filer_p
 		Mtime:        sourceChunk.Mtime,
 		ETag:         sourceChunk.ETag,
 		SourceFileId: sourceChunk.GetFileIdString(),
+		CipherKey:    sourceChunk.CipherKey,
+		IsCompressed: sourceChunk.IsCompressed,
 	}, nil
 }
 
-func (fs *FilerSink) fetchAndWrite(ctx context.Context, sourceChunk *filer_pb.FileChunk) (fileId string, err error) {
+func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string) (fileId string, err error) {
 
-	filename, header, readCloser, err := fs.filerSource.ReadPart(ctx, sourceChunk.GetFileIdString())
+	filename, header, resp, err := fs.filerSource.ReadPart(sourceChunk.GetFileIdString())
 	if err != nil {
 		return "", fmt.Errorf("read part %s: %v", sourceChunk.GetFileIdString(), err)
 	}
-	defer readCloser.Close()
+	defer util.CloseResponse(resp)
 
 	var host string
 	var auth security.EncodedJwt
 
-	if err := fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	if err := fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.AssignVolumeRequest{
 			Count:       1,
@@ -71,13 +78,18 @@ func (fs *FilerSink) fetchAndWrite(ctx context.Context, sourceChunk *filer_pb.Fi
 			Collection:  fs.collection,
 			TtlSec:      fs.ttlSec,
 			DataCenter:  fs.dataCenter,
+			DiskType:    fs.diskType,
+			Path:        path,
 		}
 
-		resp, err := client.AssignVolume(ctx, request)
+		resp, err := client.AssignVolume(context.Background(), request)
 		if err != nil {
 			glog.V(0).Infof("assign volume failure %v: %v", request, err)
 			return err
 		}
+		if resp.Error != "" {
+			return fmt.Errorf("assign volume failure %v: %v", request, resp.Error)
+		}
 
 		fileId, host, auth = resp.FileId, resp.Url, security.EncodedJwt(resp.Auth)
 
@@ -87,13 +99,16 @@ func (fs *FilerSink) fetchAndWrite(ctx context.Context, sourceChunk *filer_pb.Fi
 	}
 
 	fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
+	if fs.writeChunkByFiler {
+		fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId)
+	}
 
 	glog.V(4).Infof("replicating %s to %s header:%+v", filename, fileUrl, header)
 
-	uploadResult, err := operation.Upload(fileUrl, filename, readCloser,
-		"gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
+	// fetch data as is, regardless whether it is encrypted or not
+	uploadResult, err, _ := operation.Upload(fileUrl, filename, false, resp.Body, "gzip" == header.Get("Content-Encoding"), header.Get("Content-Type"), nil, auth)
 	if err != nil {
-		glog.V(0).Infof("upload data %v to %s: %v", filename, fileUrl, err)
+		glog.V(0).Infof("upload source data %v to %s: %v", sourceChunk.GetFileIdString(), fileUrl, err)
 		return "", fmt.Errorf("upload data: %v", err)
 	}
 	if uploadResult.Error != "" {
@@ -104,19 +119,16 @@ func (fs *FilerSink) fetchAndWrite(ctx context.Context, sourceChunk *filer_pb.Fi
 	return
 }
 
-func (fs *FilerSink) withFilerClient(ctx context.Context, fn func(filer_pb.SeaweedFilerClient) error) error {
+var _ = filer_pb.FilerClient(&FilerSink{})
 
-	return util.WithCachedGrpcClient(ctx, func(grpcConnection *grpc.ClientConn) error {
+func (fs *FilerSink) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+
+	return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, fs.grpcAddress, fs.grpcDialOption)
 
 }
-
-func volumeId(fileId string) string {
-	lastCommaIndex := strings.LastIndex(fileId, ",")
-	if lastCommaIndex > 0 {
-		return fileId[:lastCommaIndex]
-	}
-	return fileId
+func (fs *FilerSink) AdjustedUrl(location *filer_pb.Location) string {
+	return location.Url
 }
diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go
index f99c7fdf6..d7c5fccc3 100644
--- a/weed/replication/sink/filersink/filer_sink.go
+++ b/weed/replication/sink/filersink/filer_sink.go
@@ -3,11 +3,14 @@ package filersink
 import (
 	"context"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/security"
-	"github.com/spf13/viper"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/wdclient"
+
 	"google.golang.org/grpc"
 
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/security"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/sink"
@@ -16,14 +19,18 @@ import (
 )
 
 type FilerSink struct {
-	filerSource    *source.FilerSource
-	grpcAddress    string
-	dir            string
-	replication    string
-	collection     string
-	ttlSec         int32
-	dataCenter     string
-	grpcDialOption grpc.DialOption
+	filerSource       *source.FilerSource
+	grpcAddress       string
+	dir               string
+	replication       string
+	collection        string
+	ttlSec            int32
+	diskType          string
+	dataCenter        string
+	grpcDialOption    grpc.DialOption
+	address           string
+	writeChunkByFiler bool
+	isIncremental     bool
 }
 
 func init() {
@@ -38,58 +45,63 @@ func (fs *FilerSink) GetSinkToDirectory() string {
 	return fs.dir
 }
 
-func (fs *FilerSink) Initialize(configuration util.Configuration) error {
-	return fs.initialize(
-		configuration.GetString("grpcAddress"),
-		configuration.GetString("directory"),
-		configuration.GetString("replication"),
-		configuration.GetString("collection"),
-		configuration.GetInt("ttlSec"),
-	)
+func (fs *FilerSink) IsIncremental() bool {
+	return fs.isIncremental
+}
+
+func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
+	fs.isIncremental = configuration.GetBool(prefix + "is_incremental")
+	return fs.DoInitialize(
+		"",
+		configuration.GetString(prefix+"grpcAddress"),
+		configuration.GetString(prefix+"directory"),
+		configuration.GetString(prefix+"replication"),
+		configuration.GetString(prefix+"collection"),
+		configuration.GetInt(prefix+"ttlSec"),
+		configuration.GetString(prefix+"disk"),
+		security.LoadClientTLS(util.GetViper(), "grpc.client"),
+		false)
 }
 
 func (fs *FilerSink) SetSourceFiler(s *source.FilerSource) {
 	fs.filerSource = s
 }
 
-func (fs *FilerSink) initialize(grpcAddress string, dir string,
-	replication string, collection string, ttlSec int) (err error) {
+func (fs *FilerSink) DoInitialize(address, grpcAddress string, dir string,
+	replication string, collection string, ttlSec int, diskType string, grpcDialOption grpc.DialOption, writeChunkByFiler bool) (err error) {
+	fs.address = address
+	if fs.address == "" {
+		fs.address = pb.GrpcAddressToServerAddress(grpcAddress)
+	}
 	fs.grpcAddress = grpcAddress
 	fs.dir = dir
 	fs.replication = replication
 	fs.collection = collection
 	fs.ttlSec = int32(ttlSec)
-	fs.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client")
+	fs.diskType = diskType
+	fs.grpcDialOption = grpcDialOption
+	fs.writeChunkByFiler = writeChunkByFiler
 	return nil
 }
 
-func (fs *FilerSink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
-	return fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
-
-		dir, name := filer2.FullPath(key).DirAndName()
-
-		request := &filer_pb.DeleteEntryRequest{
-			Directory:    dir,
-			Name:         name,
-			IsDeleteData: deleteIncludeChunks,
-		}
+func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
 
-		glog.V(1).Infof("delete entry: %v", request)
-		_, err := client.DeleteEntry(ctx, request)
-		if err != nil {
-			glog.V(0).Infof("delete entry %s: %v", key, err)
-			return fmt.Errorf("delete entry %s: %v", key, err)
-		}
+	dir, name := util.FullPath(key).DirAndName()
 
-		return nil
-	})
+	glog.V(4).Infof("delete entry: %v", key)
+	err := filer_pb.Remove(fs, dir, name, deleteIncludeChunks, true, true, true, signatures)
+	if err != nil {
+		glog.V(0).Infof("delete entry %s: %v", key, err)
+		return fmt.Errorf("delete entry %s: %v", key, err)
+	}
+	return nil
 }
 
-func (fs *FilerSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
+func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 
-	return fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	return fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
-		dir, name := filer2.FullPath(key).DirAndName()
+		dir, name := util.FullPath(key).DirAndName()
 
 		// look up existing entry
 		lookupRequest := &filer_pb.LookupDirectoryEntryRequest{
@@ -97,21 +109,21 @@ func (fs *FilerSink) CreateEntry(ctx context.Context, key string, entry *filer_p
 			Name:      name,
 		}
 		glog.V(1).Infof("lookup: %v", lookupRequest)
-		if resp, err := client.LookupDirectoryEntry(ctx, lookupRequest); err == nil {
-			if filer2.ETag(resp.Entry.Chunks) == filer2.ETag(entry.Chunks) {
-				glog.V(0).Infof("already replicated %s", key)
+		if resp, err := filer_pb.LookupEntry(client, lookupRequest); err == nil {
+			if filer.ETag(resp.Entry) == filer.ETag(entry) {
+				glog.V(3).Infof("already replicated %s", key)
 				return nil
 			}
 		}
 
-		replicatedChunks, err := fs.replicateChunks(ctx, entry.Chunks)
+		replicatedChunks, err := fs.replicateChunks(entry.Chunks, key)
 
 		if err != nil {
-			glog.V(0).Infof("replicate entry chunks %s: %v", key, err)
-			return fmt.Errorf("replicate entry chunks %s: %v", key, err)
+			// only warning here since the source chunk may have been deleted already
+			glog.Warningf("replicate entry chunks %s: %v", key, err)
 		}
 
-		glog.V(0).Infof("replicated %s %+v ===> %+v", key, entry.Chunks, replicatedChunks)
+		glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.Chunks, replicatedChunks)
 
 		request := &filer_pb.CreateEntryRequest{
 			Directory: dir,
@@ -120,11 +132,14 @@ func (fs *FilerSink) CreateEntry(ctx context.Context, key string, entry *filer_p
 				IsDirectory: entry.IsDirectory,
 				Attributes:  entry.Attributes,
 				Chunks:      replicatedChunks,
+				Content:     entry.Content,
 			},
+			IsFromOtherCluster: true,
+			Signatures:         signatures,
 		}
 
-		glog.V(1).Infof("create: %v", request)
-		if _, err := client.CreateEntry(ctx, request); err != nil {
+		glog.V(3).Infof("create: %v", request)
+		if err := filer_pb.CreateEntry(client, request); err != nil {
 			glog.V(0).Infof("create entry %s: %v", key, err)
 			return fmt.Errorf("create entry %s: %v", key, err)
 		}
@@ -133,13 +148,13 @@ func (fs *FilerSink) CreateEntry(ctx context.Context, key string, entry *filer_p
 	})
 }
 
-func (fs *FilerSink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 
-	dir, name := filer2.FullPath(key).DirAndName()
+	dir, name := util.FullPath(key).DirAndName()
 
 	// read existing entry
 	var existingEntry *filer_pb.Entry
-	err = fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	err = fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.LookupDirectoryEntryRequest{
 			Directory: dir,
@@ -147,7 +162,7 @@ func (fs *FilerSink) UpdateEntry(ctx context.Context, key string, oldEntry *file
 		}
 
 		glog.V(4).Infof("lookup entry: %v", request)
-		resp, err := client.LookupDirectoryEntry(ctx, request)
+		resp, err := filer_pb.LookupEntry(client, request)
 		if err != nil {
 			glog.V(0).Infof("lookup %s: %v", key, err)
 			return err
@@ -162,28 +177,31 @@ func (fs *FilerSink) UpdateEntry(ctx context.Context, key string, oldEntry *file
 		return false, fmt.Errorf("lookup %s: %v", key, err)
 	}
 
-	glog.V(0).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
+	glog.V(4).Infof("oldEntry %+v, newEntry %+v, existingEntry: %+v", oldEntry, newEntry, existingEntry)
 
 	if existingEntry.Attributes.Mtime > newEntry.Attributes.Mtime {
 		// skip if already changed
 		// this usually happens when the messages are not ordered
-		glog.V(0).Infof("late updates %s", key)
-	} else if filer2.ETag(newEntry.Chunks) == filer2.ETag(existingEntry.Chunks) {
+		glog.V(2).Infof("late updates %s", key)
+	} else if filer.ETag(newEntry) == filer.ETag(existingEntry) {
 		// skip if no change
 		// this usually happens when retrying the replication
-		glog.V(0).Infof("already replicated %s", key)
+		glog.V(3).Infof("already replicated %s", key)
 	} else {
 		// find out what changed
-		deletedChunks, newChunks := compareChunks(oldEntry, newEntry)
+		deletedChunks, newChunks, err := compareChunks(filer.LookupFn(fs), oldEntry, newEntry)
+		if err != nil {
+			return true, fmt.Errorf("replicte %s compare chunks error: %v", key, err)
+		}
 
 		// delete the chunks that are deleted from the source
 		if deleteIncludeChunks {
 			// remove the deleted chunks. Actual data deletion happens in filer UpdateEntry FindUnusedFileChunks
-			existingEntry.Chunks = filer2.MinusChunks(existingEntry.Chunks, deletedChunks)
+			existingEntry.Chunks = filer.DoMinusChunks(existingEntry.Chunks, deletedChunks)
 		}
 
 		// replicate the chunks that are new in the source
-		replicatedChunks, err := fs.replicateChunks(ctx, newChunks)
+		replicatedChunks, err := fs.replicateChunks(newChunks, key)
 		if err != nil {
 			return true, fmt.Errorf("replicte %s chunks error: %v", key, err)
 		}
@@ -191,14 +209,16 @@ func (fs *FilerSink) UpdateEntry(ctx context.Context, key string, oldEntry *file
 	}
 
 	// save updated meta data
-	return true, fs.withFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
+	return true, fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		request := &filer_pb.UpdateEntryRequest{
-			Directory: newParentPath,
-			Entry:     existingEntry,
+			Directory:          newParentPath,
+			Entry:              existingEntry,
+			IsFromOtherCluster: true,
+			Signatures:         signatures,
 		}
 
-		if _, err := client.UpdateEntry(ctx, request); err != nil {
+		if _, err := client.UpdateEntry(context.Background(), request); err != nil {
 			return fmt.Errorf("update existingEntry %s: %v", key, err)
 		}
 
@@ -206,8 +226,21 @@ func (fs *FilerSink) UpdateEntry(ctx context.Context, key string, oldEntry *file
 	})
 
 }
-func compareChunks(oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk) {
-	deletedChunks = filer2.MinusChunks(oldEntry.Chunks, newEntry.Chunks)
-	newChunks = filer2.MinusChunks(newEntry.Chunks, oldEntry.Chunks)
+func compareChunks(lookupFileIdFn wdclient.LookupFileIdFunctionType, oldEntry, newEntry *filer_pb.Entry) (deletedChunks, newChunks []*filer_pb.FileChunk, err error) {
+	aData, aMeta, aErr := filer.ResolveChunkManifest(lookupFileIdFn, oldEntry.Chunks)
+	if aErr != nil {
+		return nil, nil, aErr
+	}
+	bData, bMeta, bErr := filer.ResolveChunkManifest(lookupFileIdFn, newEntry.Chunks)
+	if bErr != nil {
+		return nil, nil, bErr
+	}
+
+	deletedChunks = append(deletedChunks, filer.DoMinusChunks(aData, bData)...)
+	deletedChunks = append(deletedChunks, filer.DoMinusChunks(aMeta, bMeta)...)
+
+	newChunks = append(newChunks, filer.DoMinusChunks(bData, aData)...)
+	newChunks = append(newChunks, filer.DoMinusChunks(bMeta, aMeta)...)
+
 	return
 }
diff --git a/weed/replication/sink/gcssink/gcs_sink.go b/weed/replication/sink/gcssink/gcs_sink.go
index abd7c49b9..5cf5b7317 100644
--- a/weed/replication/sink/gcssink/gcs_sink.go
+++ b/weed/replication/sink/gcssink/gcs_sink.go
@@ -3,23 +3,26 @@ package gcssink
 import (
 	"context"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
 	"os"
 
 	"cloud.google.com/go/storage"
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"google.golang.org/api/option"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/sink"
 	"github.com/chrislusf/seaweedfs/weed/replication/source"
 	"github.com/chrislusf/seaweedfs/weed/util"
-	"google.golang.org/api/option"
 )
 
 type GcsSink struct {
-	client      *storage.Client
-	bucket      string
-	dir         string
-	filerSource *source.FilerSource
+	client        *storage.Client
+	bucket        string
+	dir           string
+	filerSource   *source.FilerSource
+	isIncremental bool
 }
 
 func init() {
@@ -34,11 +37,16 @@ func (g *GcsSink) GetSinkToDirectory() string {
 	return g.dir
 }
 
-func (g *GcsSink) Initialize(configuration util.Configuration) error {
+func (g *GcsSink) IsIncremental() bool {
+	return g.isIncremental
+}
+
+func (g *GcsSink) Initialize(configuration util.Configuration, prefix string) error {
+	g.isIncremental = configuration.GetBool(prefix + "is_incremental")
 	return g.initialize(
-		configuration.GetString("google_application_credentials"),
-		configuration.GetString("bucket"),
-		configuration.GetString("directory"),
+		configuration.GetString(prefix+"google_application_credentials"),
+		configuration.GetString(prefix+"bucket"),
+		configuration.GetString(prefix+"directory"),
 	)
 }
 
@@ -50,7 +58,6 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
 	g.bucket = bucketName
 	g.dir = dir
 
-	ctx := context.Background()
 	// Creates a client.
 	if google_application_credentials == "" {
 		var found bool
@@ -59,7 +66,7 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
 			glog.Fatalf("need to specific GOOGLE_APPLICATION_CREDENTIALS env variable or google_application_credentials in replication.toml")
 		}
 	}
-	client, err := storage.NewClient(ctx, option.WithCredentialsFile(google_application_credentials))
+	client, err := storage.NewClient(context.Background(), option.WithCredentialsFile(google_application_credentials))
 	if err != nil {
 		glog.Fatalf("Failed to create client: %v", err)
 	}
@@ -69,13 +76,13 @@ func (g *GcsSink) initialize(google_application_credentials, bucketName, dir str
 	return nil
 }
 
-func (g *GcsSink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
+func (g *GcsSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
 
 	if isDirectory {
 		key = key + "/"
 	}
 
-	if err := g.client.Bucket(g.bucket).Object(key).Delete(ctx); err != nil {
+	if err := g.client.Bucket(g.bucket).Object(key).Delete(context.Background()); err != nil {
 		return fmt.Errorf("gcs delete %s%s: %v", g.bucket, key, err)
 	}
 
@@ -83,35 +90,24 @@ func (g *GcsSink) DeleteEntry(ctx context.Context, key string, isDirectory, dele
 
 }
 
-func (g *GcsSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
+func (g *GcsSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 
 	if entry.IsDirectory {
 		return nil
 	}
 
-	totalSize := filer2.TotalSize(entry.Chunks)
-	chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize))
-
-	wc := g.client.Bucket(g.bucket).Object(key).NewWriter(ctx)
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(g.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
 
-	for _, chunk := range chunkViews {
-
-		fileUrl, err := g.filerSource.LookupFileId(ctx, chunk.FileId)
-		if err != nil {
-			return err
-		}
-
-		_, err = util.ReadUrlAsStream(fileUrl, chunk.Offset, int(chunk.Size), func(data []byte) {
-			wc.Write(data)
-		})
-
-		if err != nil {
-			return err
-		}
+	wc := g.client.Bucket(g.bucket).Object(key).NewWriter(context.Background())
+	defer wc.Close()
 
+	writeFunc := func(data []byte) error {
+		_, writeErr := wc.Write(data)
+		return writeErr
 	}
 
-	if err := wc.Close(); err != nil {
+	if err := repl_util.CopyFromChunkViews(chunkViews, g.filerSource, writeFunc); err != nil {
 		return err
 	}
 
@@ -119,7 +115,7 @@ func (g *GcsSink) CreateEntry(ctx context.Context, key string, entry *filer_pb.E
 
 }
 
-func (g *GcsSink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (g *GcsSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 	// TODO improve efficiency
 	return false, nil
 }
diff --git a/weed/replication/sink/localsink/local_incremental_sink.go b/weed/replication/sink/localsink/local_incremental_sink.go
new file mode 100644
index 000000000..a1d49e28a
--- /dev/null
+++ b/weed/replication/sink/localsink/local_incremental_sink.go
@@ -0,0 +1,17 @@
+package localsink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+)
+
+type LocalIncSink struct {
+	LocalSink
+}
+
+func (localincsink *LocalIncSink) GetName() string {
+	return "local_incremental"
+}
+
+func init() {
+	sink.Sinks = append(sink.Sinks, &LocalIncSink{})
+}
diff --git a/weed/replication/sink/localsink/local_sink.go b/weed/replication/sink/localsink/local_sink.go
new file mode 100644
index 000000000..2b9b3e69a
--- /dev/null
+++ b/weed/replication/sink/localsink/local_sink.go
@@ -0,0 +1,105 @@
+package localsink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/replication/repl_util"
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+	"github.com/chrislusf/seaweedfs/weed/replication/source"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+type LocalSink struct {
+	Dir         string
+	filerSource *source.FilerSource
+}
+
+func init() {
+	sink.Sinks = append(sink.Sinks, &LocalSink{})
+}
+
+func (localsink *LocalSink) SetSourceFiler(s *source.FilerSource) {
+	localsink.filerSource = s
+}
+
+func (localsink *LocalSink) GetName() string {
+	return "local"
+}
+
+func (localsink *LocalSink) isMultiPartEntry(key string) bool {
+	return strings.HasSuffix(key, ".part") && strings.Contains(key, "/.uploads/")
+}
+
+func (localsink *LocalSink) initialize(dir string) error {
+	localsink.Dir = dir
+	return nil
+}
+
+func (localsink *LocalSink) Initialize(configuration util.Configuration, prefix string) error {
+	dir := configuration.GetString(prefix + "directory")
+	glog.V(4).Infof("sink.local.directory: %v", dir)
+	return localsink.initialize(dir)
+}
+
+func (localsink *LocalSink) GetSinkToDirectory() string {
+	return localsink.Dir
+}
+
+func (localsink *LocalSink) IsIncremental() bool {
+	return true
+}
+
+func (localsink *LocalSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
+	if localsink.isMultiPartEntry(key) {
+		return nil
+	}
+	glog.V(4).Infof("Delete Entry key: %s", key)
+	if err := os.Remove(key); err != nil {
+		glog.V(0).Infof("remove entry key %s: %s", key, err)
+	}
+	return nil
+}
+
+func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
+	if entry.IsDirectory || localsink.isMultiPartEntry(key) {
+		return nil
+	}
+	glog.V(4).Infof("Create Entry key: %s", key)
+
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(localsink.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
+
+	dir := filepath.Dir(key)
+
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		glog.V(4).Infof("Create Direcotry key: %s", dir)
+		if err = os.MkdirAll(dir, 0755); err != nil {
+			return err
+		}
+	}
+
+	writeFunc := func(data []byte) error {
+		writeErr := ioutil.WriteFile(key, data, 0755)
+		return writeErr
+	}
+
+	if err := repl_util.CopyFromChunkViews(chunkViews, localsink.filerSource, writeFunc); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (localsink *LocalSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
+	if localsink.isMultiPartEntry(key) {
+		return true, nil
+	}
+	glog.V(4).Infof("Update Entry key: %s", key)
+	// do delete and create
+	return false, nil
+}
diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go
index dd54f0005..4ffd09462 100644
--- a/weed/replication/sink/replication_sink.go
+++ b/weed/replication/sink/replication_sink.go
@@ -1,7 +1,6 @@
 package sink
 
 import (
-	"context"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/source"
 	"github.com/chrislusf/seaweedfs/weed/util"
@@ -9,12 +8,13 @@ import (
 
 type ReplicationSink interface {
 	GetName() string
-	Initialize(configuration util.Configuration) error
-	DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error
-	CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error
-	UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error)
+	Initialize(configuration util.Configuration, prefix string) error
+	DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error
+	CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error
+	UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error)
 	GetSinkToDirectory() string
 	SetSourceFiler(s *source.FilerSource)
+	IsIncremental() bool
 }
 
 var (
diff --git a/weed/replication/sink/s3sink/s3_sink.go b/weed/replication/sink/s3sink/s3_sink.go
index 4cff341d0..9a36573e3 100644
--- a/weed/replication/sink/s3sink/s3_sink.go
+++ b/weed/replication/sink/s3sink/s3_sink.go
@@ -11,7 +11,8 @@ import (
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/replication/sink"
@@ -20,11 +21,13 @@ import (
 )
 
 type S3Sink struct {
-	conn        s3iface.S3API
-	region      string
-	bucket      string
-	dir         string
-	filerSource *source.FilerSource
+	conn          s3iface.S3API
+	region        string
+	bucket        string
+	dir           string
+	endpoint      string
+	filerSource   *source.FilerSource
+	isIncremental bool
 }
 
 func init() {
@@ -39,16 +42,24 @@ func (s3sink *S3Sink) GetSinkToDirectory() string {
 	return s3sink.dir
 }
 
-func (s3sink *S3Sink) Initialize(configuration util.Configuration) error {
-	glog.V(0).Infof("sink.s3.region: %v", configuration.GetString("region"))
-	glog.V(0).Infof("sink.s3.bucket: %v", configuration.GetString("bucket"))
-	glog.V(0).Infof("sink.s3.directory: %v", configuration.GetString("directory"))
+func (s3sink *S3Sink) IsIncremental() bool {
+	return s3sink.isIncremental
+}
+
+func (s3sink *S3Sink) Initialize(configuration util.Configuration, prefix string) error {
+	glog.V(0).Infof("sink.s3.region: %v", configuration.GetString(prefix+"region"))
+	glog.V(0).Infof("sink.s3.bucket: %v", configuration.GetString(prefix+"bucket"))
+	glog.V(0).Infof("sink.s3.directory: %v", configuration.GetString(prefix+"directory"))
+	glog.V(0).Infof("sink.s3.endpoint: %v", configuration.GetString(prefix+"endpoint"))
+	glog.V(0).Infof("sink.s3.is_incremental: %v", configuration.GetString(prefix+"is_incremental"))
+	s3sink.isIncremental = configuration.GetBool(prefix + "is_incremental")
 	return s3sink.initialize(
-		configuration.GetString("aws_access_key_id"),
-		configuration.GetString("aws_secret_access_key"),
-		configuration.GetString("region"),
-		configuration.GetString("bucket"),
-		configuration.GetString("directory"),
+		configuration.GetString(prefix+"aws_access_key_id"),
+		configuration.GetString(prefix+"aws_secret_access_key"),
+		configuration.GetString(prefix+"region"),
+		configuration.GetString(prefix+"bucket"),
+		configuration.GetString(prefix+"directory"),
+		configuration.GetString(prefix+"endpoint"),
 	)
 }
 
@@ -56,13 +67,16 @@ func (s3sink *S3Sink) SetSourceFiler(s *source.FilerSource) {
 	s3sink.filerSource = s
 }
 
-func (s3sink *S3Sink) initialize(awsAccessKeyId, awsSecretAccessKey, region, bucket, dir string) error {
+func (s3sink *S3Sink) initialize(awsAccessKeyId, awsSecretAccessKey, region, bucket, dir, endpoint string) error {
 	s3sink.region = region
 	s3sink.bucket = bucket
 	s3sink.dir = dir
+	s3sink.endpoint = endpoint
 
 	config := &aws.Config{
-		Region: aws.String(s3sink.region),
+		Region:           aws.String(s3sink.region),
+		Endpoint:         aws.String(s3sink.endpoint),
+		S3ForcePathStyle: aws.Bool(true),
 	}
 	if awsAccessKeyId != "" && awsSecretAccessKey != "" {
 		config.Credentials = credentials.NewStaticCredentials(awsAccessKeyId, awsSecretAccessKey, "")
@@ -77,7 +91,7 @@ func (s3sink *S3Sink) initialize(awsAccessKeyId, awsSecretAccessKey, region, buc
 	return nil
 }
 
-func (s3sink *S3Sink) DeleteEntry(ctx context.Context, key string, isDirectory, deleteIncludeChunks bool) error {
+func (s3sink *S3Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
 
 	key = cleanKey(key)
 
@@ -89,8 +103,7 @@ func (s3sink *S3Sink) DeleteEntry(ctx context.Context, key string, isDirectory,
 
 }
 
-func (s3sink *S3Sink) CreateEntry(ctx context.Context, key string, entry *filer_pb.Entry) error {
-
+func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 	key = cleanKey(key)
 
 	if entry.IsDirectory {
@@ -99,38 +112,40 @@ func (s3sink *S3Sink) CreateEntry(ctx context.Context, key string, entry *filer_
 
 	uploadId, err := s3sink.createMultipartUpload(key, entry)
 	if err != nil {
-		return err
+		return fmt.Errorf("createMultipartUpload: %v", err)
 	}
 
-	totalSize := filer2.TotalSize(entry.Chunks)
-	chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize))
+	totalSize := filer.FileSize(entry)
+	chunkViews := filer.ViewFromChunks(s3sink.filerSource.LookupFileId, entry.Chunks, 0, int64(totalSize))
+
+	parts := make([]*s3.CompletedPart, len(chunkViews))
 
-	var parts []*s3.CompletedPart
 	var wg sync.WaitGroup
 	for chunkIndex, chunk := range chunkViews {
 		partId := chunkIndex + 1
 		wg.Add(1)
-		go func(chunk *filer2.ChunkView) {
+		go func(chunk *filer.ChunkView, index int) {
 			defer wg.Done()
-			if part, uploadErr := s3sink.uploadPart(ctx, key, uploadId, partId, chunk); uploadErr != nil {
+			if part, uploadErr := s3sink.uploadPart(key, uploadId, partId, chunk); uploadErr != nil {
 				err = uploadErr
+				glog.Errorf("uploadPart: %v", uploadErr)
 			} else {
-				parts = append(parts, part)
+				parts[index] = part
 			}
-		}(chunk)
+		}(chunk, chunkIndex)
 	}
 	wg.Wait()
 
 	if err != nil {
 		s3sink.abortMultipartUpload(key, uploadId)
-		return err
+		return fmt.Errorf("uploadPart: %v", err)
 	}
 
-	return s3sink.completeMultipartUpload(ctx, key, uploadId, parts)
+	return s3sink.completeMultipartUpload(context.Background(), key, uploadId, parts)
 
 }
 
-func (s3sink *S3Sink) UpdateEntry(ctx context.Context, key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool) (foundExistingEntry bool, err error) {
+func (s3sink *S3Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) {
 	key = cleanKey(key)
 	// TODO improve efficiency
 	return false, nil
diff --git a/weed/replication/sink/s3sink/s3_write.go b/weed/replication/sink/s3sink/s3_write.go
index 0a190b27d..3dde52616 100644
--- a/weed/replication/sink/s3sink/s3_write.go
+++ b/weed/replication/sink/s3sink/s3_write.go
@@ -9,7 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/service/s3"
-	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
@@ -24,7 +24,7 @@ func (s3sink *S3Sink) deleteObject(key string) error {
 	result, err := s3sink.conn.DeleteObject(input)
 
 	if err == nil {
-		glog.V(0).Infof("[%s] delete %s: %v", s3sink.bucket, key, result)
+		glog.V(2).Infof("[%s] delete %s: %v", s3sink.bucket, key, result)
 	} else {
 		glog.Errorf("[%s] delete %s: %v", s3sink.bucket, key, err)
 	}
@@ -43,7 +43,7 @@ func (s3sink *S3Sink) createMultipartUpload(key string, entry *filer_pb.Entry) (
 	result, err := s3sink.conn.CreateMultipartUpload(input)
 
 	if err == nil {
-		glog.V(0).Infof("[%s] createMultipartUpload %s: %v", s3sink.bucket, key, result)
+		glog.V(2).Infof("[%s] createMultipartUpload %s: %v", s3sink.bucket, key, result)
 	} else {
 		glog.Errorf("[%s] createMultipartUpload %s: %v", s3sink.bucket, key, err)
 		return "", err
@@ -94,19 +94,20 @@ func (s3sink *S3Sink) completeMultipartUpload(ctx context.Context, key, uploadId
 
 	result, err := s3sink.conn.CompleteMultipartUpload(input)
 	if err == nil {
-		glog.V(0).Infof("[%s] completeMultipartUpload %s: %v", s3sink.bucket, key, result)
+		glog.V(2).Infof("[%s] completeMultipartUpload %s: %v", s3sink.bucket, key, result)
 	} else {
 		glog.Errorf("[%s] completeMultipartUpload %s: %v", s3sink.bucket, key, err)
+		return fmt.Errorf("[%s] completeMultipartUpload %s: %v", s3sink.bucket, key, err)
 	}
 
-	return err
+	return nil
 }
 
 // To upload a part
-func (s3sink *S3Sink) uploadPart(ctx context.Context, key, uploadId string, partId int, chunk *filer2.ChunkView) (*s3.CompletedPart, error) {
+func (s3sink *S3Sink) uploadPart(key, uploadId string, partId int, chunk *filer.ChunkView) (*s3.CompletedPart, error) {
 	var readSeeker io.ReadSeeker
 
-	readSeeker, err := s3sink.buildReadSeeker(ctx, chunk)
+	readSeeker, err := s3sink.buildReadSeeker(chunk)
 	if err != nil {
 		glog.Errorf("[%s] uploadPart %s %d read: %v", s3sink.bucket, key, partId, err)
 		return nil, fmt.Errorf("[%s] uploadPart %s %d read: %v", s3sink.bucket, key, partId, err)
@@ -122,7 +123,7 @@ func (s3sink *S3Sink) uploadPart(ctx context.Context, key, uploadId string, part
 
 	result, err := s3sink.conn.UploadPart(input)
 	if err == nil {
-		glog.V(0).Infof("[%s] uploadPart %s %d upload: %v", s3sink.bucket, key, partId, result)
+		glog.V(2).Infof("[%s] uploadPart %s %d upload: %v", s3sink.bucket, key, partId, result)
 	} else {
 		glog.Errorf("[%s] uploadPart %s %d upload: %v", s3sink.bucket, key, partId, err)
 	}
@@ -156,12 +157,19 @@ func (s3sink *S3Sink) uploadPartCopy(key, uploadId string, partId int64, copySou
 	return err
 }
 
-func (s3sink *S3Sink) buildReadSeeker(ctx context.Context, chunk *filer2.ChunkView) (io.ReadSeeker, error) {
-	fileUrl, err := s3sink.filerSource.LookupFileId(ctx, chunk.FileId)
+func (s3sink *S3Sink) buildReadSeeker(chunk *filer.ChunkView) (io.ReadSeeker, error) {
+	fileUrls, err := s3sink.filerSource.LookupFileId(chunk.FileId)
 	if err != nil {
 		return nil, err
 	}
 	buf := make([]byte, chunk.Size)
-	util.ReadUrl(fileUrl, chunk.Offset, int(chunk.Size), buf, true)
+	for _, fileUrl := range fileUrls {
+		_, err = util.ReadUrl(fileUrl, chunk.CipherKey, chunk.IsGzipped, false, chunk.Offset, int(chunk.Size), buf)
+		if err != nil {
+			glog.V(1).Infof("read from %s: %v", fileUrl, err)
+		} else {
+			break
+		}
+	}
 	return bytes.NewReader(buf), nil
 }
diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go
index d7b5ebc4d..e2e3575dc 100644
--- a/weed/replication/source/filer_source.go
+++ b/weed/replication/source/filer_source.go
@@ -3,13 +3,15 @@ package source
 import (
 	"context"
 	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/security"
-	"github.com/spf13/viper"
-	"google.golang.org/grpc"
 	"io"
 	"net/http"
 	"strings"
 
+	"google.golang.org/grpc"
+
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/security"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
@@ -23,32 +25,41 @@ type FilerSource struct {
 	grpcAddress    string
 	grpcDialOption grpc.DialOption
 	Dir            string
+	address        string
+	proxyByFiler   bool
 }
 
-func (fs *FilerSource) Initialize(configuration util.Configuration) error {
-	return fs.initialize(
-		configuration.GetString("grpcAddress"),
-		configuration.GetString("directory"),
+func (fs *FilerSource) Initialize(configuration util.Configuration, prefix string) error {
+	return fs.DoInitialize(
+		"",
+		configuration.GetString(prefix+"grpcAddress"),
+		configuration.GetString(prefix+"directory"),
+		false,
 	)
 }
 
-func (fs *FilerSource) initialize(grpcAddress string, dir string) (err error) {
+func (fs *FilerSource) DoInitialize(address, grpcAddress string, dir string, readChunkFromFiler bool) (err error) {
+	fs.address = address
+	if fs.address == "" {
+		fs.address = pb.GrpcAddressToServerAddress(grpcAddress)
+	}
 	fs.grpcAddress = grpcAddress
 	fs.Dir = dir
-	fs.grpcDialOption = security.LoadClientTLS(viper.Sub("grpc"), "client")
+	fs.grpcDialOption = security.LoadClientTLS(util.GetViper(), "grpc.client")
+	fs.proxyByFiler = readChunkFromFiler
 	return nil
 }
 
-func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrl string, err error) {
+func (fs *FilerSource) LookupFileId(part string) (fileUrls []string, err error) {
 
 	vid2Locations := make(map[string]*filer_pb.Locations)
 
 	vid := volumeId(part)
 
-	err = fs.withFilerClient(ctx, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
+	err = fs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 
 		glog.V(4).Infof("read lookup volume id locations: %v", vid)
-		resp, err := client.LookupVolume(ctx, &filer_pb.LookupVolumeRequest{
+		resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
 			VolumeIds: []string{vid},
 		})
 		if err != nil {
@@ -62,42 +73,65 @@ func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrl s
 
 	if err != nil {
 		glog.V(1).Infof("LookupFileId volume id %s: %v", vid, err)
-		return "", fmt.Errorf("LookupFileId volume id %s: %v", vid, err)
+		return nil, fmt.Errorf("LookupFileId volume id %s: %v", vid, err)
 	}
 
 	locations := vid2Locations[vid]
 
 	if locations == nil || len(locations.Locations) == 0 {
 		glog.V(1).Infof("LookupFileId locate volume id %s: %v", vid, err)
-		return "", fmt.Errorf("LookupFileId locate volume id %s: %v", vid, err)
+		return nil, fmt.Errorf("LookupFileId locate volume id %s: %v", vid, err)
 	}
 
-	fileUrl = fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, part)
+	if !fs.proxyByFiler {
+		for _, loc := range locations.Locations {
+			fileUrls = append(fileUrls, fmt.Sprintf("http://%s/%s?readDeleted=true", loc.Url, part))
+		}
+	} else {
+		fileUrls = append(fileUrls, fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, part))
+	}
 
 	return
 }
 
-func (fs *FilerSource) ReadPart(ctx context.Context, part string) (filename string, header http.Header, readCloser io.ReadCloser, err error) {
+func (fs *FilerSource) ReadPart(fileId string) (filename string, header http.Header, resp *http.Response, err error) {
 
-	fileUrl, err := fs.LookupFileId(ctx, part)
+	if fs.proxyByFiler {
+		return util.DownloadFile("http://" + fs.address + "/?proxyChunkId=" + fileId)
+	}
+
+	fileUrls, err := fs.LookupFileId(fileId)
 	if err != nil {
 		return "", nil, nil, err
 	}
 
-	filename, header, readCloser, err = util.DownloadFile(fileUrl)
+	for _, fileUrl := range fileUrls {
+		filename, header, resp, err = util.DownloadFile(fileUrl)
+		if err != nil {
+			glog.V(1).Infof("fail to read from %s: %v", fileUrl, err)
+		} else {
+			break
+		}
+	}
 
-	return filename, header, readCloser, err
+	return filename, header, resp, err
 }
 
-func (fs *FilerSource) withFilerClient(ctx context.Context, grpcDialOption grpc.DialOption, fn func(filer_pb.SeaweedFilerClient) error) error {
+var _ = filer_pb.FilerClient(&FilerSource{})
+
+func (fs *FilerSource) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
 
-	return util.WithCachedGrpcClient(ctx, func(grpcConnection *grpc.ClientConn) error {
+	return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
 		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
 		return fn(client)
 	}, fs.grpcAddress, fs.grpcDialOption)
 
 }
 
+func (fs *FilerSource) AdjustedUrl(location *filer_pb.Location) string {
+	return location.Url
+}
+
 func volumeId(fileId string) string {
 	lastCommaIndex := strings.LastIndex(fileId, ",")
 	if lastCommaIndex > 0 {
diff --git a/weed/replication/sub/notification_aws_sqs.go b/weed/replication/sub/notification_aws_sqs.go
index bed26c79c..642834c72 100644
--- a/weed/replication/sub/notification_aws_sqs.go
+++ b/weed/replication/sub/notification_aws_sqs.go
@@ -27,14 +27,14 @@ func (k *AwsSqsInput) GetName() string {
 	return "aws_sqs"
 }
 
-func (k *AwsSqsInput) Initialize(configuration util.Configuration) error {
-	glog.V(0).Infof("replication.notification.aws_sqs.region: %v", configuration.GetString("region"))
-	glog.V(0).Infof("replication.notification.aws_sqs.sqs_queue_name: %v", configuration.GetString("sqs_queue_name"))
+func (k *AwsSqsInput) Initialize(configuration util.Configuration, prefix string) error {
+	glog.V(0).Infof("replication.notification.aws_sqs.region: %v", configuration.GetString(prefix+"region"))
+	glog.V(0).Infof("replication.notification.aws_sqs.sqs_queue_name: %v", configuration.GetString(prefix+"sqs_queue_name"))
 	return k.initialize(
-		configuration.GetString("aws_access_key_id"),
-		configuration.GetString("aws_secret_access_key"),
-		configuration.GetString("region"),
-		configuration.GetString("sqs_queue_name"),
+		configuration.GetString(prefix+"aws_access_key_id"),
+		configuration.GetString(prefix+"aws_secret_access_key"),
+		configuration.GetString(prefix+"region"),
+		configuration.GetString(prefix+"sqs_queue_name"),
 	)
 }
 
@@ -68,7 +68,7 @@ func (k *AwsSqsInput) initialize(awsAccessKeyId, awsSecretAccessKey, region, que
 	return nil
 }
 
-func (k *AwsSqsInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *AwsSqsInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {
 
 	// receive message
 	result, err := k.svc.ReceiveMessage(&sqs.ReceiveMessageInput{
@@ -92,7 +92,9 @@ func (k *AwsSqsInput) ReceiveMessage() (key string, message *filer_pb.EventNotif
 	}
 
 	// process the message
-	key = *result.Messages[0].Attributes["key"]
+	// fmt.Printf("messages: %+v\n", result.Messages[0])
+	keyValue := result.Messages[0].MessageAttributes["key"]
+	key = *keyValue.StringValue
 	text := *result.Messages[0].Body
 	message = &filer_pb.EventNotification{}
 	err = proto.UnmarshalText(text, message)
diff --git a/weed/replication/sub/notification_gocdk_pub_sub.go b/weed/replication/sub/notification_gocdk_pub_sub.go
index eddba9ff8..b16eec2e1 100644
--- a/weed/replication/sub/notification_gocdk_pub_sub.go
+++ b/weed/replication/sub/notification_gocdk_pub_sub.go
@@ -2,13 +2,20 @@ package sub
 
 import (
 	"context"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/golang/protobuf/proto"
+	"github.com/streadway/amqp"
 	"gocloud.dev/pubsub"
 	_ "gocloud.dev/pubsub/awssnssqs"
+	"gocloud.dev/pubsub/rabbitpubsub"
+	"net/url"
+	"os"
+	"path"
+	"strings"
+	"time"
+
 	// _ "gocloud.dev/pubsub/azuresb"
 	_ "gocloud.dev/pubsub/gcppubsub"
 	_ "gocloud.dev/pubsub/natspubsub"
@@ -19,32 +26,139 @@ func init() {
 	NotificationInputs = append(NotificationInputs, &GoCDKPubSubInput{})
 }
 
+func getPath(rawUrl string) string {
+	parsedUrl, _ := url.Parse(rawUrl)
+	return path.Join(parsedUrl.Host, parsedUrl.Path)
+}
+
+func QueueDeclareAndBind(conn *amqp.Connection, exchangeUrl string, queueUrl string) error {
+	exchangeName := getPath(exchangeUrl)
+	queueName := getPath(queueUrl)
+	exchangeNameDLX := "DLX." + exchangeName
+	queueNameDLX := "DLX." + queueName
+	ch, err := conn.Channel()
+	if err != nil {
+		glog.Error(err)
+		return err
+	}
+	defer ch.Close()
+	if err := ch.ExchangeDeclare(
+		exchangeNameDLX, "fanout", false, false, false, false, nil); err != nil {
+		glog.Error(err)
+		return err
+	}
+	if err := ch.ExchangeDeclare(
+		exchangeName, "fanout", false, false, false, false, nil); err != nil {
+		glog.Error(err)
+		return err
+	}
+	if _, err := ch.QueueDeclare(
+		queueName, false, false, false, false,
+		amqp.Table{"x-dead-letter-exchange": exchangeNameDLX}); err != nil {
+		glog.Error(err)
+		return err
+	}
+	if err := ch.QueueBind(queueName, "", exchangeName, false, nil); err != nil {
+		glog.Error(err)
+		return err
+	}
+	if _, err := ch.QueueDeclare(
+		queueNameDLX, false, false, false, false,
+		amqp.Table{"x-dead-letter-exchange": exchangeName, "x-message-ttl": 600000}); err != nil {
+		glog.Error(err)
+		return err
+	}
+	if err := ch.QueueBind(queueNameDLX, "", exchangeNameDLX, false, nil); err != nil {
+		glog.Error(err)
+		return err
+	}
+	return nil
+}
+
 type GoCDKPubSubInput struct {
-	sub *pubsub.Subscription
+	sub    *pubsub.Subscription
+	subURL string
 }
 
 func (k *GoCDKPubSubInput) GetName() string {
 	return "gocdk_pub_sub"
 }
 
-func (k *GoCDKPubSubInput) Initialize(config util.Configuration) error {
-	subURL := config.GetString("sub_url")
-	glog.V(0).Infof("notification.gocdk_pub_sub.sub_url: %v", subURL)
-	sub, err := pubsub.OpenSubscription(context.Background(), subURL)
+func (k *GoCDKPubSubInput) Initialize(configuration util.Configuration, prefix string) error {
+	topicUrl := configuration.GetString(prefix + "topic_url")
+	k.subURL = configuration.GetString(prefix + "sub_url")
+	glog.V(0).Infof("notification.gocdk_pub_sub.sub_url: %v", k.subURL)
+	sub, err := pubsub.OpenSubscription(context.Background(), k.subURL)
 	if err != nil {
 		return err
 	}
+	var conn *amqp.Connection
+	if sub.As(&conn) {
+		ch, err := conn.Channel()
+		if err != nil {
+			return err
+		}
+		defer ch.Close()
+		_, err = ch.QueueInspect(getPath(k.subURL))
+		if err != nil {
+			if strings.HasPrefix(err.Error(), "Exception (404) Reason") {
+				if err := QueueDeclareAndBind(conn, topicUrl, k.subURL); err != nil {
+					return err
+				}
+			} else {
+				return err
+			}
+		}
+	}
 	k.sub = sub
 	return nil
 }
 
-func (k *GoCDKPubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
-	msg, err := k.sub.Receive(context.Background())
+func (k *GoCDKPubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {
+	ctx := context.Background()
+	msg, err := k.sub.Receive(ctx)
+	if err != nil {
+		var conn *amqp.Connection
+		if k.sub.As(&conn) && conn.IsClosed() {
+			conn.Close()
+			k.sub.Shutdown(ctx)
+			conn, err = amqp.Dial(os.Getenv("RABBIT_SERVER_URL"))
+			if err != nil {
+				glog.Error(err)
+				time.Sleep(time.Second)
+				return
+			}
+			k.sub = rabbitpubsub.OpenSubscription(conn, getPath(k.subURL), nil)
+			return
+		}
+		// This is permanent cached sub err
+		glog.Fatal(err)
+	}
+	onFailureFn = func() {
+		if msg.Nackable() {
+			isRedelivered := false
+			var delivery amqp.Delivery
+			if msg.As(&delivery) {
+				isRedelivered = delivery.Redelivered
+				glog.Warningf("onFailureFn() metadata: %+v, redelivered: %v", msg.Metadata, delivery.Redelivered)
+			}
+			if isRedelivered {
+				if err := delivery.Nack(false, false); err != nil {
+					glog.Error(err)
+				}
+			} else {
+				msg.Nack()
+			}
+		}
+	}
+	onSuccessFn = func() {
+		msg.Ack()
+	}
 	key = msg.Metadata["key"]
 	message = &filer_pb.EventNotification{}
 	err = proto.Unmarshal(msg.Body, message)
 	if err != nil {
-		return "", nil, err
+		return "", nil, onSuccessFn, onFailureFn, err
 	}
-	return key, message, nil
+	return key, message, onSuccessFn, onFailureFn, nil
 }
diff --git a/weed/replication/sub/notification_google_pub_sub.go b/weed/replication/sub/notification_google_pub_sub.go
index ad6b42a2e..f7c767d4a 100644
--- a/weed/replication/sub/notification_google_pub_sub.go
+++ b/weed/replication/sub/notification_google_pub_sub.go
@@ -27,13 +27,13 @@ func (k *GooglePubSubInput) GetName() string {
 	return "google_pub_sub"
 }
 
-func (k *GooglePubSubInput) Initialize(configuration util.Configuration) error {
-	glog.V(0).Infof("notification.google_pub_sub.project_id: %v", configuration.GetString("project_id"))
-	glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString("topic"))
+func (k *GooglePubSubInput) Initialize(configuration util.Configuration, prefix string) error {
+	glog.V(0).Infof("notification.google_pub_sub.project_id: %v", configuration.GetString(prefix+"project_id"))
+	glog.V(0).Infof("notification.google_pub_sub.topic: %v", configuration.GetString(prefix+"topic"))
 	return k.initialize(
-		configuration.GetString("google_application_credentials"),
-		configuration.GetString("project_id"),
-		configuration.GetString("topic"),
+		configuration.GetString(prefix+"google_application_credentials"),
+		configuration.GetString(prefix+"project_id"),
+		configuration.GetString(prefix+"topic"),
 	)
 }
 
@@ -85,16 +85,22 @@ func (k *GooglePubSubInput) initialize(google_application_credentials, projectId
 
 	go k.sub.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
 		k.messageChan <- m
-		m.Ack()
 	})
 
 	return err
 }
 
-func (k *GooglePubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *GooglePubSubInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {
 
 	m := <-k.messageChan
 
+	onSuccessFn = func() {
+		m.Ack()
+	}
+	onFailureFn = func() {
+		m.Nack()
+	}
+
 	// process the message
 	key = m.Attributes["key"]
 	message = &filer_pb.EventNotification{}
diff --git a/weed/replication/sub/notification_kafka.go b/weed/replication/sub/notification_kafka.go
index 1a86a8307..622a759ea 100644
--- a/weed/replication/sub/notification_kafka.go
+++ b/weed/replication/sub/notification_kafka.go
@@ -28,14 +28,14 @@ func (k *KafkaInput) GetName() string {
 	return "kafka"
 }
 
-func (k *KafkaInput) Initialize(configuration util.Configuration) error {
-	glog.V(0).Infof("replication.notification.kafka.hosts: %v\n", configuration.GetStringSlice("hosts"))
-	glog.V(0).Infof("replication.notification.kafka.topic: %v\n", configuration.GetString("topic"))
+func (k *KafkaInput) Initialize(configuration util.Configuration, prefix string) error {
+	glog.V(0).Infof("replication.notification.kafka.hosts: %v\n", configuration.GetStringSlice(prefix+"hosts"))
+	glog.V(0).Infof("replication.notification.kafka.topic: %v\n", configuration.GetString(prefix+"topic"))
 	return k.initialize(
-		configuration.GetStringSlice("hosts"),
-		configuration.GetString("topic"),
-		configuration.GetString("offsetFile"),
-		configuration.GetInt("offsetSaveIntervalSeconds"),
+		configuration.GetStringSlice(prefix+"hosts"),
+		configuration.GetString(prefix+"topic"),
+		configuration.GetString(prefix+"offsetFile"),
+		configuration.GetInt(prefix+"offsetSaveIntervalSeconds"),
 	)
 }
 
@@ -97,7 +97,7 @@ func (k *KafkaInput) initialize(hosts []string, topic string, offsetFile string,
 	return nil
 }
 
-func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error) {
 
 	msg := <-k.messageChan
 
diff --git a/weed/replication/sub/notifications.go b/weed/replication/sub/notifications.go
index 66fbef824..d5a910db9 100644
--- a/weed/replication/sub/notifications.go
+++ b/weed/replication/sub/notifications.go
@@ -9,8 +9,8 @@ type NotificationInput interface {
 	// GetName gets the name to locate the configuration in sync.toml file
 	GetName() string
 	// Initialize initializes the file store
-	Initialize(configuration util.Configuration) error
-	ReceiveMessage() (key string, message *filer_pb.EventNotification, err error)
+	Initialize(configuration util.Configuration, prefix string) error
+	ReceiveMessage() (key string, message *filer_pb.EventNotification, onSuccessFn func(), onFailureFn func(), err error)
 }
 
 var (