2 files changed, 24 insertions, 6 deletions
diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go
index a0c53cae0..db395e3c3 100644
--- a/weed/server/filer_grpc_server_sub_meta.go
+++ b/weed/server/filer_grpc_server_sub_meta.go
@@ -23,9 +23,12 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 
 	peerAddress := findClientAddress(stream.Context(), 0)
 
-	clientName := fs.addClient(req.ClientName, peerAddress)
+	alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId)
+	if alreadyKnown {
+		return fmt.Errorf("duplicated subscription detected for client %s id %d", clientName, req.ClientId)
+	}
 
-	defer fs.deleteClient(clientName)
+	defer fs.deleteClient(clientName, req.ClientId)
 
 	lastReadTime := time.Unix(0, req.SinceNs)
 	glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
@@ -81,9 +84,9 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 
 	peerAddress := findClientAddress(stream.Context(), 0)
 
-	clientName := fs.addClient(req.ClientName, peerAddress)
+	_, clientName := fs.addClient(req.ClientName, peerAddress, 0)
 
-	defer fs.deleteClient(clientName)
+	defer fs.deleteClient(clientName, 0)
 
 	lastReadTime := time.Unix(0, req.SinceNs)
 	glog.V(0).Infof(" %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime)
@@ -237,12 +240,22 @@ func hasPrefixIn(text string, prefixes []string) bool {
 	return false
 }
 
-func (fs *FilerServer) addClient(clientType string, clientAddress string) (clientName string) {
+func (fs *FilerServer) addClient(clientType string, clientAddress string, clientId int32) (alreadyKnown bool, clientName string) {
 	clientName = clientType + "@" + clientAddress
 	glog.V(0).Infof("+ listener %v", clientName)
+	if clientId != 0 {
+		fs.knownListenersLock.Lock()
+		_, alreadyKnown = fs.knownListeners[clientId]
+		fs.knownListenersLock.Unlock()
+	}
 	return
 }
 
-func (fs *FilerServer) deleteClient(clientName string) {
+func (fs *FilerServer) deleteClient(clientName string, clientId int32) {
 	glog.V(0).Infof("- listener %v", clientName)
+	if clientId != 0 {
+		fs.knownListenersLock.Lock()
+		delete(fs.knownListeners, clientId)
+		fs.knownListenersLock.Unlock()
+	}
 }
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index da385a116..e51299c6d 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -82,6 +82,10 @@ type FilerServer struct {
 	listenersLock sync.Mutex
 	listenersCond *sync.Cond
 
+	// track known metadata listeners
+	knownListenersLock sync.Mutex
+	knownListeners     map[int32]struct{}
+
 	brokers     map[string]map[string]bool
 	brokersLock sync.Mutex
 
@@ -103,6 +107,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs = &FilerServer{
 		option:                option,
 		grpcDialOption:        security.LoadClientTLS(util.GetViper(), "grpc.filer"),
+		knownListeners:        make(map[int32]struct{}),
 		brokers:               make(map[string]map[string]bool),
 		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
 	}