add WIP filer.replicate

13 files changed, 594 insertions, 96 deletions

diff --git a/weed/command/command.go b/weed/command/command.go
index 61f13c5e2..91b9bf3fc 100644
--- a/weed/command/command.go
+++ b/weed/command/command.go
@@ -14,6 +14,7 @@ var Commands = []*Command{
 	cmdCopy,
 	cmdFix,
 	cmdFilerExport,
+	cmdFilerReplicate,
 	cmdServer,
 	cmdMaster,
 	cmdFiler,
diff --git a/weed/command/filer_replication.go b/weed/command/filer_replication.go
new file mode 100644
index 000000000..18f549edc
--- /dev/null
+++ b/weed/command/filer_replication.go
@@ -0,0 +1,61 @@
+package command
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/server"
+	"github.com/spf13/viper"
+	"github.com/chrislusf/seaweedfs/weed/replication"
+)
+
+func init() {
+	cmdFilerReplicate.Run = runFilerReplicate // break init cycle
+}
+
+var cmdFilerReplicate = &Command{
+	UsageLine: "filer.replicate",
+	Short:     "replicate file changes to another destination",
+	Long: `replicate file changes to another destination
+
+	filer.replicate listens on filer notifications. If any file is updated, it will fetch the updated content,
+	and write to the other destination.
+
+	Run "weed scaffold -config replication" to generate a replication.toml file and customize the parameters.
+
+  `,
+}
+
+func runFilerReplicate(cmd *Command, args []string) bool {
+
+	weed_server.LoadConfiguration("replication", true)
+	config := viper.GetViper()
+
+	var notificationInput replication.NotificationInput
+
+	for _, input := range replication.NotificationInputs {
+		if config.GetBool("notification." + input.GetName() + ".enabled") {
+			viperSub := config.Sub("notification." + input.GetName())
+			if err := input.Initialize(viperSub); err != nil {
+				glog.Fatalf("Failed to initialize notification input for %s: %+v",
+					input.GetName(), err)
+			}
+			glog.V(0).Infof("Configure notification input to %s", input.GetName())
+			notificationInput = input
+			break
+		}
+	}
+
+	replicator := replication.NewReplicator(config.Sub("sink.filer"))
+
+	for {
+		key, m, err := notificationInput.ReceiveMessage()
+		if err != nil {
+			glog.Errorf("receive %s: +v", key, err)
+			continue
+		}
+		if err = replicator.Replicate(key, m); err != nil {
+			glog.Errorf("replicate %s: +v", key, err)
+		}
+	}
+
+	return true
+}
diff --git a/weed/command/scaffold.go b/weed/command/scaffold.go
index a5df2d18d..17d8b1884 100644
--- a/weed/command/scaffold.go
+++ b/weed/command/scaffold.go
@@ -19,7 +19,7 @@ var cmdScaffold = &Command{
 
 var (
 	outputPath = cmdScaffold.Flag.String("output", "", "if not empty, save the configuration file to this directory")
-	config     = cmdScaffold.Flag.String("config", "filer", "the configuration file to generate")
+	config     = cmdScaffold.Flag.String("config", "filer", "[filer|replication] the configuration file to generate")
 )
 
 func runScaffold(cmd *Command, args []string) bool {
@@ -28,6 +28,8 @@ func runScaffold(cmd *Command, args []string) bool {
 	switch *config {
 	case "filer":
 		content = FILER_TOML_EXAMPLE
+	case "replication":
+		content = REPLICATION_TOML_EXAMPLE
 	}
 	if content == "" {
 		println("need a valid -config option")
@@ -140,4 +142,32 @@ hosts = [
 topic = "seaweedfs_filer"
 
 `
+	REPLICATION_TOML_EXAMPLE = `
+# A sample TOML config file for replicating SeaweedFS filer store
+
+
+[source.filer]
+enabled = true
+grpcAddress = "localhost:18888"
+# id is to identify the notification source, avoid reprocessing the same events
+id = "filer1"
+# all files under this directory tree and not from this source.filer.id is replicated
+directory = "/"
+
+[notification.kafka]
+enabled = true
+hosts = [
+  "localhost:9092"
+]
+topic = "seaweedfs_filer"
+
+[sink.filer]
+enabled = true
+grpcAddress = "localhost:18888"
+# id is to identify the notification source, avoid reprocessing the same events
+id = "filer2"
+# all files under this directory tree and not from this source.filer.id is replicated
+directory = "/"
+
+`
 )
diff --git a/weed/filer2/filer.go b/weed/filer2/filer.go
index 84cac5c27..a652e6480 100644
--- a/weed/filer2/filer.go
+++ b/weed/filer2/filer.go
@@ -91,7 +91,7 @@ func (f *Filer) CreateEntry(entry *Entry) error {
 				return fmt.Errorf("mkdir %s: %v", dirPath, mkdirErr)
 			}
 
-			f.NotifyUpdateEvent(nil, dirEntry)
+			f.NotifyUpdateEvent(nil, dirEntry, false)
 
 		} else if !dirEntry.IsDirectory() {
 			return fmt.Errorf("%s is a file", dirPath)
@@ -125,7 +125,7 @@ func (f *Filer) CreateEntry(entry *Entry) error {
 		return fmt.Errorf("insert entry %s: %v", entry.FullPath, err)
 	}
 
-	f.NotifyUpdateEvent(oldEntry, entry)
+	f.NotifyUpdateEvent(oldEntry, entry, true)
 
 	f.deleteChunksIfNotNew(oldEntry, entry)
 
@@ -176,7 +176,7 @@ func (f *Filer) DeleteEntryMetaAndData(p FullPath, isRecursive bool, shouldDelet
 	}
 	glog.V(3).Infof("deleting entry %v", p)
 
-	f.NotifyUpdateEvent(entry, nil)
+	f.NotifyUpdateEvent(entry, nil, shouldDeleteChunks)
 
 	return f.store.DeleteEntry(p)
 }
diff --git a/weed/filer2/filer_notify.go b/weed/filer2/filer_notify.go
index 7fc64a1be..b8fe1dbfb 100644
--- a/weed/filer2/filer_notify.go
+++ b/weed/filer2/filer_notify.go
@@ -6,7 +6,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 )
 
-func (f *Filer) NotifyUpdateEvent(oldEntry, newEntry *Entry) {
+func (f *Filer) NotifyUpdateEvent(oldEntry, newEntry *Entry, deleteChunks bool) {
 	var key string
 	if oldEntry != nil {
 		key = string(oldEntry.FullPath)
@@ -23,13 +23,13 @@ func (f *Filer) NotifyUpdateEvent(oldEntry, newEntry *Entry) {
 		notification.Queue.SendMessage(
 			key,
 			&filer_pb.EventNotification{
-				OldEntry: toProtoEntry(oldEntry),
-				NewEntry: toProtoEntry(newEntry),
+				OldEntry:     toProtoEntry(oldEntry),
+				NewEntry:     toProtoEntry(newEntry),
+				DeleteChunks: deleteChunks,
 			},
 		)
 
 	}
-
 }
 
 func toProtoEntry(entry *Entry) *filer_pb.Entry {
diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto
index e2b175c11..c4cebb8ba 100644
--- a/weed/pb/filer.proto
+++ b/weed/pb/filer.proto
@@ -67,8 +67,9 @@ message Entry {
 }
 
 message EventNotification {
-    Entry old_entry = 2;
-    Entry new_entry = 3;
+    Entry old_entry = 1;
+    Entry new_entry = 2;
+    bool delete_chunks = 3;
 }
 
 message FileChunk {
@@ -77,6 +78,7 @@ message FileChunk {
     uint64 size = 3;
     int64 mtime = 4;
     string e_tag = 5;
+    string source_file_id = 6;
 }
 
 message FuseAttributes {
diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go
index 7ee8a7dd2..66b92bbcc 100644
--- a/weed/pb/filer_pb/filer.pb.go
+++ b/weed/pb/filer_pb/filer.pb.go
@@ -211,8 +211,9 @@ func (m *Entry) GetExtended() map[string][]byte {
 }
 
 type EventNotification struct {
-	OldEntry *Entry `protobuf:"bytes,2,opt,name=old_entry,json=oldEntry" json:"old_entry,omitempty"`
-	NewEntry *Entry `protobuf:"bytes,3,opt,name=new_entry,json=newEntry" json:"new_entry,omitempty"`
+	OldEntry     *Entry `protobuf:"bytes,1,opt,name=old_entry,json=oldEntry" json:"old_entry,omitempty"`
+	NewEntry     *Entry `protobuf:"bytes,2,opt,name=new_entry,json=newEntry" json:"new_entry,omitempty"`
+	DeleteChunks bool   `protobuf:"varint,3,opt,name=delete_chunks,json=deleteChunks" json:"delete_chunks,omitempty"`
 }
 
 func (m *EventNotification) Reset()                    { *m = EventNotification{} }
@@ -234,12 +235,20 @@ func (m *EventNotification) GetNewEntry() *Entry {
 	return nil
 }
 
+func (m *EventNotification) GetDeleteChunks() bool {
+	if m != nil {
+		return m.DeleteChunks
+	}
+	return false
+}
+
 type FileChunk struct {
-	FileId string `protobuf:"bytes,1,opt,name=file_id,json=fileId" json:"file_id,omitempty"`
-	Offset int64  `protobuf:"varint,2,opt,name=offset" json:"offset,omitempty"`
-	Size   uint64 `protobuf:"varint,3,opt,name=size" json:"size,omitempty"`
-	Mtime  int64  `protobuf:"varint,4,opt,name=mtime" json:"mtime,omitempty"`
-	ETag   string `protobuf:"bytes,5,opt,name=e_tag,json=eTag" json:"e_tag,omitempty"`
+	FileId       string `protobuf:"bytes,1,opt,name=file_id,json=fileId" json:"file_id,omitempty"`
+	Offset       int64  `protobuf:"varint,2,opt,name=offset" json:"offset,omitempty"`
+	Size         uint64 `protobuf:"varint,3,opt,name=size" json:"size,omitempty"`
+	Mtime        int64  `protobuf:"varint,4,opt,name=mtime" json:"mtime,omitempty"`
+	ETag         string `protobuf:"bytes,5,opt,name=e_tag,json=eTag" json:"e_tag,omitempty"`
+	SourceFileId string `protobuf:"bytes,6,opt,name=source_file_id,json=sourceFileId" json:"source_file_id,omitempty"`
 }
 
 func (m *FileChunk) Reset()                    { *m = FileChunk{} }
@@ -282,6 +291,13 @@ func (m *FileChunk) GetETag() string {
 	return ""
 }
 
+func (m *FileChunk) GetSourceFileId() string {
+	if m != nil {
+		return m.SourceFileId
+	}
+	return ""
+}
+
 type FuseAttributes struct {
 	FileSize    uint64 `protobuf:"varint,1,opt,name=file_size,json=fileSize" json:"file_size,omitempty"`
 	Mtime       int64  `protobuf:"varint,2,opt,name=mtime" json:"mtime,omitempty"`
@@ -1138,82 +1154,84 @@ var _SeaweedFiler_serviceDesc = grpc.ServiceDesc{
 func init() { proto.RegisterFile("filer.proto", fileDescriptor0) }
 
 var fileDescriptor0 = []byte{
-	// 1218 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x57, 0xcd, 0x6e, 0xdb, 0xc6,
-	0x13, 0x37, 0xa9, 0x0f, 0x8b, 0x23, 0x39, 0xff, 0x78, 0xe5, 0x24, 0xfc, 0x2b, 0x76, 0xaa, 0xb0,
-	0x49, 0xe1, 0xa0, 0x86, 0x61, 0xb8, 0x3d, 0x24, 0x0d, 0x0a, 0x34, 0xb0, 0x9d, 0x20, 0x80, 0xe3,
-	0x00, 0x74, 0x5c, 0xa0, 0xe8, 0x41, 0xa0, 0xc9, 0xb1, 0xba, 0x30, 0x45, 0xaa, 0xe4, 0xd2, 0x8e,
-	0xdb, 0x47, 0xe8, 0xa5, 0x7d, 0x80, 0x9e, 0x7b, 0xef, 0x03, 0xf4, 0xd2, 0x17, 0x2b, 0xf6, 0x83,
-	0xd4, 0x52, 0xa4, 0x9c, 0xf8, 0x90, 0xdb, 0xee, 0xcc, 0xec, 0xcc, 0x6f, 0x86, 0x33, 0xbf, 0x5d,
-	0x42, 0xf7, 0x8c, 0x86, 0x98, 0x6c, 0x4f, 0x93, 0x98, 0xc5, 0xa4, 0x23, 0x36, 0xa3, 0xe9, 0xa9,
-	0xf3, 0x16, 0xee, 0x1f, 0xc6, 0xf1, 0x79, 0x36, 0xdd, 0xa7, 0x09, 0xfa, 0x2c, 0x4e, 0xae, 0x0e,
-	0x22, 0x96, 0x5c, 0xb9, 0xf8, 0x73, 0x86, 0x29, 0x23, 0xeb, 0x60, 0x05, 0xb9, 0xc2, 0x36, 0x86,
-	0xc6, 0xa6, 0xe5, 0xce, 0x04, 0x84, 0x40, 0x33, 0xf2, 0x26, 0x68, 0x9b, 0x42, 0x21, 0xd6, 0xce,
-	0x01, 0xac, 0xd7, 0x3b, 0x4c, 0xa7, 0x71, 0x94, 0x22, 0x79, 0x0c, 0x2d, 0xe4, 0x02, 0xe1, 0xad,
-	0xbb, 0xfb, 0xbf, 0xed, 0x1c, 0xca, 0xb6, 0xb4, 0x93, 0x5a, 0xe7, 0x1f, 0x03, 0xc8, 0x21, 0x4d,
-	0x19, 0x17, 0x52, 0x4c, 0x3f, 0x0e, 0xcf, 0x5d, 0x68, 0x4f, 0x13, 0x3c, 0xa3, 0xef, 0x15, 0x22,
-	0xb5, 0x23, 0x5b, 0xb0, 0x9a, 0x32, 0x2f, 0x61, 0x2f, 0x93, 0x78, 0xf2, 0x92, 0x86, 0x78, 0xc4,
-	0x41, 0x37, 0x84, 0x49, 0x55, 0x41, 0xb6, 0x81, 0xd0, 0xc8, 0x0f, 0xb3, 0x94, 0x5e, 0xe0, 0x71,
-	0xae, 0xb5, 0x9b, 0x43, 0x63, 0xb3, 0xe3, 0xd6, 0x68, 0xc8, 0x1a, 0xb4, 0x42, 0x3a, 0xa1, 0xcc,
-	0x6e, 0x0d, 0x8d, 0xcd, 0x15, 0x57, 0x6e, 0x9c, 0xef, 0xa0, 0x5f, 0xc2, 0xaf, 0xd2, 0x7f, 0x02,
-	0xcb, 0x28, 0x45, 0xb6, 0x31, 0x6c, 0xd4, 0x15, 0x20, 0xd7, 0x3b, 0x7f, 0x9a, 0xd0, 0x12, 0xa2,
-	0xa2, 0xce, 0xc6, 0xac, 0xce, 0xe4, 0x21, 0xf4, 0x68, 0x3a, 0x9a, 0x15, 0xc3, 0x14, 0xf8, 0xba,
-	0x34, 0x2d, 0xea, 0x4e, 0xbe, 0x84, 0xb6, 0xff, 0x53, 0x16, 0x9d, 0xa7, 0x76, 0x43, 0x84, 0xea,
-	0xcf, 0x42, 0xf1, 0x64, 0xf7, 0xb8, 0xce, 0x55, 0x26, 0xe4, 0x29, 0x80, 0xc7, 0x58, 0x42, 0x4f,
-	0x33, 0x86, 0xa9, 0xc8, 0xb6, 0xbb, 0x6b, 0x6b, 0x07, 0xb2, 0x14, 0x5f, 0x14, 0x7a, 0x57, 0xb3,
-	0x25, 0xcf, 0xa0, 0x83, 0xef, 0x19, 0x46, 0x01, 0x06, 0x76, 0x4b, 0x04, 0xda, 0x98, 0xcb, 0x69,
-	0xfb, 0x40, 0xe9, 0x65, 0x86, 0x85, 0xf9, 0xe0, 0x39, 0xac, 0x94, 0x54, 0xe4, 0x36, 0x34, 0xce,
-	0x31, 0xff, 0xb2, 0x7c, 0xc9, 0xab, 0x7b, 0xe1, 0x85, 0x99, 0x6c, 0xb2, 0x9e, 0x2b, 0x37, 0xdf,
-	0x98, 0x4f, 0x0d, 0x27, 0x86, 0xd5, 0x83, 0x0b, 0x8c, 0xd8, 0x51, 0xcc, 0xe8, 0x19, 0xf5, 0x3d,
-	0x46, 0xe3, 0x88, 0x6c, 0x81, 0x15, 0x87, 0xc1, 0x48, 0xb6, 0x98, 0x59, 0xdf, 0x62, 0x9d, 0x38,
-	0x54, 0xe1, 0xb6, 0xc0, 0x8a, 0xf0, 0x52, 0x59, 0x37, 0x16, 0x58, 0x47, 0x78, 0x29, 0x56, 0xce,
-	0xaf, 0x60, 0x15, 0x75, 0x23, 0xf7, 0x60, 0x99, 0x1b, 0x8e, 0x68, 0xa0, 0xd0, 0xb6, 0xf9, 0xf6,
-	0x75, 0xc0, 0x9b, 0x30, 0x3e, 0x3b, 0x4b, 0x91, 0x89, 0xf0, 0x0d, 0x57, 0xed, 0xf8, 0x47, 0x4c,
-	0xe9, 0x2f, 0xb2, 0xef, 0x9a, 0xae, 0x58, 0xf3, 0xe4, 0x26, 0x8c, 0x4e, 0x50, 0xd4, 0xbb, 0xe1,
-	0xca, 0x0d, 0xe9, 0x43, 0x0b, 0x47, 0xcc, 0x1b, 0x8b, 0x86, 0xb2, 0xdc, 0x26, 0xbe, 0xf3, 0xc6,
-	0xce, 0x6f, 0x26, 0xdc, 0x2a, 0x7f, 0x04, 0x72, 0x1f, 0x2c, 0x01, 0x41, 0xb8, 0x35, 0x84, 0x5b,
-	0x31, 0xd8, 0xc7, 0x25, 0xd7, 0xa6, 0xee, 0x3a, 0x3f, 0x32, 0x89, 0x03, 0x89, 0x64, 0x45, 0x1e,
-	0x79, 0x13, 0x07, 0xc8, 0x8b, 0x9f, 0xd1, 0x40, 0x60, 0x59, 0x71, 0xf9, 0x92, 0x4b, 0xc6, 0x34,
-	0x50, 0x8d, 0xcd, 0x97, 0x3c, 0x3b, 0x3f, 0x11, 0x7e, 0xdb, 0x32, 0x3b, 0xb9, 0xe3, 0xd9, 0x4d,
-	0xb8, 0x74, 0x59, 0x42, 0xe6, 0x6b, 0x32, 0x84, 0x6e, 0x82, 0xd3, 0x50, 0x7d, 0x1a, 0xbb, 0x23,
-	0x54, 0xba, 0x88, 0x3c, 0x00, 0xf0, 0xe3, 0x30, 0x44, 0x5f, 0x18, 0x58, 0xc2, 0x40, 0x93, 0xf0,
-	0x22, 0x33, 0x16, 0x8e, 0x52, 0xf4, 0x6d, 0x18, 0x1a, 0x9b, 0x2d, 0xb7, 0xcd, 0x58, 0x78, 0x8c,
-	0xbe, 0x33, 0x86, 0xff, 0xbf, 0x42, 0x31, 0x5c, 0x57, 0x5a, 0x57, 0x2a, 0x92, 0xa8, 0x1b, 0x97,
-	0x0d, 0x80, 0xa9, 0x97, 0x60, 0xc4, 0xf8, 0xc8, 0x28, 0x7a, 0xb0, 0xa4, 0x64, 0x9f, 0x26, 0xfa,
-	0xd7, 0x6c, 0xe8, 0x5f, 0xd3, 0xf9, 0xdd, 0x84, 0x41, 0x5d, 0x24, 0x35, 0xce, 0xe5, 0xa9, 0x31,
-	0x6e, 0x30, 0x35, 0xb3, 0xe1, 0x34, 0x3f, 0x3c, 0x9c, 0x47, 0xda, 0x88, 0xc9, 0x59, 0xde, 0x9d,
-	0x99, 0x2f, 0x86, 0xf7, 0x69, 0xe6, 0x6e, 0x07, 0xee, 0xbc, 0x42, 0x26, 0x40, 0xc6, 0x11, 0xc3,
-	0x88, 0xe5, 0x75, 0x5f, 0x34, 0x12, 0xce, 0x2e, 0xdc, 0x9d, 0x3f, 0xa1, 0xea, 0x67, 0xc3, 0xb2,
-	0x2f, 0x45, 0xe2, 0x48, 0xcf, 0xcd, 0xb7, 0xce, 0x0f, 0x40, 0xf6, 0x12, 0xf4, 0x18, 0xde, 0xe0,
-	0x3e, 0x2a, 0xee, 0x16, 0xf3, 0xda, 0xbb, 0xe5, 0x0e, 0xf4, 0x4b, 0xae, 0x25, 0x16, 0x1e, 0xf1,
-	0x64, 0x1a, 0x7c, 0xaa, 0x88, 0x25, 0xd7, 0x2a, 0xe2, 0xdf, 0x06, 0x90, 0x7d, 0x0c, 0xf1, 0x46,
-	0x21, 0x6b, 0x2e, 0xdd, 0xca, 0x65, 0xd0, 0xa8, 0x5e, 0x06, 0x8f, 0xe0, 0x16, 0x37, 0x11, 0xd1,
-	0x46, 0x81, 0xc7, 0x3c, 0x75, 0xa3, 0xf5, 0x68, 0x2a, 0x21, 0xec, 0x7b, 0xcc, 0x53, 0x8e, 0x12,
-	0xf4, 0xb3, 0x84, 0x5f, 0x72, 0x62, 0xf2, 0x85, 0x23, 0x37, 0x17, 0xf1, 0x5c, 0x4a, 0x98, 0x55,
-	0x2e, 0x7f, 0x19, 0xd0, 0x7f, 0x91, 0xa6, 0x74, 0x1c, 0x7d, 0x1f, 0x87, 0xd9, 0x04, 0xf3, 0x64,
-	0xd6, 0xa0, 0xe5, 0xc7, 0x99, 0xfa, 0xbe, 0x2d, 0x57, 0x6e, 0xe6, 0x06, 0xdf, 0xac, 0x0c, 0xfe,
-	0x1c, 0x75, 0x34, 0xaa, 0xd4, 0xa1, 0x51, 0x43, 0x53, 0xa7, 0x06, 0xf2, 0x19, 0x74, 0x79, 0x7a,
-	0x23, 0x1f, 0x23, 0x86, 0x89, 0xe2, 0x50, 0xe0, 0xa2, 0x3d, 0x21, 0x71, 0x2e, 0x60, 0xad, 0x0c,
-	0x54, 0xf5, 0xe2, 0x42, 0x46, 0xe7, 0xbc, 0x98, 0x84, 0x0a, 0x25, 0x5f, 0x0a, 0x36, 0xc9, 0x4e,
-	0x43, 0xea, 0x8f, 0xb8, 0xa2, 0xa1, 0xd8, 0x44, 0x48, 0x4e, 0x92, 0x70, 0x96, 0x73, 0x53, 0xcb,
-	0xd9, 0xf9, 0x1a, 0xfa, 0xf2, 0x65, 0x54, 0x2e, 0xd0, 0x06, 0xc0, 0x85, 0x10, 0x8c, 0x68, 0x20,
-	0x1f, 0x05, 0x96, 0x6b, 0x49, 0xc9, 0xeb, 0x20, 0x75, 0xbe, 0x05, 0xeb, 0x30, 0x96, 0x39, 0xa7,
-	0x64, 0x07, 0xac, 0x30, 0xdf, 0xa8, 0xf7, 0x03, 0x99, 0xb5, 0x5c, 0x6e, 0xe7, 0xce, 0x8c, 0x9c,
-	0xe7, 0xd0, 0xc9, 0xc5, 0x79, 0x1e, 0xc6, 0xa2, 0x3c, 0xcc, 0xb9, 0x3c, 0x9c, 0x7f, 0x0d, 0x58,
-	0x2b, 0x43, 0x56, 0xa5, 0x3a, 0x81, 0x95, 0x22, 0xc4, 0x68, 0xe2, 0x4d, 0x15, 0x96, 0x1d, 0x1d,
-	0x4b, 0xf5, 0x58, 0x01, 0x30, 0x7d, 0xe3, 0x4d, 0x65, 0xf7, 0xf4, 0x42, 0x4d, 0x34, 0x78, 0x07,
-	0xab, 0x15, 0x93, 0x1a, 0x6a, 0x7a, 0xa2, 0x53, 0x53, 0x89, 0x39, 0x8b, 0xd3, 0x3a, 0x5f, 0x3d,
-	0x83, 0x7b, 0xb2, 0x61, 0xf7, 0x8a, 0xfe, 0xca, 0x6b, 0x5f, 0x6e, 0x43, 0x63, 0xbe, 0x0d, 0x9d,
-	0x01, 0xd8, 0xd5, 0xa3, 0x32, 0x99, 0xdd, 0x3f, 0xda, 0xd0, 0x3b, 0x46, 0xef, 0x12, 0x31, 0xe0,
-	0xcc, 0x96, 0x90, 0x71, 0x5e, 0xac, 0xf2, 0xcb, 0x97, 0x3c, 0x9e, 0xaf, 0x4a, 0xed, 0x53, 0x7b,
-	0xf0, 0xc5, 0x87, 0xcc, 0xd4, 0xa0, 0x2d, 0x91, 0x43, 0xe8, 0x6a, 0x4f, 0x4b, 0xb2, 0xae, 0x1d,
-	0xac, 0xbc, 0x98, 0x07, 0x1b, 0x0b, 0xb4, 0x85, 0x37, 0x0f, 0x48, 0xf5, 0x06, 0x21, 0x9f, 0x5f,
-	0x7f, 0xbf, 0x48, 0xdf, 0x8f, 0x3e, 0xe6, 0x12, 0x92, 0x80, 0x35, 0xc2, 0xd5, 0x01, 0x57, 0x29,
-	0x5e, 0x07, 0x5c, 0xc7, 0xd2, 0xc2, 0x9b, 0x46, 0xa6, 0xba, 0xb7, 0x2a, 0x7d, 0xeb, 0xde, 0xea,
-	0x18, 0x58, 0x78, 0xd3, 0xe8, 0x4c, 0xf7, 0x56, 0x65, 0x66, 0xdd, 0x5b, 0x1d, 0x07, 0x2e, 0x91,
-	0xb7, 0xd0, 0xd3, 0xb9, 0x85, 0x68, 0x07, 0x6a, 0xc8, 0x71, 0xf0, 0x60, 0x91, 0x5a, 0x77, 0xa8,
-	0x8f, 0x92, 0xee, 0xb0, 0x86, 0x4c, 0x74, 0x87, 0x75, 0x13, 0xe8, 0x2c, 0x91, 0x1f, 0xe1, 0xf6,
-	0x7c, 0x4b, 0x93, 0x87, 0xf3, 0x69, 0x55, 0x26, 0x65, 0xe0, 0x5c, 0x67, 0x92, 0x3b, 0x3f, 0x6d,
-	0x8b, 0xdf, 0xcb, 0xaf, 0xfe, 0x0b, 0x00, 0x00, 0xff, 0xff, 0xbf, 0xe3, 0x2a, 0x44, 0x6d, 0x0e,
-	0x00, 0x00,
+	// 1251 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xb4, 0x57, 0x5f, 0x6f, 0xd4, 0x46,
+	0x10, 0xc7, 0xbe, 0x3f, 0x39, 0xcf, 0xdd, 0x51, 0xd8, 0x0b, 0xe0, 0x1e, 0x84, 0x1e, 0x06, 0xaa,
+	0xa0, 0xa2, 0x08, 0xa5, 0x7d, 0x80, 0xa2, 0x4a, 0x45, 0x49, 0x40, 0x48, 0x01, 0x24, 0x07, 0x2a,
+	0x55, 0x7d, 0xb0, 0x1c, 0x7b, 0x72, 0x5d, 0xc5, 0x67, 0x5f, 0xed, 0x75, 0x42, 0xfa, 0x15, 0xfa,
+	0xd2, 0x4a, 0x7d, 0xed, 0x43, 0x9f, 0xfa, 0xde, 0x0f, 0xd0, 0x97, 0x7e, 0xb1, 0x6a, 0xff, 0xd8,
+	0xb7, 0x3e, 0xfb, 0x02, 0xa8, 0xe2, 0xcd, 0x3b, 0x33, 0x3b, 0xf3, 0x9b, 0xd9, 0x99, 0xdf, 0xae,
+	0xa1, 0x7f, 0x44, 0x23, 0x4c, 0xb7, 0xe6, 0x69, 0xc2, 0x12, 0xd2, 0x13, 0x0b, 0x6f, 0x7e, 0xe8,
+	0xbc, 0x82, 0xeb, 0xfb, 0x49, 0x72, 0x9c, 0xcf, 0x77, 0x69, 0x8a, 0x01, 0x4b, 0xd2, 0xb3, 0xbd,
+	0x98, 0xa5, 0x67, 0x2e, 0xfe, 0x94, 0x63, 0xc6, 0xc8, 0x0d, 0xb0, 0xc2, 0x42, 0x61, 0x1b, 0x13,
+	0x63, 0xd3, 0x72, 0x17, 0x02, 0x42, 0xa0, 0x1d, 0xfb, 0x33, 0xb4, 0x4d, 0xa1, 0x10, 0xdf, 0xce,
+	0x1e, 0xdc, 0x68, 0x76, 0x98, 0xcd, 0x93, 0x38, 0x43, 0x72, 0x17, 0x3a, 0xc8, 0x05, 0xc2, 0x5b,
+	0x7f, 0xfb, 0x93, 0xad, 0x02, 0xca, 0x96, 0xb4, 0x93, 0x5a, 0xe7, 0x1f, 0x03, 0xc8, 0x3e, 0xcd,
+	0x18, 0x17, 0x52, 0xcc, 0xde, 0x0f, 0xcf, 0x55, 0xe8, 0xce, 0x53, 0x3c, 0xa2, 0x6f, 0x15, 0x22,
+	0xb5, 0x22, 0xf7, 0xe1, 0x72, 0xc6, 0xfc, 0x94, 0x3d, 0x4d, 0x93, 0xd9, 0x53, 0x1a, 0xe1, 0x4b,
+	0x0e, 0xba, 0x25, 0x4c, 0xea, 0x0a, 0xb2, 0x05, 0x84, 0xc6, 0x41, 0x94, 0x67, 0xf4, 0x04, 0x0f,
+	0x0a, 0xad, 0xdd, 0x9e, 0x18, 0x9b, 0x3d, 0xb7, 0x41, 0x43, 0xd6, 0xa1, 0x13, 0xd1, 0x19, 0x65,
+	0x76, 0x67, 0x62, 0x6c, 0x0e, 0x5d, 0xb9, 0x70, 0xbe, 0x85, 0x51, 0x05, 0xbf, 0x4a, 0xff, 0x1e,
+	0xac, 0xa1, 0x14, 0xd9, 0xc6, 0xa4, 0xd5, 0x54, 0x80, 0x42, 0xef, 0xfc, 0x61, 0x42, 0x47, 0x88,
+	0xca, 0x3a, 0x1b, 0x8b, 0x3a, 0x93, 0x5b, 0x30, 0xa0, 0x99, 0xb7, 0x28, 0x86, 0x29, 0xf0, 0xf5,
+	0x69, 0x56, 0xd6, 0x9d, 0x7c, 0x01, 0xdd, 0xe0, 0xc7, 0x3c, 0x3e, 0xce, 0xec, 0x96, 0x08, 0x35,
+	0x5a, 0x84, 0xe2, 0xc9, 0xee, 0x70, 0x9d, 0xab, 0x4c, 0xc8, 0x43, 0x00, 0x9f, 0xb1, 0x94, 0x1e,
+	0xe6, 0x0c, 0x33, 0x91, 0x6d, 0x7f, 0xdb, 0xd6, 0x36, 0xe4, 0x19, 0x3e, 0x29, 0xf5, 0xae, 0x66,
+	0x4b, 0x1e, 0x41, 0x0f, 0xdf, 0x32, 0x8c, 0x43, 0x0c, 0xed, 0x8e, 0x08, 0xb4, 0xb1, 0x94, 0xd3,
+	0xd6, 0x9e, 0xd2, 0xcb, 0x0c, 0x4b, 0xf3, 0xf1, 0x63, 0x18, 0x56, 0x54, 0xe4, 0x12, 0xb4, 0x8e,
+	0xb1, 0x38, 0x59, 0xfe, 0xc9, 0xab, 0x7b, 0xe2, 0x47, 0xb9, 0x6c, 0xb2, 0x81, 0x2b, 0x17, 0x5f,
+	0x9b, 0x0f, 0x0d, 0xe7, 0x77, 0x03, 0x2e, 0xef, 0x9d, 0x60, 0xcc, 0x5e, 0x26, 0x8c, 0x1e, 0xd1,
+	0xc0, 0x67, 0x34, 0x89, 0xc9, 0x7d, 0xb0, 0x92, 0x28, 0xf4, 0xce, 0xed, 0xb1, 0x5e, 0x12, 0xa9,
+	0x78, 0xf7, 0xc1, 0x8a, 0xf1, 0x54, 0x59, 0x9b, 0x2b, 0xac, 0x63, 0x3c, 0x95, 0xd6, 0xb7, 0x61,
+	0x18, 0x62, 0x84, 0x0c, 0xbd, 0xb2, 0xae, 0xbc, 0xe8, 0x03, 0x29, 0x14, 0xf5, 0xcc, 0x9c, 0x3f,
+	0x0d, 0xb0, 0xca, 0xf2, 0x92, 0x6b, 0xb0, 0xc6, 0xdd, 0x79, 0x34, 0x54, 0x49, 0x75, 0xf9, 0xf2,
+	0x79, 0xc8, 0x7b, 0x35, 0x39, 0x3a, 0xca, 0x90, 0x89, 0xb0, 0x2d, 0x57, 0xad, 0xf8, 0x59, 0x67,
+	0xf4, 0x67, 0xd9, 0x9e, 0x6d, 0x57, 0x7c, 0xf3, 0x1a, 0xcc, 0x18, 0x9d, 0xa1, 0x38, 0x96, 0x96,
+	0x2b, 0x17, 0x64, 0x04, 0x1d, 0xf4, 0x98, 0x3f, 0x15, 0x7d, 0x67, 0xb9, 0x6d, 0x7c, 0xed, 0x4f,
+	0xc9, 0x1d, 0xb8, 0x98, 0x25, 0x79, 0x1a, 0xa0, 0x57, 0x84, 0xed, 0x0a, 0xed, 0x40, 0x4a, 0x9f,
+	0x8a, 0xe0, 0xce, 0x2f, 0x26, 0x5c, 0xac, 0x9e, 0x28, 0xb9, 0x0e, 0x96, 0xd8, 0x21, 0x82, 0x1b,
+	0x22, 0xb8, 0x60, 0x89, 0x83, 0x0a, 0x00, 0x53, 0x07, 0x50, 0x6c, 0x99, 0x25, 0xa1, 0xc4, 0x3b,
+	0x94, 0x5b, 0x5e, 0x24, 0x21, 0xf2, 0x93, 0xcc, 0x69, 0x28, 0x10, 0x0f, 0x5d, 0xfe, 0xc9, 0x25,
+	0x53, 0x1a, 0xaa, 0x29, 0xe1, 0x9f, 0xbc, 0x06, 0x41, 0x2a, 0xfc, 0x76, 0x65, 0x0d, 0xe4, 0x8a,
+	0xd7, 0x60, 0xc6, 0xa5, 0x6b, 0x32, 0x31, 0xfe, 0x4d, 0x26, 0xd0, 0x4f, 0x71, 0x1e, 0xa9, 0x63,
+	0xb6, 0x7b, 0x42, 0xa5, 0x8b, 0xc8, 0x4d, 0x80, 0x20, 0x89, 0x22, 0x0c, 0x84, 0x81, 0x25, 0x0c,
+	0x34, 0x09, 0x3f, 0x0a, 0xc6, 0x22, 0x2f, 0xc3, 0xc0, 0x86, 0x89, 0xb1, 0xd9, 0x71, 0xbb, 0x8c,
+	0x45, 0x07, 0x18, 0x38, 0x53, 0xf8, 0xf4, 0x19, 0x8a, 0x49, 0x3d, 0xd3, 0x5a, 0x5c, 0x31, 0x4e,
+	0xd3, 0xec, 0x6d, 0x00, 0xcc, 0xfd, 0x14, 0x63, 0xc6, 0xe7, 0x4f, 0x71, 0x8d, 0x25, 0x25, 0xbb,
+	0x34, 0xd5, 0xcf, 0xbc, 0xa5, 0x9f, 0xb9, 0xf3, 0xab, 0x09, 0xe3, 0xa6, 0x48, 0x8a, 0x1b, 0xaa,
+	0x23, 0x68, 0x7c, 0xc0, 0x08, 0x2e, 0x26, 0xdd, 0x7c, 0xf7, 0xa4, 0xbf, 0xd4, 0xe6, 0x55, 0x12,
+	0xc3, 0xf6, 0xc2, 0x7c, 0x35, 0xbc, 0x8f, 0x33, 0xc4, 0x0f, 0xe0, 0xca, 0x33, 0x64, 0x02, 0x64,
+	0x12, 0x33, 0x8c, 0x59, 0x51, 0xf7, 0x55, 0x83, 0xe3, 0x6c, 0xc3, 0xd5, 0xe5, 0x1d, 0xaa, 0x7e,
+	0x36, 0xac, 0x05, 0x52, 0x24, 0xb6, 0x0c, 0xdc, 0x62, 0xe9, 0x7c, 0x0f, 0x64, 0x27, 0x45, 0x9f,
+	0xe1, 0x07, 0x5c, 0x6e, 0xe5, 0x45, 0x65, 0x9e, 0x7b, 0x51, 0x5d, 0x81, 0x51, 0xc5, 0xb5, 0xc4,
+	0xc2, 0x23, 0xbe, 0x99, 0x87, 0x1f, 0x2b, 0x62, 0xc5, 0xb5, 0x8a, 0xf8, 0xb7, 0x01, 0x64, 0x57,
+	0x10, 0xd1, 0xff, 0xbb, 0xc1, 0x6b, 0x37, 0x4b, 0xab, 0x7e, 0xb3, 0xdc, 0x81, 0x8b, 0xdc, 0x44,
+	0x72, 0x61, 0xe8, 0x33, 0x5f, 0x5d, 0x8f, 0x03, 0x9a, 0x49, 0x08, 0xbb, 0x3e, 0xf3, 0x95, 0xa3,
+	0x14, 0x83, 0x3c, 0xe5, 0x37, 0xa6, 0x98, 0x7c, 0xe1, 0xc8, 0x2d, 0x44, 0x3c, 0x97, 0x0a, 0x66,
+	0x95, 0xcb, 0x5f, 0x06, 0x8c, 0x9e, 0x64, 0x19, 0x9d, 0xc6, 0xdf, 0x25, 0x51, 0x3e, 0xc3, 0x22,
+	0x99, 0x75, 0xe8, 0x04, 0x49, 0xae, 0xce, 0xb7, 0xe3, 0xca, 0xc5, 0xd2, 0xe0, 0x9b, 0xb5, 0xc1,
+	0x5f, 0xa2, 0x8e, 0x56, 0x9d, 0x3a, 0x34, 0x6a, 0x68, 0xeb, 0xd4, 0x40, 0x3e, 0x83, 0x3e, 0x4f,
+	0xcf, 0x0b, 0x30, 0x66, 0x98, 0x2a, 0xa6, 0x05, 0x2e, 0xda, 0x11, 0x12, 0xe7, 0x04, 0xd6, 0xab,
+	0x40, 0x55, 0x2f, 0xae, 0xe4, 0x7d, 0xce, 0x8b, 0x69, 0xa4, 0x50, 0xf2, 0x4f, 0xc1, 0x26, 0xf9,
+	0x61, 0x44, 0x03, 0x8f, 0x2b, 0x5a, 0x8a, 0x4d, 0x84, 0xe4, 0x4d, 0x1a, 0x2d, 0x72, 0x6e, 0x6b,
+	0x39, 0x3b, 0x5f, 0xc1, 0x48, 0x3e, 0xb3, 0xaa, 0x05, 0xda, 0x00, 0x38, 0x11, 0x02, 0x8f, 0x86,
+	0xf2, 0x85, 0x61, 0xb9, 0x96, 0x94, 0x3c, 0x0f, 0x33, 0xe7, 0x1b, 0xb0, 0xf6, 0x13, 0x99, 0x73,
+	0x46, 0x1e, 0x80, 0x15, 0x15, 0x0b, 0xf5, 0x18, 0x21, 0x8b, 0x96, 0x2b, 0xec, 0xdc, 0x85, 0x91,
+	0xf3, 0x18, 0x7a, 0x85, 0xb8, 0xc8, 0xc3, 0x58, 0x95, 0x87, 0xb9, 0x94, 0x87, 0xf3, 0xaf, 0x01,
+	0xeb, 0x55, 0xc8, 0xaa, 0x54, 0x6f, 0x60, 0x58, 0x86, 0xf0, 0x66, 0xfe, 0x5c, 0x61, 0x79, 0xa0,
+	0x63, 0xa9, 0x6f, 0x2b, 0x01, 0x66, 0x2f, 0xfc, 0xb9, 0xec, 0x9e, 0x41, 0xa4, 0x89, 0xc6, 0xaf,
+	0xe1, 0x72, 0xcd, 0xa4, 0x81, 0x9a, 0xee, 0xe9, 0xd4, 0x54, 0x61, 0xce, 0x72, 0xb7, 0xce, 0x57,
+	0x8f, 0xe0, 0x9a, 0x6c, 0xd8, 0x9d, 0xb2, 0xbf, 0x8a, 0xda, 0x57, 0xdb, 0xd0, 0x58, 0x6e, 0x43,
+	0x67, 0x0c, 0x76, 0x7d, 0xab, 0x4c, 0x66, 0xfb, 0xb7, 0x2e, 0x0c, 0x0e, 0xd0, 0x3f, 0x45, 0x0c,
+	0x39, 0xb3, 0xa5, 0x64, 0x5a, 0x14, 0xab, 0xfa, 0x8c, 0x26, 0x77, 0x97, 0xab, 0xd2, 0xf8, 0x6e,
+	0x1f, 0x7f, 0xfe, 0x2e, 0x33, 0x35, 0x68, 0x17, 0xc8, 0x3e, 0xf4, 0xb5, 0x77, 0x2a, 0xb9, 0xa1,
+	0x6d, 0xac, 0x3d, 0xbf, 0xc7, 0x1b, 0x2b, 0xb4, 0xa5, 0x37, 0x1f, 0x48, 0xfd, 0x06, 0x21, 0xb7,
+	0xcf, 0xbf, 0x5f, 0xa4, 0xef, 0x3b, 0xef, 0x73, 0x09, 0x49, 0xc0, 0x1a, 0xe1, 0xea, 0x80, 0xeb,
+	0x14, 0xaf, 0x03, 0x6e, 0x62, 0x69, 0xe1, 0x4d, 0x23, 0x53, 0xdd, 0x5b, 0x9d, 0xbe, 0x75, 0x6f,
+	0x4d, 0x0c, 0x2c, 0xbc, 0x69, 0x74, 0xa6, 0x7b, 0xab, 0x33, 0xb3, 0xee, 0xad, 0x89, 0x03, 0x2f,
+	0x90, 0x57, 0x30, 0xd0, 0xb9, 0x85, 0x68, 0x1b, 0x1a, 0xc8, 0x71, 0x7c, 0x73, 0x95, 0x5a, 0x77,
+	0xa8, 0x8f, 0x92, 0xee, 0xb0, 0x81, 0x4c, 0x74, 0x87, 0x4d, 0x13, 0xe8, 0x5c, 0x20, 0x3f, 0xc0,
+	0xa5, 0xe5, 0x96, 0x26, 0xb7, 0x96, 0xd3, 0xaa, 0x4d, 0xca, 0xd8, 0x39, 0xcf, 0xa4, 0x70, 0x7e,
+	0xd8, 0x15, 0xff, 0xaa, 0x5f, 0xfe, 0x17, 0x00, 0x00, 0xff, 0xff, 0x08, 0x7d, 0x7a, 0x7c, 0xba,
+	0x0e, 0x00, 0x00,
 }
diff --git a/weed/replication/notification_kafka.go b/weed/replication/notification_kafka.go
new file mode 100644
index 000000000..ce3b86ae9
--- /dev/null
+++ b/weed/replication/notification_kafka.go
@@ -0,0 +1,77 @@
+package replication
+
+import (
+	"github.com/Shopify/sarama"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/golang/protobuf/proto"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+)
+
+func init() {
+	NotificationInputs = append(NotificationInputs, &KafkaInput{
+	})
+}
+
+type KafkaInput struct {
+	topic       string
+	consumer    sarama.Consumer
+	messageChan chan *sarama.ConsumerMessage
+}
+
+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"))
+	return k.initialize(
+		configuration.GetStringSlice("hosts"),
+		configuration.GetString("topic"),
+	)
+}
+
+func (k *KafkaInput) initialize(hosts []string, topic string) (err error) {
+	config := sarama.NewConfig()
+	config.Consumer.Return.Errors = true
+	k.consumer, err = sarama.NewConsumer(hosts, config)
+	k.topic = topic
+	k.messageChan = make(chan *sarama.ConsumerMessage, 1)
+
+	partitions, err := k.consumer.Partitions(topic)
+	if err != nil {
+		panic(err)
+	}
+
+	for _, partition := range partitions {
+		partitionConsumer, err := k.consumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
+		if err != nil {
+			panic(err)
+		}
+		go func() {
+			for {
+				select {
+				case err := <-partitionConsumer.Errors():
+					fmt.Println(err)
+				case msg := <-partitionConsumer.Messages():
+					k.messageChan <- msg
+				}
+			}
+		}()
+	}
+
+	return nil
+}
+
+func (k *KafkaInput) ReceiveMessage() (key string, message *filer_pb.EventNotification, err error) {
+
+	msg := <-k.messageChan
+
+	key = string(msg.Key)
+	message = &filer_pb.EventNotification{}
+	err = proto.Unmarshal(msg.Value, message)
+
+	return
+}
diff --git a/weed/replication/notifications.go b/weed/replication/notifications.go
new file mode 100644
index 000000000..ff40c3aad
--- /dev/null
+++ b/weed/replication/notifications.go
@@ -0,0 +1,18 @@
+package replication
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+)
+
+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)
+}
+
+var (
+	NotificationInputs []NotificationInput
+)
diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go
new file mode 100644
index 000000000..c223021c2
--- /dev/null
+++ b/weed/replication/replicator.go
@@ -0,0 +1,31 @@
+package replication
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/replication/sink"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+)
+
+type Replicator struct {
+	sink *sink.FilerSink
+}
+
+func NewReplicator(config util.Configuration) *Replicator {
+
+	sink := &sink.FilerSink{}
+	sink.Initialize(config)
+
+	return &Replicator{
+		sink: sink,
+	}
+}
+
+func (r *Replicator) Replicate(key string, message *filer_pb.EventNotification) error {
+	if message.OldEntry != nil && message.NewEntry == nil {
+		return r.sink.DeleteEntry(message.OldEntry, message.DeleteChunks)
+	}
+	if message.OldEntry == nil && message.NewEntry != nil {
+		return r.sink.CreateEntry(message.NewEntry)
+	}
+	return r.sink.UpdateEntry(message.OldEntry, message.NewEntry, message.DeleteChunks)
+}
diff --git a/weed/replication/sink/filer_sink.go b/weed/replication/sink/filer_sink.go
new file mode 100644
index 000000000..57ebeddff
--- /dev/null
+++ b/weed/replication/sink/filer_sink.go
@@ -0,0 +1,163 @@
+package sink
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"fmt"
+	"strings"
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"context"
+	"sync"
+)
+
+type ReplicationSink interface {
+	DeleteEntry(entry *filer_pb.Entry, deleteIncludeChunks bool) error
+	CreateEntry(entry *filer_pb.Entry) error
+	UpdateEntry(oldEntry, newEntry *filer_pb.Entry, deleteIncludeChunks bool) error
+}
+
+type FilerSink struct {
+	grpcAddress string
+	id          string
+	dir         string
+}
+
+func (fs *FilerSink) Initialize(configuration util.Configuration) error {
+	return fs.initialize(
+		configuration.GetString("grpcAddress"),
+		configuration.GetString("id"),
+		configuration.GetString("directory"),
+	)
+}
+
+func (fs *FilerSink) initialize(grpcAddress string, id string, dir string) (err error) {
+	fs.grpcAddress = grpcAddress
+	fs.id = id
+	fs.dir = dir
+	return nil
+}
+
+func (fs *FilerSink) DeleteEntry(entry *filer_pb.Entry, deleteIncludeChunks bool) error {
+	return fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		dir, name := filer2.FullPath(entry.Name).DirAndName()
+
+		request := &filer_pb.DeleteEntryRequest{
+			Directory:    dir,
+			Name:         name,
+			IsDirectory:  entry.IsDirectory,
+			IsDeleteData: deleteIncludeChunks,
+		}
+
+		glog.V(1).Infof("delete entry: %v", request)
+		_, err := client.DeleteEntry(context.Background(), request)
+		if err != nil {
+			glog.V(0).Infof("delete entry %s: %v", entry.Name, err)
+			return fmt.Errorf("delete entry %s: %v", entry.Name, err)
+		}
+
+		return nil
+	})
+}
+
+func (fs *FilerSink) CreateEntry(entry *filer_pb.Entry) error {
+
+	replicatedChunks, err := replicateChunks(entry.Chunks)
+
+	if err != nil {
+		glog.V(0).Infof("replicate entry chunks %s: %v", entry.Name, err)
+		return fmt.Errorf("replicate entry chunks %s: %v", entry.Name, err)
+	}
+
+	return fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		dir, name := filer2.FullPath(entry.Name).DirAndName()
+
+		request := &filer_pb.CreateEntryRequest{
+			Directory: dir,
+			Entry: &filer_pb.Entry{
+				Name:        name,
+				IsDirectory: entry.IsDirectory,
+				Attributes:  entry.Attributes,
+				Chunks:      replicatedChunks,
+			},
+		}
+
+		glog.V(1).Infof("create: %v", request)
+		if _, err := client.CreateEntry(context.Background(), request); err != nil {
+			glog.V(0).Infof("create entry %s: %v", entry.Name, err)
+			return fmt.Errorf("create entry %s: %v", entry.Name, err)
+		}
+
+		return nil
+	})
+}
+
+func (fs *FilerSink) UpdateEntry(oldEntry, newEntry *filer_pb.Entry, deleteIncludeChunks bool) error {
+	return nil
+}
+
+func (fs *FilerSink) withFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+
+	grpcConnection, err := util.GrpcDial(fs.grpcAddress)
+	if err != nil {
+		return fmt.Errorf("fail to dial %s: %v", fs.grpcAddress, err)
+	}
+	defer grpcConnection.Close()
+
+	client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+
+	return fn(client)
+}
+
+func volumeId(fileId string) string {
+	lastCommaIndex := strings.LastIndex(fileId, ",")
+	if lastCommaIndex > 0 {
+		return fileId[:lastCommaIndex]
+	}
+	return fileId
+}
+
+func replicateChunks(sourceChunks []*filer_pb.FileChunk) (replicatedChunks []*filer_pb.FileChunk, err error) {
+	if len(sourceChunks) == 0 {
+		return
+	}
+	var wg sync.WaitGroup
+	for _, s := range sourceChunks {
+		wg.Add(1)
+		go func(chunk *filer_pb.FileChunk) {
+			defer wg.Done()
+			replicatedChunk, e := replicateOneChunk(chunk)
+			if e != nil {
+				err = e
+			}
+			replicatedChunks = append(replicatedChunks, replicatedChunk)
+		}(s)
+	}
+	wg.Wait()
+
+	return
+}
+
+func replicateOneChunk(sourceChunk *filer_pb.FileChunk) (*filer_pb.FileChunk, error) {
+
+	fileId, err := fetchAndWrite(sourceChunk)
+	if err != nil {
+		return nil, fmt.Errorf("copy %s: %v", sourceChunk.FileId, err)
+	}
+
+	return &filer_pb.FileChunk{
+		FileId:       fileId,
+		Offset:       sourceChunk.Offset,
+		Size:         sourceChunk.Size,
+		Mtime:        sourceChunk.Mtime,
+		ETag:         sourceChunk.ETag,
+		SourceFileId: sourceChunk.FileId,
+	}, nil
+}
+
+func fetchAndWrite(sourceChunk *filer_pb.FileChunk) (fileId string, err error) {
+
+	return
+}
diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go
new file mode 100644
index 000000000..49c623815
--- /dev/null
+++ b/weed/replication/source/filer_source.go
@@ -0,0 +1,97 @@
+package source
+
+import (
+	"io"
+	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"strings"
+	"context"
+)
+
+type ReplicationSource interface {
+	ReadPart(part string) io.ReadCloser
+}
+
+type FilerSource struct {
+	grpcAddress string
+	id          string
+	dir         string
+}
+
+func (fs *FilerSource) Initialize(configuration util.Configuration) error {
+	return fs.initialize(
+		configuration.GetString("grpcAddress"),
+		configuration.GetString("id"),
+		configuration.GetString("directory"),
+	)
+}
+
+func (fs *FilerSource) initialize(grpcAddress string, id string, dir string) (err error) {
+	fs.grpcAddress = grpcAddress
+	fs.id = id
+	fs.dir = dir
+	return nil
+}
+
+func (fs *FilerSource) ReadPart(part string) (readCloser io.ReadCloser, err error) {
+
+	vid2Locations := make(map[string]*filer_pb.Locations)
+
+	vid := volumeId(part)
+
+	err = fs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+		glog.V(4).Infof("read lookup volume id locations: %v", vid)
+		resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
+			VolumeIds: []string{vid},
+		})
+		if err != nil {
+			return err
+		}
+
+		vid2Locations = resp.LocationsMap
+
+		return nil
+	})
+
+	if err != nil {
+		glog.V(1).Infof("replication lookup volume id: %v", vid, err)
+		return nil, fmt.Errorf("replicationlookup volume id %v: %v", vid, err)
+	}
+
+	locations := vid2Locations[vid]
+
+	if locations == nil || len(locations.Locations) == 0 {
+		glog.V(1).Infof("replication locate volume id: %v", vid, err)
+		return nil, fmt.Errorf("replication locate volume id %v: %v", vid, err)
+	}
+
+	fileUrl := fmt.Sprintf("http://%s/%s", locations.Locations[0].Url, part)
+
+	_, readCloser, err = util.DownloadUrl(fileUrl)
+
+	return readCloser, err
+}
+
+func (fs *FilerSource) withFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+
+	grpcConnection, err := util.GrpcDial(fs.grpcAddress)
+	if err != nil {
+		return fmt.Errorf("fail to dial %s: %v", fs.grpcAddress, err)
+	}
+	defer grpcConnection.Close()
+
+	client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+
+	return fn(client)
+}
+
+func volumeId(fileId string) string {
+	lastCommaIndex := strings.LastIndex(fileId, ",")
+	if lastCommaIndex > 0 {
+		return fileId[:lastCommaIndex]
+	}
+	return fileId
+}
diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go
index 98e8699b0..6c3390a26 100644
--- a/weed/server/filer_grpc_server.go
+++ b/weed/server/filer_grpc_server.go
@@ -188,7 +188,7 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr
 		}
 	}
 
-	fs.filer.NotifyUpdateEvent(entry, newEntry)
+	fs.filer.NotifyUpdateEvent(entry, newEntry, true)
 
 	return &filer_pb.UpdateEntryResponse{}, err
 }