fix on finding lock owner

weed/cluster/lock_manager/distributed_lock_manager.go

@@ -55,7 +55,7 @@ func (dlm *DistributedLockManager) FindLockOwner(key string) (owner string, move
 		return
 	}
 	if movedTo != dlm.Host {
-		glog.V(0).Infof("lock %s not on %s: to %s", key, dlm.Host, movedTo)
+		glog.V(0).Infof("lock %s not on current %s but on %s", key, dlm.Host, movedTo)
 		return
 	}
 	owner, err = dlm.lockManager.GetLockOwner(key)

weed/server/filer_grpc_server_dlm.go

@@ -83,7 +83,7 @@ func (fs *FilerServer) DistributedUnlock(ctx context.Context, req *filer_pb.Unlo
 
 func (fs *FilerServer) FindLockOwner(ctx context.Context, req *filer_pb.FindLockOwnerRequest) (*filer_pb.FindLockOwnerResponse, error) {
 	owner, movedTo, err := fs.filer.Dlm.FindLockOwner(req.Name)
-	if !req.IsMoved && movedTo != "" && err == lock_manager.LockNotFound {
+	if !req.IsMoved && movedTo != "" || err == lock_manager.LockNotFound {
 		err = pb.WithFilerClient(false, 0, movedTo, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 			secondResp, err := client.FindLockOwner(context.Background(), &filer_pb.FindLockOwnerRequest{
 				Name:    req.Name,

weed/shell/command_mq_topic_list.go

@@ -55,7 +55,7 @@ func findBrokerBalancer(commandEnv *CommandEnv) (brokerBalancer string, err erro
 			Name: pub_balancer.LockBrokerBalancer,
 		})
 		if err != nil {
-			return err
+			return fmt.Errorf("FindLockOwner: %v", err)
 		}
 		brokerBalancer = resp.Owner
 		return nil