remove old code

24 files changed, 299 insertions, 3516 deletions

diff --git a/weed/mq/broker.go b/weed/mq/broker.go
new file mode 100644
index 000000000..8debcec0b
--- /dev/null
+++ b/weed/mq/broker.go
@@ -0,0 +1,12 @@
+package mq
+
+const LAST_MINUTES = 10
+
+type TopicStat struct {
+	MessageCounts [LAST_MINUTES]int64
+	ByteCounts    [LAST_MINUTES]int64
+}
+
+func NewTopicStat() *TopicStat {
+	return &TopicStat{}
+}
diff --git a/weed/mq/broker/broker_append.go b/weed/mq/broker/broker_append.go
deleted file mode 100644
index c8e0da93c..000000000
--- a/weed/mq/broker/broker_append.go
+++ /dev/null
@@ -1,130 +0,0 @@
-package broker
-
-import (
-	"context"
-	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/security"
-	"io"
-
-	"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/pb/mq_pb"
-	"github.com/chrislusf/seaweedfs/weed/util"
-)
-
-func (broker *MessageQueueBroker) appendToFile(targetFile string, topicConfig *mq_pb.TopicConfiguration, data []byte) error {
-
-	assignResult, uploadResult, err2 := broker.assignAndUpload(topicConfig, data)
-	if err2 != nil {
-		return err2
-	}
-
-	dir, name := util.FullPath(targetFile).DirAndName()
-
-	// append the chunk
-	if err := broker.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-
-		request := &filer_pb.AppendToEntryRequest{
-			Directory: dir,
-			EntryName: name,
-			Chunks:    []*filer_pb.FileChunk{uploadResult.ToPbFileChunk(assignResult.Fid, 0)},
-		}
-
-		_, err := client.AppendToEntry(context.Background(), request)
-		if err != nil {
-			glog.V(0).Infof("append to file %v: %v", request, err)
-			return err
-		}
-
-		return nil
-	}); err != nil {
-		return fmt.Errorf("append to file %v: %v", targetFile, err)
-	}
-
-	return nil
-}
-
-func (broker *MessageQueueBroker) assignAndUpload(topicConfig *mq_pb.TopicConfiguration, data []byte) (*operation.AssignResult, *operation.UploadResult, error) {
-
-	var assignResult = &operation.AssignResult{}
-
-	// assign a volume location
-	if err := broker.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-
-		assignErr := util.Retry("assignVolume", func() error {
-			request := &filer_pb.AssignVolumeRequest{
-				Count:       1,
-				Replication: topicConfig.Replication,
-				Collection:  topicConfig.Collection,
-			}
-
-			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)
-			}
-
-			assignResult.Auth = security.EncodedJwt(resp.Auth)
-			assignResult.Fid = resp.FileId
-			assignResult.Url = resp.Location.Url
-			assignResult.PublicUrl = resp.Location.PublicUrl
-			assignResult.GrpcPort = int(resp.Location.GrpcPort)
-			assignResult.Count = uint64(resp.Count)
-
-			return nil
-		})
-		if assignErr != nil {
-			return assignErr
-		}
-
-		return nil
-	}); err != nil {
-		return nil, nil, err
-	}
-
-	// upload data
-	targetUrl := fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid)
-	uploadOption := &operation.UploadOption{
-		UploadUrl:         targetUrl,
-		Filename:          "",
-		Cipher:            broker.option.Cipher,
-		IsInputCompressed: false,
-		MimeType:          "",
-		PairMap:           nil,
-		Jwt:               assignResult.Auth,
-	}
-	uploadResult, err := operation.UploadData(data, uploadOption)
-	if err != nil {
-		return nil, nil, fmt.Errorf("upload data %s: %v", targetUrl, err)
-	}
-	// println("uploaded to", targetUrl)
-	return assignResult, uploadResult, nil
-}
-
-var _ = filer_pb.FilerClient(&MessageQueueBroker{})
-
-func (broker *MessageQueueBroker) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) (err error) {
-
-	for _, filer := range broker.option.Filers {
-		if err = pb.WithFilerClient(streamingMode, filer, broker.grpcDialOption, fn); err != nil {
-			if err == io.EOF {
-				return
-			}
-			glog.V(0).Infof("fail to connect to %s: %v", filer, err)
-		} else {
-			break
-		}
-	}
-
-	return
-
-}
-
-func (broker *MessageQueueBroker) AdjustedUrl(location *filer_pb.Location) string {
-	return location.Url
-}
diff --git a/weed/mq/broker/broker_grpc_server.go b/weed/mq/broker/broker_grpc_server.go
deleted file mode 100644
index 2cb4187ae..000000000
--- a/weed/mq/broker/broker_grpc_server.go
+++ /dev/null
@@ -1,37 +0,0 @@
-package broker
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-)
-
-func (broker *MessageQueueBroker) ConfigureTopic(c context.Context, request *mq_pb.ConfigureTopicRequest) (*mq_pb.ConfigureTopicResponse, error) {
-	panic("implement me")
-}
-
-func (broker *MessageQueueBroker) DeleteTopic(c context.Context, request *mq_pb.DeleteTopicRequest) (*mq_pb.DeleteTopicResponse, error) {
-	resp := &mq_pb.DeleteTopicResponse{}
-	dir, entry := genTopicDirEntry(request.Namespace, request.Topic)
-	if exists, err := filer_pb.Exists(broker, dir, entry, true); err != nil {
-		return nil, err
-	} else if exists {
-		err = filer_pb.Remove(broker, dir, entry, true, true, true, false, nil)
-	}
-	return resp, nil
-}
-
-func (broker *MessageQueueBroker) GetTopicConfiguration(c context.Context, request *mq_pb.GetTopicConfigurationRequest) (*mq_pb.GetTopicConfigurationResponse, error) {
-	panic("implement me")
-}
-
-func genTopicDir(namespace, topic string) string {
-	return fmt.Sprintf("%s/%s/%s", filer.TopicsDir, namespace, topic)
-}
-
-func genTopicDirEntry(namespace, topic string) (dir, entry string) {
-	return fmt.Sprintf("%s/%s", filer.TopicsDir, namespace), topic
-}
diff --git a/weed/mq/broker/broker_grpc_server_discovery.go b/weed/mq/broker/broker_grpc_server_discovery.go
index e276091a9..94e89cd41 100644
--- a/weed/mq/broker/broker_grpc_server_discovery.go
+++ b/weed/mq/broker/broker_grpc_server_discovery.go
@@ -2,7 +2,6 @@ package broker
 
 import (
 	"context"
-	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"time"
@@ -10,57 +9,8 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
 )
 
-/*
-Topic discovery:
-
-When pub or sub connects, it ask for the whole broker list, and run consistent hashing to find the broker.
-
-The broker will check peers whether it is already hosted by some other broker, if that broker is alive and acknowledged alive, redirect to it.
-Otherwise, just host the topic.
-
-So, if the pub or sub connects around the same time, they would connect to the same broker. Everyone is happy.
-If one of the pub or sub connects very late, and the system topo changed quite a bit with new servers added or old servers died, checking peers will help.
-
-*/
-
-func (broker *MessageQueueBroker) FindBroker(c context.Context, request *mq_pb.FindBrokerRequest) (*mq_pb.FindBrokerResponse, error) {
-
-	t := &mq_pb.FindBrokerResponse{}
-	var peers []string
-
-	targetTopicPartition := fmt.Sprintf(TopicPartitionFmt, request.Namespace, request.Topic, request.Parition)
-
-	for _, filer := range broker.option.Filers {
-		err := broker.withFilerClient(false, filer, func(client filer_pb.SeaweedFilerClient) error {
-			resp, err := client.LocateBroker(context.Background(), &filer_pb.LocateBrokerRequest{
-				Resource: targetTopicPartition,
-			})
-			if err != nil {
-				return err
-			}
-			if resp.Found && len(resp.Resources) > 0 {
-				t.Broker = resp.Resources[0].GrpcAddresses
-				return nil
-			}
-			for _, b := range resp.Resources {
-				peers = append(peers, b.GrpcAddresses)
-			}
-			return nil
-		})
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	t.Broker = PickMember(peers, []byte(targetTopicPartition))
-
-	return t, nil
-
-}
-
 func (broker *MessageQueueBroker) checkFilers() {
 
 	// contact a filer about masters
diff --git a/weed/mq/broker/broker_grpc_server_publish.go b/weed/mq/broker/broker_grpc_server_publish.go
deleted file mode 100644
index eb76dd5dc..000000000
--- a/weed/mq/broker/broker_grpc_server_publish.go
+++ /dev/null
@@ -1,112 +0,0 @@
-package broker
-
-import (
-	"crypto/md5"
-	"fmt"
-	"io"
-
-	"github.com/golang/protobuf/proto"
-
-	"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/pb/mq_pb"
-)
-
-func (broker *MessageQueueBroker) Publish(stream mq_pb.SeaweedMessaging_PublishServer) error {
-
-	// process initial request
-	in, err := stream.Recv()
-	if err == io.EOF {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-
-	// TODO look it up
-	topicConfig := &mq_pb.TopicConfiguration{
-		// IsTransient: true,
-	}
-
-	// send init response
-	initResponse := &mq_pb.PublishResponse{
-		Config:   nil,
-		Redirect: nil,
-	}
-	err = stream.Send(initResponse)
-	if err != nil {
-		return err
-	}
-	if initResponse.Redirect != nil {
-		return nil
-	}
-
-	// get lock
-	tp := TopicPartition{
-		Namespace: in.Init.Namespace,
-		Topic:     in.Init.Topic,
-		Partition: in.Init.Partition,
-	}
-
-	tpDir := fmt.Sprintf("%s/%s/%s", filer.TopicsDir, tp.Namespace, tp.Topic)
-	md5File := fmt.Sprintf("p%02d.md5", tp.Partition)
-	// println("chan data stored under", tpDir, "as", md5File)
-
-	if exists, err := filer_pb.Exists(broker, tpDir, md5File, false); err == nil && exists {
-		return fmt.Errorf("channel is already closed")
-	}
-
-	tl := broker.topicManager.RequestLock(tp, topicConfig, true)
-	defer broker.topicManager.ReleaseLock(tp, true)
-
-	md5hash := md5.New()
-	// process each message
-	for {
-		// println("recv")
-		in, err := stream.Recv()
-		// glog.V(0).Infof("recieved %v err: %v", in, err)
-		if err == io.EOF {
-			return nil
-		}
-		if err != nil {
-			return err
-		}
-
-		if in.Data == nil {
-			continue
-		}
-
-		// fmt.Printf("received: %d : %s\n", len(in.Data.Value), string(in.Data.Value))
-
-		data, err := proto.Marshal(in.Data)
-		if err != nil {
-			glog.Errorf("marshall error: %v\n", err)
-			continue
-		}
-
-		tl.logBuffer.AddToBuffer(in.Data.Key, data, in.Data.EventTimeNs)
-
-		if in.Data.IsClose {
-			// println("server received closing")
-			break
-		}
-
-		md5hash.Write(in.Data.Value)
-
-	}
-
-	if err := broker.appendToFile(tpDir+"/"+md5File, topicConfig, md5hash.Sum(nil)); err != nil {
-		glog.V(0).Infof("err writing %s: %v", md5File, err)
-	}
-
-	// fmt.Printf("received md5 %X\n", md5hash.Sum(nil))
-
-	// send the close ack
-	// println("server send ack closing")
-	if err := stream.Send(&mq_pb.PublishResponse{IsClosed: true}); err != nil {
-		glog.V(0).Infof("err sending close response: %v", err)
-	}
-	return nil
-
-}
diff --git a/weed/mq/broker/broker_grpc_server_subscribe.go b/weed/mq/broker/broker_grpc_server_subscribe.go
deleted file mode 100644
index 3743218b1..000000000
--- a/weed/mq/broker/broker_grpc_server_subscribe.go
+++ /dev/null
@@ -1,178 +0,0 @@
-package broker
-
-import (
-	"fmt"
-	"github.com/chrislusf/seaweedfs/weed/util"
-	"github.com/chrislusf/seaweedfs/weed/util/log_buffer"
-	"io"
-	"strings"
-	"time"
-
-	"github.com/golang/protobuf/proto"
-
-	"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/pb/mq_pb"
-)
-
-func (broker *MessageQueueBroker) Subscribe(stream mq_pb.SeaweedMessaging_SubscribeServer) error {
-
-	// process initial request
-	in, err := stream.Recv()
-	if err == io.EOF {
-		return nil
-	}
-	if err != nil {
-		return err
-	}
-
-	var processedTsNs int64
-	var messageCount int64
-	subscriberId := in.Init.SubscriberId
-
-	// TODO look it up
-	topicConfig := &mq_pb.TopicConfiguration{
-		// IsTransient: true,
-	}
-
-	// get lock
-	tp := TopicPartition{
-		Namespace: in.Init.Namespace,
-		Topic:     in.Init.Topic,
-		Partition: in.Init.Partition,
-	}
-	fmt.Printf("+ subscriber %s for %s\n", subscriberId, tp.String())
-	defer func() {
-		fmt.Printf("- subscriber %s for %s %d messages last %v\n", subscriberId, tp.String(), messageCount, time.Unix(0, processedTsNs))
-	}()
-
-	lock := broker.topicManager.RequestLock(tp, topicConfig, false)
-	defer broker.topicManager.ReleaseLock(tp, false)
-
-	isConnected := true
-	go func() {
-		for isConnected {
-			if _, err := stream.Recv(); err != nil {
-				// println("disconnecting connection to", subscriberId, tp.String())
-				isConnected = false
-				lock.cond.Signal()
-			}
-		}
-	}()
-
-	lastReadTime := time.Now()
-	switch in.Init.StartPosition {
-	case mq_pb.SubscriberMessage_InitMessage_TIMESTAMP:
-		lastReadTime = time.Unix(0, in.Init.TimestampNs)
-	case mq_pb.SubscriberMessage_InitMessage_LATEST:
-	case mq_pb.SubscriberMessage_InitMessage_EARLIEST:
-		lastReadTime = time.Unix(0, 0)
-	}
-
-	// how to process each message
-	// an error returned will end the subscription
-	eachMessageFn := func(m *mq_pb.Message) error {
-		err := stream.Send(&mq_pb.BrokerMessage{
-			Data: m,
-		})
-		if err != nil {
-			glog.V(0).Infof("=> subscriber %v: %+v", subscriberId, err)
-		}
-		return err
-	}
-
-	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) error {
-		m := &mq_pb.Message{}
-		if err = proto.Unmarshal(logEntry.Data, m); err != nil {
-			glog.Errorf("unexpected unmarshal mq_pb.Message: %v", err)
-			return err
-		}
-		// fmt.Printf("sending : %d bytes ts %d\n", len(m.Value), logEntry.TsNs)
-		if err = eachMessageFn(m); err != nil {
-			glog.Errorf("sending %d bytes to %s: %s", len(m.Value), subscriberId, err)
-			return err
-		}
-		if m.IsClose {
-			// println("processed EOF")
-			return io.EOF
-		}
-		processedTsNs = logEntry.TsNs
-		messageCount++
-		return nil
-	}
-
-	// fmt.Printf("subscriber %s read %d on disk log %v\n", subscriberId, messageCount, lastReadTime)
-
-	for {
-
-		if err = broker.readPersistedLogBuffer(&tp, lastReadTime, eachLogEntryFn); err != nil {
-			if err != io.EOF {
-				// println("stopping from persisted logs", err.Error())
-				return err
-			}
-		}
-
-		if processedTsNs != 0 {
-			lastReadTime = time.Unix(0, processedTsNs)
-		}
-
-		lastReadTime, _, err = lock.logBuffer.LoopProcessLogData("broker", lastReadTime, 0, func() bool {
-			lock.Mutex.Lock()
-			lock.cond.Wait()
-			lock.Mutex.Unlock()
-			return isConnected
-		}, eachLogEntryFn)
-		if err != nil {
-			if err == log_buffer.ResumeFromDiskError {
-				continue
-			}
-			glog.Errorf("processed to %v: %v", lastReadTime, err)
-			if err != log_buffer.ResumeError {
-				break
-			}
-		}
-	}
-
-	return err
-
-}
-
-func (broker *MessageQueueBroker) readPersistedLogBuffer(tp *TopicPartition, startTime time.Time, eachLogEntryFn func(logEntry *filer_pb.LogEntry) error) (err error) {
-	startTime = startTime.UTC()
-	startDate := fmt.Sprintf("%04d-%02d-%02d", startTime.Year(), startTime.Month(), startTime.Day())
-	startHourMinute := fmt.Sprintf("%02d-%02d", startTime.Hour(), startTime.Minute())
-
-	sizeBuf := make([]byte, 4)
-	startTsNs := startTime.UnixNano()
-
-	topicDir := genTopicDir(tp.Namespace, tp.Topic)
-	partitionSuffix := fmt.Sprintf(".part%02d", tp.Partition)
-
-	return filer_pb.List(broker, topicDir, "", func(dayEntry *filer_pb.Entry, isLast bool) error {
-		dayDir := fmt.Sprintf("%s/%s", topicDir, dayEntry.Name)
-		return filer_pb.List(broker, dayDir, "", func(hourMinuteEntry *filer_pb.Entry, isLast bool) error {
-			if dayEntry.Name == startDate {
-				hourMinute := util.FileNameBase(hourMinuteEntry.Name)
-				if strings.Compare(hourMinute, startHourMinute) < 0 {
-					return nil
-				}
-			}
-			if !strings.HasSuffix(hourMinuteEntry.Name, partitionSuffix) {
-				return nil
-			}
-			// println("partition", tp.Partition, "processing", dayDir, "/", hourMinuteEntry.Name)
-			chunkedFileReader := filer.NewChunkStreamReader(broker, hourMinuteEntry.Chunks)
-			defer chunkedFileReader.Close()
-			if _, err := filer.ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, 0, eachLogEntryFn); err != nil {
-				chunkedFileReader.Close()
-				if err == io.EOF {
-					return err
-				}
-				return fmt.Errorf("reading %s/%s: %v", dayDir, hourMinuteEntry.Name, err)
-			}
-			return nil
-		}, "", false, 24*60)
-	}, startDate, true, 366)
-
-}
diff --git a/weed/mq/broker/broker_server.go b/weed/mq/broker/broker_server.go
index 36f216a48..63e248797 100644
--- a/weed/mq/broker/broker_server.go
+++ b/weed/mq/broker/broker_server.go
@@ -1,15 +1,11 @@
 package broker
 
 import (
-	"context"
 	"github.com/chrislusf/seaweedfs/weed/cluster"
 	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
-	"time"
-
 	"google.golang.org/grpc"
 
-	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -33,7 +29,6 @@ type MessageQueueBroker struct {
 	option         *MessageQueueBrokerOption
 	grpcDialOption grpc.DialOption
 	MasterClient   *wdclient.MasterClient
-	topicManager   *TopicManager
 }
 
 func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.DialOption) (mqBroker *MessageQueueBroker, err error) {
@@ -44,72 +39,13 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
 		MasterClient:   wdclient.NewMasterClient(grpcDialOption, option.FilerGroup, cluster.BrokerType, pb.NewServerAddress(option.Ip, option.Port, 0), option.DataCenter, option.Rack, option.Masters),
 	}
 
-	mqBroker.topicManager = NewTopicManager(mqBroker)
-
 	mqBroker.checkFilers()
 
-	go mqBroker.keepConnectedToOneFiler()
 	go mqBroker.MasterClient.KeepConnectedToMaster()
 
 	return mqBroker, nil
 }
 
-func (broker *MessageQueueBroker) keepConnectedToOneFiler() {
-
-	for {
-		for _, filer := range broker.option.Filers {
-			broker.withFilerClient(false, filer, func(client filer_pb.SeaweedFilerClient) error {
-				ctx, cancel := context.WithCancel(context.Background())
-				defer cancel()
-				stream, err := client.KeepConnected(ctx)
-				if err != nil {
-					glog.V(0).Infof("%s:%d failed to keep connected to %s: %v", broker.option.Ip, broker.option.Port, filer, err)
-					return err
-				}
-
-				initRequest := &filer_pb.KeepConnectedRequest{
-					Name:     broker.option.Ip,
-					GrpcPort: uint32(broker.option.Port),
-				}
-				for _, tp := range broker.topicManager.ListTopicPartitions() {
-					initRequest.Resources = append(initRequest.Resources, tp.String())
-				}
-				if err := stream.Send(&filer_pb.KeepConnectedRequest{
-					Name:     broker.option.Ip,
-					GrpcPort: uint32(broker.option.Port),
-				}); err != nil {
-					glog.V(0).Infof("broker %s:%d failed to init at %s: %v", broker.option.Ip, broker.option.Port, filer, err)
-					return err
-				}
-
-				// TODO send events of adding/removing topics
-
-				glog.V(0).Infof("conntected with filer: %v", filer)
-				for {
-					if err := stream.Send(&filer_pb.KeepConnectedRequest{
-						Name:     broker.option.Ip,
-						GrpcPort: uint32(broker.option.Port),
-					}); err != nil {
-						glog.V(0).Infof("%s:%d failed to sendto %s: %v", broker.option.Ip, broker.option.Port, filer, err)
-						return err
-					}
-					// println("send heartbeat")
-					if _, err := stream.Recv(); err != nil {
-						glog.V(0).Infof("%s:%d failed to receive from %s: %v", broker.option.Ip, broker.option.Port, filer, err)
-						return err
-					}
-					// println("received reply")
-					time.Sleep(11 * time.Second)
-					// println("woke up")
-				}
-				return nil
-			})
-			time.Sleep(3 * time.Second)
-		}
-	}
-
-}
-
 func (broker *MessageQueueBroker) withFilerClient(streamingMode bool, filer pb.ServerAddress, fn func(filer_pb.SeaweedFilerClient) error) error {
 
 	return pb.WithFilerClient(streamingMode, filer, broker.grpcDialOption, fn)
diff --git a/weed/mq/broker/topic_manager.go b/weed/mq/broker/topic_manager.go
deleted file mode 100644
index 34f063d9a..000000000
--- a/weed/mq/broker/topic_manager.go
+++ /dev/null
@@ -1,124 +0,0 @@
-package broker
-
-import (
-	"fmt"
-	"sync"
-	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/filer"
-	"github.com/chrislusf/seaweedfs/weed/glog"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-	"github.com/chrislusf/seaweedfs/weed/util/log_buffer"
-)
-
-type TopicPartition struct {
-	Namespace string
-	Topic     string
-	Partition int32
-}
-
-const (
-	TopicPartitionFmt = "%s/%s_%02d"
-)
-
-func (tp *TopicPartition) String() string {
-	return fmt.Sprintf(TopicPartitionFmt, tp.Namespace, tp.Topic, tp.Partition)
-}
-
-type TopicControl struct {
-	sync.Mutex
-	cond            *sync.Cond
-	subscriberCount int
-	publisherCount  int
-	logBuffer       *log_buffer.LogBuffer
-}
-
-type TopicManager struct {
-	sync.Mutex
-	topicControls map[TopicPartition]*TopicControl
-	broker        *MessageQueueBroker
-}
-
-func NewTopicManager(messageBroker *MessageQueueBroker) *TopicManager {
-	return &TopicManager{
-		topicControls: make(map[TopicPartition]*TopicControl),
-		broker:        messageBroker,
-	}
-}
-
-func (tm *TopicManager) buildLogBuffer(tl *TopicControl, tp TopicPartition, topicConfig *mq_pb.TopicConfiguration) *log_buffer.LogBuffer {
-
-	flushFn := func(startTime, stopTime time.Time, buf []byte) {
-
-		if topicConfig.IsTransient {
-			// return
-		}
-
-		// fmt.Printf("flushing with topic config %+v\n", topicConfig)
-
-		startTime, stopTime = startTime.UTC(), stopTime.UTC()
-		targetFile := fmt.Sprintf(
-			"%s/%s/%s/%04d-%02d-%02d/%02d-%02d.part%02d",
-			filer.TopicsDir, tp.Namespace, tp.Topic,
-			startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(),
-			tp.Partition,
-		)
-
-		if err := tm.broker.appendToFile(targetFile, topicConfig, buf); err != nil {
-			glog.V(0).Infof("log write failed %s: %v", targetFile, err)
-		}
-	}
-	logBuffer := log_buffer.NewLogBuffer("broker", time.Minute, flushFn, func() {
-		tl.cond.Broadcast()
-	})
-
-	return logBuffer
-}
-
-func (tm *TopicManager) RequestLock(partition TopicPartition, topicConfig *mq_pb.TopicConfiguration, isPublisher bool) *TopicControl {
-	tm.Lock()
-	defer tm.Unlock()
-
-	tc, found := tm.topicControls[partition]
-	if !found {
-		tc = &TopicControl{}
-		tc.cond = sync.NewCond(&tc.Mutex)
-		tm.topicControls[partition] = tc
-		tc.logBuffer = tm.buildLogBuffer(tc, partition, topicConfig)
-	}
-	if isPublisher {
-		tc.publisherCount++
-	} else {
-		tc.subscriberCount++
-	}
-	return tc
-}
-
-func (tm *TopicManager) ReleaseLock(partition TopicPartition, isPublisher bool) {
-	tm.Lock()
-	defer tm.Unlock()
-
-	lock, found := tm.topicControls[partition]
-	if !found {
-		return
-	}
-	if isPublisher {
-		lock.publisherCount--
-	} else {
-		lock.subscriberCount--
-	}
-	if lock.subscriberCount <= 0 && lock.publisherCount <= 0 {
-		delete(tm.topicControls, partition)
-		lock.logBuffer.Shutdown()
-	}
-}
-
-func (tm *TopicManager) ListTopicPartitions() (tps []TopicPartition) {
-	tm.Lock()
-	defer tm.Unlock()
-
-	for k := range tm.topicControls {
-		tps = append(tps, k)
-	}
-	return
-}
diff --git a/weed/mq/msgclient/chan_config.go b/weed/mq/msgclient/chan_config.go
deleted file mode 100644
index a75678815..000000000
--- a/weed/mq/msgclient/chan_config.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package msgclient
-
-func (mc *MessagingClient) DeleteChannel(chanName string) error {
-	return mc.DeleteTopic("chan", chanName)
-}
diff --git a/weed/mq/msgclient/chan_pub.go b/weed/mq/msgclient/chan_pub.go
deleted file mode 100644
index f4ffe832a..000000000
--- a/weed/mq/msgclient/chan_pub.go
+++ /dev/null
@@ -1,76 +0,0 @@
-package msgclient
-
-import (
-	"crypto/md5"
-	"hash"
-	"io"
-	"log"
-
-	"google.golang.org/grpc"
-
-	"github.com/chrislusf/seaweedfs/weed/mq/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-)
-
-type PubChannel struct {
-	client         mq_pb.SeaweedMessaging_PublishClient
-	grpcConnection *grpc.ClientConn
-	md5hash        hash.Hash
-}
-
-func (mc *MessagingClient) NewPubChannel(chanName string) (*PubChannel, error) {
-	tp := broker.TopicPartition{
-		Namespace: "chan",
-		Topic:     chanName,
-		Partition: 0,
-	}
-	grpcConnection, err := mc.findBroker(tp)
-	if err != nil {
-		return nil, err
-	}
-	pc, err := setupPublisherClient(grpcConnection, tp)
-	if err != nil {
-		return nil, err
-	}
-	return &PubChannel{
-		client:         pc,
-		grpcConnection: grpcConnection,
-		md5hash:        md5.New(),
-	}, nil
-}
-
-func (pc *PubChannel) Publish(m []byte) error {
-	err := pc.client.Send(&mq_pb.PublishRequest{
-		Data: &mq_pb.Message{
-			Value: m,
-		},
-	})
-	if err == nil {
-		pc.md5hash.Write(m)
-	}
-	return err
-}
-func (pc *PubChannel) Close() error {
-
-	// println("send closing")
-	if err := pc.client.Send(&mq_pb.PublishRequest{
-		Data: &mq_pb.Message{
-			IsClose: true,
-		},
-	}); err != nil {
-		log.Printf("err send close: %v", err)
-	}
-	// println("receive closing")
-	if _, err := pc.client.Recv(); err != nil && err != io.EOF {
-		log.Printf("err receive close: %v", err)
-	}
-	// println("close connection")
-	if err := pc.grpcConnection.Close(); err != nil {
-		log.Printf("err connection close: %v", err)
-	}
-	return nil
-}
-
-func (pc *PubChannel) Md5() []byte {
-	return pc.md5hash.Sum(nil)
-}
diff --git a/weed/mq/msgclient/chan_sub.go b/weed/mq/msgclient/chan_sub.go
deleted file mode 100644
index 859b482ef..000000000
--- a/weed/mq/msgclient/chan_sub.go
+++ /dev/null
@@ -1,85 +0,0 @@
-package msgclient
-
-import (
-	"context"
-	"crypto/md5"
-	"hash"
-	"io"
-	"log"
-	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/mq/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-)
-
-type SubChannel struct {
-	ch      chan []byte
-	stream  mq_pb.SeaweedMessaging_SubscribeClient
-	md5hash hash.Hash
-	cancel  context.CancelFunc
-}
-
-func (mc *MessagingClient) NewSubChannel(subscriberId, chanName string) (*SubChannel, error) {
-	tp := broker.TopicPartition{
-		Namespace: "chan",
-		Topic:     chanName,
-		Partition: 0,
-	}
-	grpcConnection, err := mc.findBroker(tp)
-	if err != nil {
-		return nil, err
-	}
-	ctx, cancel := context.WithCancel(context.Background())
-	sc, err := setupSubscriberClient(ctx, grpcConnection, tp, subscriberId, time.Unix(0, 0))
-	if err != nil {
-		return nil, err
-	}
-
-	t := &SubChannel{
-		ch:      make(chan []byte),
-		stream:  sc,
-		md5hash: md5.New(),
-		cancel:  cancel,
-	}
-
-	go func() {
-		for {
-			resp, subErr := t.stream.Recv()
-			if subErr == io.EOF {
-				return
-			}
-			if subErr != nil {
-				log.Printf("fail to receive from netchan %s: %v", chanName, subErr)
-				return
-			}
-			if resp.Data == nil {
-				// this could be heartbeat from broker
-				continue
-			}
-			if resp.Data.IsClose {
-				t.stream.Send(&mq_pb.SubscriberMessage{
-					IsClose: true,
-				})
-				close(t.ch)
-				cancel()
-				return
-			}
-			t.ch <- resp.Data.Value
-			t.md5hash.Write(resp.Data.Value)
-		}
-	}()
-
-	return t, nil
-}
-
-func (sc *SubChannel) Channel() chan []byte {
-	return sc.ch
-}
-
-func (sc *SubChannel) Md5() []byte {
-	return sc.md5hash.Sum(nil)
-}
-
-func (sc *SubChannel) Cancel() {
-	sc.cancel()
-}
diff --git a/weed/mq/msgclient/client.go b/weed/mq/msgclient/client.go
deleted file mode 100644
index cc64f1acb..000000000
--- a/weed/mq/msgclient/client.go
+++ /dev/null
@@ -1,55 +0,0 @@
-package msgclient
-
-import (
-	"context"
-	"fmt"
-	"log"
-
-	"google.golang.org/grpc"
-
-	"github.com/chrislusf/seaweedfs/weed/mq/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-	"github.com/chrislusf/seaweedfs/weed/security"
-	"github.com/chrislusf/seaweedfs/weed/util"
-)
-
-type MessagingClient struct {
-	bootstrapBrokers []string
-	grpcConnections  map[broker.TopicPartition]*grpc.ClientConn
-	grpcDialOption   grpc.DialOption
-}
-
-func NewMessagingClient(bootstrapBrokers ...string) *MessagingClient {
-	return &MessagingClient{
-		bootstrapBrokers: bootstrapBrokers,
-		grpcConnections:  make(map[broker.TopicPartition]*grpc.ClientConn),
-		grpcDialOption:   security.LoadClientTLS(util.GetViper(), "grpc.msg_client"),
-	}
-}
-
-func (mc *MessagingClient) findBroker(tp broker.TopicPartition) (*grpc.ClientConn, error) {
-
-	for _, broker := range mc.bootstrapBrokers {
-		grpcConnection, err := pb.GrpcDial(context.Background(), broker, mc.grpcDialOption)
-		if err != nil {
-			log.Printf("dial broker %s: %v", broker, err)
-			continue
-		}
-		defer grpcConnection.Close()
-
-		resp, err := mq_pb.NewSeaweedMessagingClient(grpcConnection).FindBroker(context.Background(),
-			&mq_pb.FindBrokerRequest{
-				Namespace: tp.Namespace,
-				Topic:     tp.Topic,
-				Parition:  tp.Partition,
-			})
-		if err != nil {
-			return nil, err
-		}
-
-		targetBroker := resp.Broker
-		return pb.GrpcDial(context.Background(), targetBroker, mc.grpcDialOption)
-	}
-	return nil, fmt.Errorf("no broker found for %+v", tp)
-}
diff --git a/weed/mq/msgclient/config.go b/weed/mq/msgclient/config.go
deleted file mode 100644
index 263ee856e..000000000
--- a/weed/mq/msgclient/config.go
+++ /dev/null
@@ -1,63 +0,0 @@
-package msgclient
-
-import (
-	"context"
-	"log"
-
-	"github.com/chrislusf/seaweedfs/weed/mq/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-)
-
-func (mc *MessagingClient) configureTopic(tp broker.TopicPartition) error {
-
-	return mc.withAnyBroker(func(client mq_pb.SeaweedMessagingClient) error {
-		_, err := client.ConfigureTopic(context.Background(),
-			&mq_pb.ConfigureTopicRequest{
-				Namespace: tp.Namespace,
-				Topic:     tp.Topic,
-				Configuration: &mq_pb.TopicConfiguration{
-					PartitionCount: 0,
-					Collection:     "",
-					Replication:    "",
-					IsTransient:    false,
-					Partitoning:    0,
-				},
-			})
-		return err
-	})
-
-}
-
-func (mc *MessagingClient) DeleteTopic(namespace, topic string) error {
-
-	return mc.withAnyBroker(func(client mq_pb.SeaweedMessagingClient) error {
-		_, err := client.DeleteTopic(context.Background(),
-			&mq_pb.DeleteTopicRequest{
-				Namespace: namespace,
-				Topic:     topic,
-			})
-		return err
-	})
-}
-
-func (mc *MessagingClient) withAnyBroker(fn func(client mq_pb.SeaweedMessagingClient) error) error {
-
-	var lastErr error
-	for _, broker := range mc.bootstrapBrokers {
-		grpcConnection, err := pb.GrpcDial(context.Background(), broker, mc.grpcDialOption)
-		if err != nil {
-			log.Printf("dial broker %s: %v", broker, err)
-			continue
-		}
-		defer grpcConnection.Close()
-
-		err = fn(mq_pb.NewSeaweedMessagingClient(grpcConnection))
-		if err == nil {
-			return nil
-		}
-		lastErr = err
-	}
-
-	return lastErr
-}
diff --git a/weed/mq/msgclient/publisher.go b/weed/mq/msgclient/publisher.go
deleted file mode 100644
index 823791d10..000000000
--- a/weed/mq/msgclient/publisher.go
+++ /dev/null
@@ -1,118 +0,0 @@
-package msgclient
-
-import (
-	"context"
-
-	"github.com/OneOfOne/xxhash"
-	"google.golang.org/grpc"
-
-	"github.com/chrislusf/seaweedfs/weed/mq/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-)
-
-type Publisher struct {
-	publishClients     []mq_pb.SeaweedMessaging_PublishClient
-	topicConfiguration *mq_pb.TopicConfiguration
-	messageCount       uint64
-	publisherId        string
-}
-
-func (mc *MessagingClient) NewPublisher(publisherId, namespace, topic string) (*Publisher, error) {
-	// read topic configuration
-	topicConfiguration := &mq_pb.TopicConfiguration{
-		PartitionCount: 4,
-	}
-	publishClients := make([]mq_pb.SeaweedMessaging_PublishClient, topicConfiguration.PartitionCount)
-	for i := 0; i < int(topicConfiguration.PartitionCount); i++ {
-		tp := broker.TopicPartition{
-			Namespace: namespace,
-			Topic:     topic,
-			Partition: int32(i),
-		}
-		grpcClientConn, err := mc.findBroker(tp)
-		if err != nil {
-			return nil, err
-		}
-		client, err := setupPublisherClient(grpcClientConn, tp)
-		if err != nil {
-			return nil, err
-		}
-		publishClients[i] = client
-	}
-	return &Publisher{
-		publishClients:     publishClients,
-		topicConfiguration: topicConfiguration,
-	}, nil
-}
-
-func setupPublisherClient(grpcConnection *grpc.ClientConn, tp broker.TopicPartition) (mq_pb.SeaweedMessaging_PublishClient, error) {
-
-	stream, err := mq_pb.NewSeaweedMessagingClient(grpcConnection).Publish(context.Background())
-	if err != nil {
-		return nil, err
-	}
-
-	// send init message
-	err = stream.Send(&mq_pb.PublishRequest{
-		Init: &mq_pb.PublishRequest_InitMessage{
-			Namespace: tp.Namespace,
-			Topic:     tp.Topic,
-			Partition: tp.Partition,
-		},
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	// process init response
-	initResponse, err := stream.Recv()
-	if err != nil {
-		return nil, err
-	}
-	if initResponse.Redirect != nil {
-		// TODO follow redirection
-	}
-	if initResponse.Config != nil {
-	}
-
-	// setup looks for control messages
-	doneChan := make(chan error, 1)
-	go func() {
-		for {
-			in, err := stream.Recv()
-			if err != nil {
-				doneChan <- err
-				return
-			}
-			if in.Redirect != nil {
-			}
-			if in.Config != nil {
-			}
-		}
-	}()
-
-	return stream, nil
-
-}
-
-func (p *Publisher) Publish(m *mq_pb.Message) error {
-	hashValue := p.messageCount
-	p.messageCount++
-	if p.topicConfiguration.Partitoning == mq_pb.TopicConfiguration_NonNullKeyHash {
-		if m.Key != nil {
-			hashValue = xxhash.Checksum64(m.Key)
-		}
-	} else if p.topicConfiguration.Partitoning == mq_pb.TopicConfiguration_KeyHash {
-		hashValue = xxhash.Checksum64(m.Key)
-	} else {
-		// round robin
-	}
-
-	idx := int(hashValue) % len(p.publishClients)
-	if idx < 0 {
-		idx += len(p.publishClients)
-	}
-	return p.publishClients[idx].Send(&mq_pb.PublishRequest{
-		Data: m,
-	})
-}
diff --git a/weed/mq/msgclient/subscriber.go b/weed/mq/msgclient/subscriber.go
deleted file mode 100644
index f3da40fb3..000000000
--- a/weed/mq/msgclient/subscriber.go
+++ /dev/null
@@ -1,120 +0,0 @@
-package msgclient
-
-import (
-	"context"
-	"io"
-	"sync"
-	"time"
-
-	"github.com/chrislusf/seaweedfs/weed/mq/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
-	"google.golang.org/grpc"
-)
-
-type Subscriber struct {
-	subscriberClients []mq_pb.SeaweedMessaging_SubscribeClient
-	subscriberCancels []context.CancelFunc
-	subscriberId      string
-}
-
-func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string, partitionId int, startTime time.Time) (*Subscriber, error) {
-	// read topic configuration
-	topicConfiguration := &mq_pb.TopicConfiguration{
-		PartitionCount: 4,
-	}
-	subscriberClients := make([]mq_pb.SeaweedMessaging_SubscribeClient, topicConfiguration.PartitionCount)
-	subscriberCancels := make([]context.CancelFunc, topicConfiguration.PartitionCount)
-
-	for i := 0; i < int(topicConfiguration.PartitionCount); i++ {
-		if partitionId >= 0 && i != partitionId {
-			continue
-		}
-		tp := broker.TopicPartition{
-			Namespace: namespace,
-			Topic:     topic,
-			Partition: int32(i),
-		}
-		grpcClientConn, err := mc.findBroker(tp)
-		if err != nil {
-			return nil, err
-		}
-		ctx, cancel := context.WithCancel(context.Background())
-		client, err := setupSubscriberClient(ctx, grpcClientConn, tp, subscriberId, startTime)
-		if err != nil {
-			return nil, err
-		}
-		subscriberClients[i] = client
-		subscriberCancels[i] = cancel
-	}
-
-	return &Subscriber{
-		subscriberClients: subscriberClients,
-		subscriberCancels: subscriberCancels,
-		subscriberId:      subscriberId,
-	}, nil
-}
-
-func setupSubscriberClient(ctx context.Context, grpcConnection *grpc.ClientConn, tp broker.TopicPartition, subscriberId string, startTime time.Time) (stream mq_pb.SeaweedMessaging_SubscribeClient, err error) {
-	stream, err = mq_pb.NewSeaweedMessagingClient(grpcConnection).Subscribe(ctx)
-	if err != nil {
-		return
-	}
-
-	// send init message
-	err = stream.Send(&mq_pb.SubscriberMessage{
-		Init: &mq_pb.SubscriberMessage_InitMessage{
-			Namespace:     tp.Namespace,
-			Topic:         tp.Topic,
-			Partition:     tp.Partition,
-			StartPosition: mq_pb.SubscriberMessage_InitMessage_TIMESTAMP,
-			TimestampNs:   startTime.UnixNano(),
-			SubscriberId:  subscriberId,
-		},
-	})
-	if err != nil {
-		return
-	}
-
-	return stream, nil
-}
-
-func doSubscribe(subscriberClient mq_pb.SeaweedMessaging_SubscribeClient, processFn func(m *mq_pb.Message)) error {
-	for {
-		resp, listenErr := subscriberClient.Recv()
-		if listenErr == io.EOF {
-			return nil
-		}
-		if listenErr != nil {
-			println(listenErr.Error())
-			return listenErr
-		}
-		if resp.Data == nil {
-			// this could be heartbeat from broker
-			continue
-		}
-		processFn(resp.Data)
-	}
-}
-
-// Subscribe starts goroutines to process the messages
-func (s *Subscriber) Subscribe(processFn func(m *mq_pb.Message)) {
-	var wg sync.WaitGroup
-	for i := 0; i < len(s.subscriberClients); i++ {
-		if s.subscriberClients[i] != nil {
-			wg.Add(1)
-			go func(subscriberClient mq_pb.SeaweedMessaging_SubscribeClient) {
-				defer wg.Done()
-				doSubscribe(subscriberClient, processFn)
-			}(s.subscriberClients[i])
-		}
-	}
-	wg.Wait()
-}
-
-func (s *Subscriber) Shutdown() {
-	for i := 0; i < len(s.subscriberClients); i++ {
-		if s.subscriberCancels[i] != nil {
-			s.subscriberCancels[i]()
-		}
-	}
-}
diff --git a/weed/mq/topic.go b/weed/mq/topic.go
new file mode 100644
index 000000000..fc1923af1
--- /dev/null
+++ b/weed/mq/topic.go
@@ -0,0 +1,23 @@
+package mq
+
+import "time"
+
+type Namespace string
+
+type Topic struct {
+	namespace Namespace
+	name      string
+}
+
+type Partition struct {
+	rangeStart int
+	rangeStop  int // exclusive
+	ringSize   int
+}
+
+type Segment struct {
+	topic        Topic
+	id           int32
+	partition    Partition
+	lastModified time.Time
+}
diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto
index 6277c3c6e..3f0ebe33c 100644
--- a/weed/pb/filer.proto
+++ b/weed/pb/filer.proto
@@ -60,12 +60,6 @@ service SeaweedFiler {
     rpc SubscribeLocalMetadata (SubscribeMetadataRequest) returns (stream SubscribeMetadataResponse) {
     }
 
-    rpc KeepConnected (stream KeepConnectedRequest) returns (stream KeepConnectedResponse) {
-    }
-
-    rpc LocateBroker (LocateBrokerRequest) returns (LocateBrokerResponse) {
-    }
-
     rpc KvGet (KvGetRequest) returns (KvGetResponse) {
     }
 
diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go
index 6fc2f6d8f..a57b4c70a 100644
--- a/weed/pb/filer_pb/filer.pb.go
+++ b/weed/pb/filer_pb/filer.pb.go
@@ -4180,216 +4180,204 @@ var file_filer_proto_rawDesc = []byte{
 	0x5f, 0x6e, 0x6f, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20,
 	0x01, 0x28, 0x0b, 0x32, 0x1b, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x45,
 	0x76, 0x65, 0x6e, 0x74, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x52, 0x11, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x69, 0x63, 0x61, 0x74,
-	0x69, 0x6f, 0x6e, 0x12, 0x13, 0x0a, 0x05, 0x74, 0x73, 0x5f, 0x6e, 0x73, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x03, 0x52, 0x04, 0x74, 0x73, 0x4e, 0x73, 0x22, 0x61, 0x0a, 0x08, 0x4c, 0x6f, 0x67, 0x45,
-	0x6e, 0x74, 0x72, 0x79, 0x12, 0x13, 0x0a, 0x05, 0x74, 0x73, 0x5f, 0x6e, 0x73, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x03, 0x52, 0x04, 0x74, 0x73, 0x4e, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x70, 0x61, 0x72,
-	0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x68, 0x61, 0x73, 0x68, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x10, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e,
-	0x4b, 0x65, 0x79, 0x48, 0x61, 0x73, 0x68, 0x12, 0x12, 0x0a, 0x04, 0x64, 0x61, 0x74, 0x61, 0x18,
-	0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x64, 0x61, 0x74, 0x61, 0x22, 0x65, 0x0a, 0x14, 0x4b,
-	0x65, 0x65, 0x70, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x67, 0x72, 0x70, 0x63, 0x5f,
-	0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x67, 0x72, 0x70, 0x63,
-	0x50, 0x6f, 0x72, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x73, 0x22, 0x17, 0x0a, 0x15, 0x4b, 0x65, 0x65, 0x70, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63,
-	0x74, 0x65, 0x64, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x31, 0x0a, 0x13, 0x4c,
-	0x6f, 0x63, 0x61, 0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0xcd,
-	0x01, 0x0a, 0x14, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x66, 0x6f, 0x75, 0x6e, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x08, 0x52, 0x05, 0x66, 0x6f, 0x75, 0x6e, 0x64, 0x12, 0x45, 0x0a,
-	0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b,
-	0x32, 0x27, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x63, 0x61,
-	0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0x73, 0x1a, 0x58, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x12, 0x25, 0x0a, 0x0e, 0x67, 0x72, 0x70, 0x63, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73,
-	0x65, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x67, 0x72, 0x70, 0x63, 0x41, 0x64,
-	0x64, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x12, 0x25, 0x0a, 0x0e, 0x72, 0x65, 0x73, 0x6f, 0x75,
-	0x72, 0x63, 0x65, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52,
-	0x0d, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x20,
-	0x0a, 0x0c, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10,
-	0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x6b, 0x65, 0x79,
-	0x22, 0x3b, 0x0a, 0x0d, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
-	0x65, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c,
-	0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x36, 0x0a,
-	0x0c, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a,
-	0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12,
-	0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x05,
-	0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, 0x25, 0x0a, 0x0d, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0xbd, 0x03, 0x0a,
-	0x09, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65,
-	0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x07, 0x76, 0x65, 0x72,
-	0x73, 0x69, 0x6f, 0x6e, 0x12, 0x3a, 0x0a, 0x09, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f,
-	0x70, 0x62, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x2e, 0x50, 0x61, 0x74,
-	0x68, 0x43, 0x6f, 0x6e, 0x66, 0x52, 0x09, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73,
-	0x1a, 0xd9, 0x02, 0x0a, 0x08, 0x50, 0x61, 0x74, 0x68, 0x43, 0x6f, 0x6e, 0x66, 0x12, 0x27, 0x0a,
-	0x0f, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x70, 0x72, 0x65, 0x66, 0x69, 0x78,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x50, 0x72, 0x65, 0x66, 0x69, 0x78, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63,
-	0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x6f, 0x6c, 0x6c,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x20, 0x0a, 0x0b, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x72, 0x65, 0x70,
-	0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x10, 0x0a, 0x03, 0x74, 0x74, 0x6c, 0x18,
-	0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x74, 0x74, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x64, 0x69,
-	0x73, 0x6b, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x64,
-	0x69, 0x73, 0x6b, 0x54, 0x79, 0x70, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x66, 0x73, 0x79, 0x6e, 0x63,
-	0x18, 0x06, 0x20, 0x01, 0x28, 0x08, 0x52, 0x05, 0x66, 0x73, 0x79, 0x6e, 0x63, 0x12, 0x2e, 0x0a,
-	0x13, 0x76, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x5f, 0x67, 0x72, 0x6f, 0x77, 0x74, 0x68, 0x5f, 0x63,
-	0x6f, 0x75, 0x6e, 0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x11, 0x76, 0x6f, 0x6c, 0x75,
-	0x6d, 0x65, 0x47, 0x72, 0x6f, 0x77, 0x74, 0x68, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1b, 0x0a,
-	0x09, 0x72, 0x65, 0x61, 0x64, 0x5f, 0x6f, 0x6e, 0x6c, 0x79, 0x18, 0x08, 0x20, 0x01, 0x28, 0x08,
-	0x52, 0x08, 0x72, 0x65, 0x61, 0x64, 0x4f, 0x6e, 0x6c, 0x79, 0x12, 0x1f, 0x0a, 0x0b, 0x64, 0x61,
-	0x74, 0x61, 0x5f, 0x63, 0x65, 0x6e, 0x74, 0x65, 0x72, 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x0a, 0x64, 0x61, 0x74, 0x61, 0x43, 0x65, 0x6e, 0x74, 0x65, 0x72, 0x12, 0x12, 0x0a, 0x04, 0x72,
-	0x61, 0x63, 0x6b, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x72, 0x61, 0x63, 0x6b, 0x12,
-	0x1b, 0x0a, 0x09, 0x64, 0x61, 0x74, 0x61, 0x5f, 0x6e, 0x6f, 0x64, 0x65, 0x18, 0x0b, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x4e, 0x6f, 0x64, 0x65, 0x22, 0x5a, 0x0a, 0x26,
-	0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63,
-	0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74,
-	0x6f, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63,
-	0x74, 0x6f, 0x72, 0x79, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x50, 0x0a, 0x27, 0x43, 0x61, 0x63, 0x68,
-	0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c,
-	0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x45, 0x6e,
-	0x74, 0x72, 0x79, 0x52, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x32, 0x82, 0x0f, 0x0a, 0x0c, 0x53,
-	0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x12, 0x67, 0x0a, 0x14, 0x4c,
-	0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x6e,
-	0x74, 0x72, 0x79, 0x12, 0x25, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c,
-	0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x6e,
-	0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x66, 0x69, 0x6c,
-	0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x44, 0x69, 0x72, 0x65,
-	0x63, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
-	0x73, 0x65, 0x22, 0x00, 0x12, 0x4e, 0x0a, 0x0b, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6e, 0x74, 0x72,
-	0x69, 0x65, 0x73, 0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c,
-	0x69, 0x73, 0x74, 0x45, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x69, 0x73,
-	0x74, 0x45, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x30, 0x01, 0x12, 0x4c, 0x0a, 0x0b, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x45, 0x6e,
-	0x74, 0x72, 0x79, 0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43,
-	0x72, 0x65, 0x61, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x72, 0x65,
-	0x61, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x12, 0x4c, 0x0a, 0x0b, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72,
-	0x79, 0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x55, 0x70, 0x64,
-	0x61, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74,
-	0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00,
-	0x12, 0x52, 0x0a, 0x0d, 0x41, 0x70, 0x70, 0x65, 0x6e, 0x64, 0x54, 0x6f, 0x45, 0x6e, 0x74, 0x72,
-	0x79, 0x12, 0x1e, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x70, 0x70,
-	0x65, 0x6e, 0x64, 0x54, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x1f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x70, 0x70,
-	0x65, 0x6e, 0x64, 0x54, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
-	0x73, 0x65, 0x22, 0x00, 0x12, 0x4c, 0x0a, 0x0b, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x45, 0x6e,
-	0x74, 0x72, 0x79, 0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44,
-	0x65, 0x6c, 0x65, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c,
-	0x65, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x12, 0x5e, 0x0a, 0x11, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61,
-	0x6d, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f,
-	0x70, 0x62, 0x2e, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45,
-	0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69,
-	0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e,
-	0x61, 0x6d, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x12, 0x60, 0x0a, 0x11, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x6e, 0x61,
-	0x6d, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f,
-	0x70, 0x62, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45,
-	0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69,
-	0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x6e,
-	0x61, 0x6d, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x30, 0x01, 0x12, 0x4f, 0x0a, 0x0c, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x56, 0x6f,
-	0x6c, 0x75, 0x6d, 0x65, 0x12, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
-	0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41,
-	0x73, 0x73, 0x69, 0x67, 0x6e, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x4f, 0x0a, 0x0c, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x56,
-	0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x12, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
-	0x2e, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
-	0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x73, 0x70,
-	0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x55, 0x0a, 0x0e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63,
-	0x74, 0x69, 0x6f, 0x6e, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x1f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72,
-	0x5f, 0x70, 0x62, 0x2e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x4c, 0x69,
-	0x73, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x66, 0x69, 0x6c, 0x65,
-	0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x4c,
-	0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x5b, 0x0a,
-	0x10, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f,
-	0x6e, 0x12, 0x21, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c,
-	0x65, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
-	0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x49, 0x0a, 0x0a, 0x53, 0x74,
-	0x61, 0x74, 0x69, 0x73, 0x74, 0x69, 0x63, 0x73, 0x12, 0x1b, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72,
-	0x5f, 0x70, 0x62, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x69, 0x73, 0x74, 0x69, 0x63, 0x73, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
-	0x2e, 0x53, 0x74, 0x61, 0x74, 0x69, 0x73, 0x74, 0x69, 0x63, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x37, 0x0a, 0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x15, 0x2e,
-	0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
-	0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x6a,
-	0x0a, 0x15, 0x47, 0x65, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67,
-	0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x26, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f,
-	0x70, 0x62, 0x2e, 0x47, 0x65, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x27, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x47, 0x65, 0x74, 0x46, 0x69,
-	0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x60, 0x0a, 0x11, 0x53, 0x75,
-	0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x12,
-	0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63,
-	0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53,
-	0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12, 0x65, 0x0a, 0x16,
-	0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x4d, 0x65,
-	0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x12, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70,
-	0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64,
-	0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69, 0x6c,
-	0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4d,
-	0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
-	0x00, 0x30, 0x01, 0x12, 0x56, 0x0a, 0x0d, 0x4b, 0x65, 0x65, 0x70, 0x43, 0x6f, 0x6e, 0x6e, 0x65,
-	0x63, 0x74, 0x65, 0x64, 0x12, 0x1e, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
-	0x4b, 0x65, 0x65, 0x70, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
-	0x4b, 0x65, 0x65, 0x70, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x65, 0x64, 0x52, 0x65, 0x73,
-	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x4f, 0x0a, 0x0c, 0x4c,
-	0x6f, 0x63, 0x61, 0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x1d, 0x2e, 0x66, 0x69,
-	0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x65, 0x42, 0x72, 0x6f,
-	0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, 0x66, 0x69, 0x6c,
-	0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b,
-	0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x05,
-	0x4b, 0x76, 0x47, 0x65, 0x74, 0x12, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
-	0x2e, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e,
-	0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x05, 0x4b, 0x76, 0x50, 0x75,
-	0x74, 0x12, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x50,
-	0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x66, 0x69, 0x6c, 0x65,
-	0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
-	0x73, 0x65, 0x22, 0x00, 0x12, 0x88, 0x01, 0x0a, 0x1f, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65,
+	0x12, 0x13, 0x0a, 0x05, 0x74, 0x73, 0x5f, 0x6e, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03, 0x52,
+	0x04, 0x74, 0x73, 0x4e, 0x73, 0x22, 0x61, 0x0a, 0x08, 0x4c, 0x6f, 0x67, 0x45, 0x6e, 0x74, 0x72,
+	0x79, 0x12, 0x13, 0x0a, 0x05, 0x74, 0x73, 0x5f, 0x6e, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03,
+	0x52, 0x04, 0x74, 0x73, 0x4e, 0x73, 0x12, 0x2c, 0x0a, 0x12, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74,
+	0x69, 0x6f, 0x6e, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x68, 0x61, 0x73, 0x68, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x05, 0x52, 0x10, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x4b, 0x65, 0x79,
+	0x48, 0x61, 0x73, 0x68, 0x12, 0x12, 0x0a, 0x04, 0x64, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01,
+	0x28, 0x0c, 0x52, 0x04, 0x64, 0x61, 0x74, 0x61, 0x22, 0x65, 0x0a, 0x14, 0x4b, 0x65, 0x65, 0x70,
+	0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x67, 0x72, 0x70, 0x63, 0x5f, 0x70, 0x6f, 0x72,
+	0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x67, 0x72, 0x70, 0x63, 0x50, 0x6f, 0x72,
+	0x74, 0x12, 0x1c, 0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x03,
+	0x20, 0x03, 0x28, 0x09, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x22,
+	0x17, 0x0a, 0x15, 0x4b, 0x65, 0x65, 0x70, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x65, 0x64,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x31, 0x0a, 0x13, 0x4c, 0x6f, 0x63, 0x61,
+	0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
+	0x1a, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0xcd, 0x01, 0x0a, 0x14,
+	0x4c, 0x6f, 0x63, 0x61, 0x74, 0x65, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x66, 0x6f, 0x75, 0x6e, 0x64, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x08, 0x52, 0x05, 0x66, 0x6f, 0x75, 0x6e, 0x64, 0x12, 0x45, 0x0a, 0x09, 0x72, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x27, 0x2e,
+	0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x65, 0x42,
+	0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2e, 0x52, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x73, 0x1a, 0x58, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x25, 0x0a,
+	0x0e, 0x67, 0x72, 0x70, 0x63, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x67, 0x72, 0x70, 0x63, 0x41, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x65, 0x73, 0x12, 0x25, 0x0a, 0x0e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0d, 0x72, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x20, 0x0a, 0x0c, 0x4b,
+	0x76, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x6b,
+	0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x3b, 0x0a,
+	0x0d, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x14,
+	0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x05, 0x76,
+	0x61, 0x6c, 0x75, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0x36, 0x0a, 0x0c, 0x4b, 0x76,
+	0x50, 0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65,
+	0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05,
+	0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x05, 0x76, 0x61, 0x6c,
+	0x75, 0x65, 0x22, 0x25, 0x0a, 0x0d, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x18, 0x01, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x05, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x22, 0xbd, 0x03, 0x0a, 0x09, 0x46, 0x69,
+	0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69,
+	0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f,
+	0x6e, 0x12, 0x3a, 0x0a, 0x09, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x02,
+	0x20, 0x03, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
+	0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x2e, 0x50, 0x61, 0x74, 0x68, 0x43, 0x6f,
+	0x6e, 0x66, 0x52, 0x09, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0xd9, 0x02,
+	0x0a, 0x08, 0x50, 0x61, 0x74, 0x68, 0x43, 0x6f, 0x6e, 0x66, 0x12, 0x27, 0x0a, 0x0f, 0x6c, 0x6f,
+	0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x70, 0x72, 0x65, 0x66, 0x69, 0x78, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x0e, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x50, 0x72, 0x65,
+	0x66, 0x69, 0x78, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x12, 0x20, 0x0a, 0x0b, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69,
+	0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x10, 0x0a, 0x03, 0x74, 0x74, 0x6c, 0x18, 0x04, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x03, 0x74, 0x74, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x64, 0x69, 0x73, 0x6b, 0x5f,
+	0x74, 0x79, 0x70, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x64, 0x69, 0x73, 0x6b,
+	0x54, 0x79, 0x70, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x66, 0x73, 0x79, 0x6e, 0x63, 0x18, 0x06, 0x20,
+	0x01, 0x28, 0x08, 0x52, 0x05, 0x66, 0x73, 0x79, 0x6e, 0x63, 0x12, 0x2e, 0x0a, 0x13, 0x76, 0x6f,
+	0x6c, 0x75, 0x6d, 0x65, 0x5f, 0x67, 0x72, 0x6f, 0x77, 0x74, 0x68, 0x5f, 0x63, 0x6f, 0x75, 0x6e,
+	0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x11, 0x76, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x47,
+	0x72, 0x6f, 0x77, 0x74, 0x68, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x72, 0x65,
+	0x61, 0x64, 0x5f, 0x6f, 0x6e, 0x6c, 0x79, 0x18, 0x08, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x72,
+	0x65, 0x61, 0x64, 0x4f, 0x6e, 0x6c, 0x79, 0x12, 0x1f, 0x0a, 0x0b, 0x64, 0x61, 0x74, 0x61, 0x5f,
+	0x63, 0x65, 0x6e, 0x74, 0x65, 0x72, 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x64, 0x61,
+	0x74, 0x61, 0x43, 0x65, 0x6e, 0x74, 0x65, 0x72, 0x12, 0x12, 0x0a, 0x04, 0x72, 0x61, 0x63, 0x6b,
+	0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x72, 0x61, 0x63, 0x6b, 0x12, 0x1b, 0x0a, 0x09,
+	0x64, 0x61, 0x74, 0x61, 0x5f, 0x6e, 0x6f, 0x64, 0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x08, 0x64, 0x61, 0x74, 0x61, 0x4e, 0x6f, 0x64, 0x65, 0x22, 0x5a, 0x0a, 0x26, 0x43, 0x61, 0x63,
+	0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f,
+	0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72,
+	0x79, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x04, 0x6e, 0x61, 0x6d, 0x65, 0x22, 0x50, 0x0a, 0x27, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65,
 	0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61,
-	0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x30, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72,
-	0x5f, 0x70, 0x62, 0x2e, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f,
-	0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73,
-	0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x31, 0x2e, 0x66, 0x69, 0x6c,
-	0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74,
-	0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c,
-	0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42,
-	0x4f, 0x0a, 0x10, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69,
-	0x65, 0x6e, 0x74, 0x42, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a,
-	0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69,
-	0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f,
-	0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
-	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x12, 0x25, 0x0a, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32,
+	0x0f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x45, 0x6e, 0x74, 0x72, 0x79,
+	0x52, 0x05, 0x65, 0x6e, 0x74, 0x72, 0x79, 0x32, 0xd9, 0x0d, 0x0a, 0x0c, 0x53, 0x65, 0x61, 0x77,
+	0x65, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x12, 0x67, 0x0a, 0x14, 0x4c, 0x6f, 0x6f, 0x6b,
+	0x75, 0x70, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x6e, 0x74, 0x72, 0x79,
+	0x12, 0x25, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x6f, 0x6b,
+	0x75, 0x70, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x79, 0x45, 0x6e, 0x74, 0x72, 0x79,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x26, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f,
+	0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x6f,
+	0x72, 0x79, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
+	0x00, 0x12, 0x4e, 0x0a, 0x0b, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73,
+	0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x69, 0x73, 0x74,
+	0x45, 0x6e, 0x74, 0x72, 0x69, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x45, 0x6e,
+	0x74, 0x72, 0x69, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30,
+	0x01, 0x12, 0x4c, 0x0a, 0x0b, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79,
+	0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x72, 0x65, 0x61,
+	0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65,
+	0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12,
+	0x4c, 0x0a, 0x0b, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x1c,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65,
+	0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x66,
+	0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x45, 0x6e,
+	0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x52, 0x0a,
+	0x0d, 0x41, 0x70, 0x70, 0x65, 0x6e, 0x64, 0x54, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x1e,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x70, 0x70, 0x65, 0x6e, 0x64,
+	0x54, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x70, 0x70, 0x65, 0x6e, 0x64,
+	0x54, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
+	0x00, 0x12, 0x4c, 0x0a, 0x0b, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79,
+	0x12, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c, 0x65,
+	0x74, 0x65, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65,
+	0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12,
+	0x5e, 0x0a, 0x11, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45,
+	0x6e, 0x74, 0x72, 0x79, 0x12, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
+	0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45, 0x6e, 0x74, 0x72,
+	0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72,
+	0x5f, 0x70, 0x62, 0x2e, 0x41, 0x74, 0x6f, 0x6d, 0x69, 0x63, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65,
+	0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12,
+	0x60, 0x0a, 0x11, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45,
+	0x6e, 0x74, 0x72, 0x79, 0x12, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
+	0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65, 0x45, 0x6e, 0x74, 0x72,
+	0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72,
+	0x5f, 0x70, 0x62, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x6e, 0x61, 0x6d, 0x65,
+	0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30,
+	0x01, 0x12, 0x4f, 0x0a, 0x0c, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x56, 0x6f, 0x6c, 0x75, 0x6d,
+	0x65, 0x12, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x73, 0x73,
+	0x69, 0x67, 0x6e, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x1e, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x41, 0x73, 0x73, 0x69,
+	0x67, 0x6e, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x12, 0x4f, 0x0a, 0x0c, 0x4c, 0x6f, 0x6f, 0x6b, 0x75, 0x70, 0x56, 0x6f, 0x6c, 0x75,
+	0x6d, 0x65, 0x12, 0x1d, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f,
+	0x6f, 0x6b, 0x75, 0x70, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x1e, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4c, 0x6f, 0x6f,
+	0x6b, 0x75, 0x70, 0x56, 0x6f, 0x6c, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x22, 0x00, 0x12, 0x55, 0x0a, 0x0e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x4c, 0x69, 0x73, 0x74, 0x12, 0x1f, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
+	0x2e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x4c, 0x69, 0x73, 0x74, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70,
+	0x62, 0x2e, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x4c, 0x69, 0x73, 0x74,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x5b, 0x0a, 0x10, 0x44, 0x65,
+	0x6c, 0x65, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x21,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65,
+	0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c,
+	0x65, 0x74, 0x65, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x49, 0x0a, 0x0a, 0x53, 0x74, 0x61, 0x74, 0x69,
+	0x73, 0x74, 0x69, 0x63, 0x73, 0x12, 0x1b, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
+	0x2e, 0x53, 0x74, 0x61, 0x74, 0x69, 0x73, 0x74, 0x69, 0x63, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x74,
+	0x61, 0x74, 0x69, 0x73, 0x74, 0x69, 0x63, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x12, 0x37, 0x0a, 0x04, 0x50, 0x69, 0x6e, 0x67, 0x12, 0x15, 0x2e, 0x66, 0x69, 0x6c,
+	0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x69, 0x6e,
+	0x67, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x6a, 0x0a, 0x15, 0x47,
+	0x65, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x12, 0x26, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e,
+	0x47, 0x65, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72,
+	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x66,
+	0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x47, 0x65, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x72,
+	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x60, 0x0a, 0x11, 0x53, 0x75, 0x62, 0x73, 0x63,
+	0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x12, 0x22, 0x2e, 0x66,
+	0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62,
+	0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x23, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73,
+	0x63, 0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12, 0x65, 0x0a, 0x16, 0x53, 0x75, 0x62,
+	0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x4d, 0x65, 0x74, 0x61, 0x64,
+	0x61, 0x74, 0x61, 0x12, 0x22, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x53,
+	0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f,
+	0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x4d, 0x65, 0x74, 0x61,
+	0x64, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30, 0x01,
+	0x12, 0x3a, 0x0a, 0x05, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x12, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65,
+	0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x17, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x47,
+	0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x05,
+	0x4b, 0x76, 0x50, 0x75, 0x74, 0x12, 0x16, 0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62,
+	0x2e, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e,
+	0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x4b, 0x76, 0x50, 0x75, 0x74, 0x52, 0x65,
+	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x88, 0x01, 0x0a, 0x1f, 0x43, 0x61, 0x63,
+	0x68, 0x65, 0x52, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f,
+	0x4c, 0x6f, 0x63, 0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x30, 0x2e, 0x66,
+	0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52, 0x65, 0x6d,
+	0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63, 0x61, 0x6c,
+	0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x31,
+	0x2e, 0x66, 0x69, 0x6c, 0x65, 0x72, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x61, 0x63, 0x68, 0x65, 0x52,
+	0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x54, 0x6f, 0x4c, 0x6f, 0x63,
+	0x61, 0x6c, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x22, 0x00, 0x42, 0x4f, 0x0a, 0x10, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73,
+	0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x0a, 0x46, 0x69, 0x6c, 0x65, 0x72, 0x50, 0x72,
+	0x6f, 0x74, 0x6f, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
+	0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65,
+	0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x66, 0x69, 0x6c, 0x65,
+	0x72, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -4509,35 +4497,31 @@ var file_filer_proto_depIdxs = []int32{
 	39, // 39: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest
 	41, // 40: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest
 	41, // 41: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest
-	44, // 42: filer_pb.SeaweedFiler.KeepConnected:input_type -> filer_pb.KeepConnectedRequest
-	46, // 43: filer_pb.SeaweedFiler.LocateBroker:input_type -> filer_pb.LocateBrokerRequest
-	48, // 44: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
-	50, // 45: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
-	53, // 46: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
-	1,  // 47: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
-	3,  // 48: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
-	13, // 49: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
-	15, // 50: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse
-	17, // 51: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse
-	19, // 52: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse
-	21, // 53: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse
-	23, // 54: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse
-	25, // 55: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse
-	29, // 56: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse
-	32, // 57: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse
-	34, // 58: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse
-	36, // 59: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse
-	38, // 60: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse
-	40, // 61: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse
-	42, // 62: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse
-	42, // 63: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse
-	45, // 64: filer_pb.SeaweedFiler.KeepConnected:output_type -> filer_pb.KeepConnectedResponse
-	47, // 65: filer_pb.SeaweedFiler.LocateBroker:output_type -> filer_pb.LocateBrokerResponse
-	49, // 66: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
-	51, // 67: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
-	54, // 68: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
-	47, // [47:69] is the sub-list for method output_type
-	25, // [25:47] is the sub-list for method input_type
+	48, // 42: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
+	50, // 43: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
+	53, // 44: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
+	1,  // 45: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
+	3,  // 46: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
+	13, // 47: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
+	15, // 48: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse
+	17, // 49: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse
+	19, // 50: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse
+	21, // 51: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse
+	23, // 52: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse
+	25, // 53: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse
+	29, // 54: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse
+	32, // 55: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse
+	34, // 56: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse
+	36, // 57: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse
+	38, // 58: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse
+	40, // 59: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse
+	42, // 60: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse
+	42, // 61: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse
+	49, // 62: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
+	51, // 63: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
+	54, // 64: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
+	45, // [45:65] is the sub-list for method output_type
+	25, // [25:45] is the sub-list for method input_type
 	25, // [25:25] is the sub-list for extension type_name
 	25, // [25:25] is the sub-list for extension extendee
 	0,  // [0:25] is the sub-list for field type_name
diff --git a/weed/pb/filer_pb/filer_grpc.pb.go b/weed/pb/filer_pb/filer_grpc.pb.go
index 4a5f47d71..270e13e6f 100644
--- a/weed/pb/filer_pb/filer_grpc.pb.go
+++ b/weed/pb/filer_pb/filer_grpc.pb.go
@@ -35,8 +35,6 @@ type SeaweedFilerClient interface {
 	GetFilerConfiguration(ctx context.Context, in *GetFilerConfigurationRequest, opts ...grpc.CallOption) (*GetFilerConfigurationResponse, error)
 	SubscribeMetadata(ctx context.Context, in *SubscribeMetadataRequest, opts ...grpc.CallOption) (SeaweedFiler_SubscribeMetadataClient, error)
 	SubscribeLocalMetadata(ctx context.Context, in *SubscribeMetadataRequest, opts ...grpc.CallOption) (SeaweedFiler_SubscribeLocalMetadataClient, error)
-	KeepConnected(ctx context.Context, opts ...grpc.CallOption) (SeaweedFiler_KeepConnectedClient, error)
-	LocateBroker(ctx context.Context, in *LocateBrokerRequest, opts ...grpc.CallOption) (*LocateBrokerResponse, error)
 	KvGet(ctx context.Context, in *KvGetRequest, opts ...grpc.CallOption) (*KvGetResponse, error)
 	KvPut(ctx context.Context, in *KvPutRequest, opts ...grpc.CallOption) (*KvPutResponse, error)
 	CacheRemoteObjectToLocalCluster(ctx context.Context, in *CacheRemoteObjectToLocalClusterRequest, opts ...grpc.CallOption) (*CacheRemoteObjectToLocalClusterResponse, error)
@@ -295,46 +293,6 @@ func (x *seaweedFilerSubscribeLocalMetadataClient) Recv() (*SubscribeMetadataRes
 	return m, nil
 }
 
-func (c *seaweedFilerClient) KeepConnected(ctx context.Context, opts ...grpc.CallOption) (SeaweedFiler_KeepConnectedClient, error) {
-	stream, err := c.cc.NewStream(ctx, &SeaweedFiler_ServiceDesc.Streams[4], "/filer_pb.SeaweedFiler/KeepConnected", opts...)
-	if err != nil {
-		return nil, err
-	}
-	x := &seaweedFilerKeepConnectedClient{stream}
-	return x, nil
-}
-
-type SeaweedFiler_KeepConnectedClient interface {
-	Send(*KeepConnectedRequest) error
-	Recv() (*KeepConnectedResponse, error)
-	grpc.ClientStream
-}
-
-type seaweedFilerKeepConnectedClient struct {
-	grpc.ClientStream
-}
-
-func (x *seaweedFilerKeepConnectedClient) Send(m *KeepConnectedRequest) error {
-	return x.ClientStream.SendMsg(m)
-}
-
-func (x *seaweedFilerKeepConnectedClient) Recv() (*KeepConnectedResponse, error) {
-	m := new(KeepConnectedResponse)
-	if err := x.ClientStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
-
-func (c *seaweedFilerClient) LocateBroker(ctx context.Context, in *LocateBrokerRequest, opts ...grpc.CallOption) (*LocateBrokerResponse, error) {
-	out := new(LocateBrokerResponse)
-	err := c.cc.Invoke(ctx, "/filer_pb.SeaweedFiler/LocateBroker", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
 func (c *seaweedFilerClient) KvGet(ctx context.Context, in *KvGetRequest, opts ...grpc.CallOption) (*KvGetResponse, error) {
 	out := new(KvGetResponse)
 	err := c.cc.Invoke(ctx, "/filer_pb.SeaweedFiler/KvGet", in, out, opts...)
@@ -383,8 +341,6 @@ type SeaweedFilerServer interface {
 	GetFilerConfiguration(context.Context, *GetFilerConfigurationRequest) (*GetFilerConfigurationResponse, error)
 	SubscribeMetadata(*SubscribeMetadataRequest, SeaweedFiler_SubscribeMetadataServer) error
 	SubscribeLocalMetadata(*SubscribeMetadataRequest, SeaweedFiler_SubscribeLocalMetadataServer) error
-	KeepConnected(SeaweedFiler_KeepConnectedServer) error
-	LocateBroker(context.Context, *LocateBrokerRequest) (*LocateBrokerResponse, error)
 	KvGet(context.Context, *KvGetRequest) (*KvGetResponse, error)
 	KvPut(context.Context, *KvPutRequest) (*KvPutResponse, error)
 	CacheRemoteObjectToLocalCluster(context.Context, *CacheRemoteObjectToLocalClusterRequest) (*CacheRemoteObjectToLocalClusterResponse, error)
@@ -446,12 +402,6 @@ func (UnimplementedSeaweedFilerServer) SubscribeMetadata(*SubscribeMetadataReque
 func (UnimplementedSeaweedFilerServer) SubscribeLocalMetadata(*SubscribeMetadataRequest, SeaweedFiler_SubscribeLocalMetadataServer) error {
 	return status.Errorf(codes.Unimplemented, "method SubscribeLocalMetadata not implemented")
 }
-func (UnimplementedSeaweedFilerServer) KeepConnected(SeaweedFiler_KeepConnectedServer) error {
-	return status.Errorf(codes.Unimplemented, "method KeepConnected not implemented")
-}
-func (UnimplementedSeaweedFilerServer) LocateBroker(context.Context, *LocateBrokerRequest) (*LocateBrokerResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method LocateBroker not implemented")
-}
 func (UnimplementedSeaweedFilerServer) KvGet(context.Context, *KvGetRequest) (*KvGetResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method KvGet not implemented")
 }
@@ -792,50 +742,6 @@ func (x *seaweedFilerSubscribeLocalMetadataServer) Send(m *SubscribeMetadataResp
 	return x.ServerStream.SendMsg(m)
 }
 
-func _SeaweedFiler_KeepConnected_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(SeaweedFilerServer).KeepConnected(&seaweedFilerKeepConnectedServer{stream})
-}
-
-type SeaweedFiler_KeepConnectedServer interface {
-	Send(*KeepConnectedResponse) error
-	Recv() (*KeepConnectedRequest, error)
-	grpc.ServerStream
-}
-
-type seaweedFilerKeepConnectedServer struct {
-	grpc.ServerStream
-}
-
-func (x *seaweedFilerKeepConnectedServer) Send(m *KeepConnectedResponse) error {
-	return x.ServerStream.SendMsg(m)
-}
-
-func (x *seaweedFilerKeepConnectedServer) Recv() (*KeepConnectedRequest, error) {
-	m := new(KeepConnectedRequest)
-	if err := x.ServerStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
-
-func _SeaweedFiler_LocateBroker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(LocateBrokerRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(SeaweedFilerServer).LocateBroker(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/filer_pb.SeaweedFiler/LocateBroker",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(SeaweedFilerServer).LocateBroker(ctx, req.(*LocateBrokerRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
 func _SeaweedFiler_KvGet_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(KvGetRequest)
 	if err := dec(in); err != nil {
@@ -950,10 +856,6 @@ var SeaweedFiler_ServiceDesc = grpc.ServiceDesc{
 			Handler:    _SeaweedFiler_GetFilerConfiguration_Handler,
 		},
 		{
-			MethodName: "LocateBroker",
-			Handler:    _SeaweedFiler_LocateBroker_Handler,
-		},
-		{
 			MethodName: "KvGet",
 			Handler:    _SeaweedFiler_KvGet_Handler,
 		},
@@ -987,12 +889,6 @@ var SeaweedFiler_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _SeaweedFiler_SubscribeLocalMetadata_Handler,
 			ServerStreams: true,
 		},
-		{
-			StreamName:    "KeepConnected",
-			Handler:       _SeaweedFiler_KeepConnected_Handler,
-			ServerStreams: true,
-			ClientStreams: true,
-		},
 	},
 	Metadata: "filer.proto",
 }
diff --git a/weed/pb/mq.proto b/weed/pb/mq.proto
index abe538485..44291c173 100644
--- a/weed/pb/mq.proto
+++ b/weed/pb/mq.proto
@@ -10,21 +10,6 @@ option java_outer_classname = "MessagQueueProto";
 
 service SeaweedMessaging {
 
-    rpc Subscribe (stream SubscriberMessage) returns (stream BrokerMessage) {
-    }
-
-    rpc Publish (stream PublishRequest) returns (stream PublishResponse) {
-    }
-
-    rpc DeleteTopic (DeleteTopicRequest) returns (DeleteTopicResponse) {
-    }
-
-    rpc ConfigureTopic (ConfigureTopicRequest) returns (ConfigureTopicResponse) {
-    }
-
-    rpc GetTopicConfiguration (GetTopicConfigurationRequest) returns (GetTopicConfigurationResponse) {
-    }
-
     rpc FindBroker (FindBrokerRequest) returns (FindBrokerResponse) {
     }
 
@@ -32,85 +17,6 @@ service SeaweedMessaging {
 
 //////////////////////////////////////////////////
 
-message SubscriberMessage {
-    message InitMessage {
-        string namespace = 1;
-        string topic = 2;
-        int32 partition = 3;
-        enum StartPosition {
-            LATEST = 0; // Start at the newest message
-            EARLIEST = 1; // Start at the oldest message
-            TIMESTAMP = 2; // Start after a specified timestamp, exclusive
-        }
-        StartPosition startPosition = 4; // Where to begin consuming from
-        int64 timestampNs = 5; // timestamp in nano seconds
-        string subscriber_id = 6; // uniquely identify a subscriber to track consumption
-    }
-    InitMessage init = 1;
-    message AckMessage {
-        int64 message_id = 1;
-    }
-    AckMessage ack = 2;
-    bool is_close = 3;
-}
-
-message Message {
-    int64 event_time_ns = 1 [jstype = JS_STRING];
-    bytes key = 2; // Message key
-    bytes value = 3; // Message payload
-    map<string, bytes> headers = 4; // Message headers
-    bool is_close = 5;
-}
-
-message BrokerMessage {
-    Message data = 1;
-}
-
-message PublishRequest {
-    message InitMessage {
-        string namespace = 1; // only needed on the initial request
-        string topic = 2; // only needed on the initial request
-        int32 partition = 3;
-    }
-    InitMessage init = 1;
-    Message data = 2;
-}
-
-message PublishResponse {
-    message ConfigMessage {
-        int32 partition_count = 1;
-    }
-    ConfigMessage config = 1;
-    message RedirectMessage {
-        string new_broker = 1;
-    }
-    RedirectMessage redirect = 2;
-    bool is_closed = 3;
-}
-
-message DeleteTopicRequest {
-    string namespace = 1;
-    string topic = 2;
-}
-message DeleteTopicResponse {
-}
-
-message ConfigureTopicRequest {
-    string namespace = 1;
-    string topic = 2;
-    TopicConfiguration configuration = 3;
-}
-message ConfigureTopicResponse {
-}
-
-message GetTopicConfigurationRequest {
-    string namespace = 1;
-    string topic = 2;
-}
-message GetTopicConfigurationResponse {
-    TopicConfiguration configuration = 1;
-}
-
 message FindBrokerRequest {
     string namespace = 1;
     string topic = 2;
@@ -120,16 +26,3 @@ message FindBrokerRequest {
 message FindBrokerResponse {
     string broker = 1;
 }
-
-message TopicConfiguration {
-    int32 partition_count = 1;
-    string collection = 2;
-    string replication = 3;
-    bool is_transient = 4;
-    enum Partitioning {
-        NonNullKeyHash = 0; // If not null, hash by key value. If null, round robin
-        KeyHash = 1; // hash by key value
-        RoundRobin = 2; // round robin pick one partition
-    }
-    Partitioning partitoning = 5;
-}
diff --git a/weed/pb/mq_pb/mq.pb.go b/weed/pb/mq_pb/mq.pb.go
index a47ce128d..dac27c959 100644
--- a/weed/pb/mq_pb/mq.pb.go
+++ b/weed/pb/mq_pb/mq.pb.go
@@ -20,707 +20,6 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-type SubscriberMessage_InitMessage_StartPosition int32
-
-const (
-	SubscriberMessage_InitMessage_LATEST    SubscriberMessage_InitMessage_StartPosition = 0 // Start at the newest message
-	SubscriberMessage_InitMessage_EARLIEST  SubscriberMessage_InitMessage_StartPosition = 1 // Start at the oldest message
-	SubscriberMessage_InitMessage_TIMESTAMP SubscriberMessage_InitMessage_StartPosition = 2 // Start after a specified timestamp, exclusive
-)
-
-// Enum value maps for SubscriberMessage_InitMessage_StartPosition.
-var (
-	SubscriberMessage_InitMessage_StartPosition_name = map[int32]string{
-		0: "LATEST",
-		1: "EARLIEST",
-		2: "TIMESTAMP",
-	}
-	SubscriberMessage_InitMessage_StartPosition_value = map[string]int32{
-		"LATEST":    0,
-		"EARLIEST":  1,
-		"TIMESTAMP": 2,
-	}
-)
-
-func (x SubscriberMessage_InitMessage_StartPosition) Enum() *SubscriberMessage_InitMessage_StartPosition {
-	p := new(SubscriberMessage_InitMessage_StartPosition)
-	*p = x
-	return p
-}
-
-func (x SubscriberMessage_InitMessage_StartPosition) String() string {
-	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
-}
-
-func (SubscriberMessage_InitMessage_StartPosition) Descriptor() protoreflect.EnumDescriptor {
-	return file_mq_proto_enumTypes[0].Descriptor()
-}
-
-func (SubscriberMessage_InitMessage_StartPosition) Type() protoreflect.EnumType {
-	return &file_mq_proto_enumTypes[0]
-}
-
-func (x SubscriberMessage_InitMessage_StartPosition) Number() protoreflect.EnumNumber {
-	return protoreflect.EnumNumber(x)
-}
-
-// Deprecated: Use SubscriberMessage_InitMessage_StartPosition.Descriptor instead.
-func (SubscriberMessage_InitMessage_StartPosition) EnumDescriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{0, 0, 0}
-}
-
-type TopicConfiguration_Partitioning int32
-
-const (
-	TopicConfiguration_NonNullKeyHash TopicConfiguration_Partitioning = 0 // If not null, hash by key value. If null, round robin
-	TopicConfiguration_KeyHash        TopicConfiguration_Partitioning = 1 // hash by key value
-	TopicConfiguration_RoundRobin     TopicConfiguration_Partitioning = 2 // round robin pick one partition
-)
-
-// Enum value maps for TopicConfiguration_Partitioning.
-var (
-	TopicConfiguration_Partitioning_name = map[int32]string{
-		0: "NonNullKeyHash",
-		1: "KeyHash",
-		2: "RoundRobin",
-	}
-	TopicConfiguration_Partitioning_value = map[string]int32{
-		"NonNullKeyHash": 0,
-		"KeyHash":        1,
-		"RoundRobin":     2,
-	}
-)
-
-func (x TopicConfiguration_Partitioning) Enum() *TopicConfiguration_Partitioning {
-	p := new(TopicConfiguration_Partitioning)
-	*p = x
-	return p
-}
-
-func (x TopicConfiguration_Partitioning) String() string {
-	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
-}
-
-func (TopicConfiguration_Partitioning) Descriptor() protoreflect.EnumDescriptor {
-	return file_mq_proto_enumTypes[1].Descriptor()
-}
-
-func (TopicConfiguration_Partitioning) Type() protoreflect.EnumType {
-	return &file_mq_proto_enumTypes[1]
-}
-
-func (x TopicConfiguration_Partitioning) Number() protoreflect.EnumNumber {
-	return protoreflect.EnumNumber(x)
-}
-
-// Deprecated: Use TopicConfiguration_Partitioning.Descriptor instead.
-func (TopicConfiguration_Partitioning) EnumDescriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{13, 0}
-}
-
-type SubscriberMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Init    *SubscriberMessage_InitMessage `protobuf:"bytes,1,opt,name=init,proto3" json:"init,omitempty"`
-	Ack     *SubscriberMessage_AckMessage  `protobuf:"bytes,2,opt,name=ack,proto3" json:"ack,omitempty"`
-	IsClose bool                           `protobuf:"varint,3,opt,name=is_close,json=isClose,proto3" json:"is_close,omitempty"`
-}
-
-func (x *SubscriberMessage) Reset() {
-	*x = SubscriberMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[0]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *SubscriberMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*SubscriberMessage) ProtoMessage() {}
-
-func (x *SubscriberMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[0]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use SubscriberMessage.ProtoReflect.Descriptor instead.
-func (*SubscriberMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{0}
-}
-
-func (x *SubscriberMessage) GetInit() *SubscriberMessage_InitMessage {
-	if x != nil {
-		return x.Init
-	}
-	return nil
-}
-
-func (x *SubscriberMessage) GetAck() *SubscriberMessage_AckMessage {
-	if x != nil {
-		return x.Ack
-	}
-	return nil
-}
-
-func (x *SubscriberMessage) GetIsClose() bool {
-	if x != nil {
-		return x.IsClose
-	}
-	return false
-}
-
-type Message struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	EventTimeNs int64             `protobuf:"varint,1,opt,name=event_time_ns,json=eventTimeNs,proto3" json:"event_time_ns,omitempty"`
-	Key         []byte            `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`                                                                                                 // Message key
-	Value       []byte            `protobuf:"bytes,3,opt,name=value,proto3" json:"value,omitempty"`                                                                                             // Message payload
-	Headers     map[string][]byte `protobuf:"bytes,4,rep,name=headers,proto3" json:"headers,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` // Message headers
-	IsClose     bool              `protobuf:"varint,5,opt,name=is_close,json=isClose,proto3" json:"is_close,omitempty"`
-}
-
-func (x *Message) Reset() {
-	*x = Message{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[1]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *Message) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*Message) ProtoMessage() {}
-
-func (x *Message) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[1]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use Message.ProtoReflect.Descriptor instead.
-func (*Message) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{1}
-}
-
-func (x *Message) GetEventTimeNs() int64 {
-	if x != nil {
-		return x.EventTimeNs
-	}
-	return 0
-}
-
-func (x *Message) GetKey() []byte {
-	if x != nil {
-		return x.Key
-	}
-	return nil
-}
-
-func (x *Message) GetValue() []byte {
-	if x != nil {
-		return x.Value
-	}
-	return nil
-}
-
-func (x *Message) GetHeaders() map[string][]byte {
-	if x != nil {
-		return x.Headers
-	}
-	return nil
-}
-
-func (x *Message) GetIsClose() bool {
-	if x != nil {
-		return x.IsClose
-	}
-	return false
-}
-
-type BrokerMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Data *Message `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
-}
-
-func (x *BrokerMessage) Reset() {
-	*x = BrokerMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[2]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *BrokerMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*BrokerMessage) ProtoMessage() {}
-
-func (x *BrokerMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[2]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use BrokerMessage.ProtoReflect.Descriptor instead.
-func (*BrokerMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{2}
-}
-
-func (x *BrokerMessage) GetData() *Message {
-	if x != nil {
-		return x.Data
-	}
-	return nil
-}
-
-type PublishRequest struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Init *PublishRequest_InitMessage `protobuf:"bytes,1,opt,name=init,proto3" json:"init,omitempty"`
-	Data *Message                    `protobuf:"bytes,2,opt,name=data,proto3" json:"data,omitempty"`
-}
-
-func (x *PublishRequest) Reset() {
-	*x = PublishRequest{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[3]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *PublishRequest) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*PublishRequest) ProtoMessage() {}
-
-func (x *PublishRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[3]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use PublishRequest.ProtoReflect.Descriptor instead.
-func (*PublishRequest) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{3}
-}
-
-func (x *PublishRequest) GetInit() *PublishRequest_InitMessage {
-	if x != nil {
-		return x.Init
-	}
-	return nil
-}
-
-func (x *PublishRequest) GetData() *Message {
-	if x != nil {
-		return x.Data
-	}
-	return nil
-}
-
-type PublishResponse struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Config   *PublishResponse_ConfigMessage   `protobuf:"bytes,1,opt,name=config,proto3" json:"config,omitempty"`
-	Redirect *PublishResponse_RedirectMessage `protobuf:"bytes,2,opt,name=redirect,proto3" json:"redirect,omitempty"`
-	IsClosed bool                             `protobuf:"varint,3,opt,name=is_closed,json=isClosed,proto3" json:"is_closed,omitempty"`
-}
-
-func (x *PublishResponse) Reset() {
-	*x = PublishResponse{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[4]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *PublishResponse) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*PublishResponse) ProtoMessage() {}
-
-func (x *PublishResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[4]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use PublishResponse.ProtoReflect.Descriptor instead.
-func (*PublishResponse) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{4}
-}
-
-func (x *PublishResponse) GetConfig() *PublishResponse_ConfigMessage {
-	if x != nil {
-		return x.Config
-	}
-	return nil
-}
-
-func (x *PublishResponse) GetRedirect() *PublishResponse_RedirectMessage {
-	if x != nil {
-		return x.Redirect
-	}
-	return nil
-}
-
-func (x *PublishResponse) GetIsClosed() bool {
-	if x != nil {
-		return x.IsClosed
-	}
-	return false
-}
-
-type DeleteTopicRequest struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"`
-	Topic     string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"`
-}
-
-func (x *DeleteTopicRequest) Reset() {
-	*x = DeleteTopicRequest{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[5]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *DeleteTopicRequest) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*DeleteTopicRequest) ProtoMessage() {}
-
-func (x *DeleteTopicRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[5]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use DeleteTopicRequest.ProtoReflect.Descriptor instead.
-func (*DeleteTopicRequest) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{5}
-}
-
-func (x *DeleteTopicRequest) GetNamespace() string {
-	if x != nil {
-		return x.Namespace
-	}
-	return ""
-}
-
-func (x *DeleteTopicRequest) GetTopic() string {
-	if x != nil {
-		return x.Topic
-	}
-	return ""
-}
-
-type DeleteTopicResponse struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-}
-
-func (x *DeleteTopicResponse) Reset() {
-	*x = DeleteTopicResponse{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[6]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *DeleteTopicResponse) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*DeleteTopicResponse) ProtoMessage() {}
-
-func (x *DeleteTopicResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[6]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use DeleteTopicResponse.ProtoReflect.Descriptor instead.
-func (*DeleteTopicResponse) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{6}
-}
-
-type ConfigureTopicRequest struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Namespace     string              `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"`
-	Topic         string              `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"`
-	Configuration *TopicConfiguration `protobuf:"bytes,3,opt,name=configuration,proto3" json:"configuration,omitempty"`
-}
-
-func (x *ConfigureTopicRequest) Reset() {
-	*x = ConfigureTopicRequest{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[7]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *ConfigureTopicRequest) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*ConfigureTopicRequest) ProtoMessage() {}
-
-func (x *ConfigureTopicRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[7]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use ConfigureTopicRequest.ProtoReflect.Descriptor instead.
-func (*ConfigureTopicRequest) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{7}
-}
-
-func (x *ConfigureTopicRequest) GetNamespace() string {
-	if x != nil {
-		return x.Namespace
-	}
-	return ""
-}
-
-func (x *ConfigureTopicRequest) GetTopic() string {
-	if x != nil {
-		return x.Topic
-	}
-	return ""
-}
-
-func (x *ConfigureTopicRequest) GetConfiguration() *TopicConfiguration {
-	if x != nil {
-		return x.Configuration
-	}
-	return nil
-}
-
-type ConfigureTopicResponse struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-}
-
-func (x *ConfigureTopicResponse) Reset() {
-	*x = ConfigureTopicResponse{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[8]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *ConfigureTopicResponse) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*ConfigureTopicResponse) ProtoMessage() {}
-
-func (x *ConfigureTopicResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[8]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use ConfigureTopicResponse.ProtoReflect.Descriptor instead.
-func (*ConfigureTopicResponse) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{8}
-}
-
-type GetTopicConfigurationRequest struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"`
-	Topic     string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"`
-}
-
-func (x *GetTopicConfigurationRequest) Reset() {
-	*x = GetTopicConfigurationRequest{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[9]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *GetTopicConfigurationRequest) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*GetTopicConfigurationRequest) ProtoMessage() {}
-
-func (x *GetTopicConfigurationRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[9]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use GetTopicConfigurationRequest.ProtoReflect.Descriptor instead.
-func (*GetTopicConfigurationRequest) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{9}
-}
-
-func (x *GetTopicConfigurationRequest) GetNamespace() string {
-	if x != nil {
-		return x.Namespace
-	}
-	return ""
-}
-
-func (x *GetTopicConfigurationRequest) GetTopic() string {
-	if x != nil {
-		return x.Topic
-	}
-	return ""
-}
-
-type GetTopicConfigurationResponse struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Configuration *TopicConfiguration `protobuf:"bytes,1,opt,name=configuration,proto3" json:"configuration,omitempty"`
-}
-
-func (x *GetTopicConfigurationResponse) Reset() {
-	*x = GetTopicConfigurationResponse{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[10]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *GetTopicConfigurationResponse) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*GetTopicConfigurationResponse) ProtoMessage() {}
-
-func (x *GetTopicConfigurationResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[10]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use GetTopicConfigurationResponse.ProtoReflect.Descriptor instead.
-func (*GetTopicConfigurationResponse) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{10}
-}
-
-func (x *GetTopicConfigurationResponse) GetConfiguration() *TopicConfiguration {
-	if x != nil {
-		return x.Configuration
-	}
-	return nil
-}
-
 type FindBrokerRequest struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -734,7 +33,7 @@ type FindBrokerRequest struct {
 func (x *FindBrokerRequest) Reset() {
 	*x = FindBrokerRequest{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[11]
+		mi := &file_mq_proto_msgTypes[0]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -747,7 +46,7 @@ func (x *FindBrokerRequest) String() string {
 func (*FindBrokerRequest) ProtoMessage() {}
 
 func (x *FindBrokerRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[11]
+	mi := &file_mq_proto_msgTypes[0]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -760,7 +59,7 @@ func (x *FindBrokerRequest) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use FindBrokerRequest.ProtoReflect.Descriptor instead.
 func (*FindBrokerRequest) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{11}
+	return file_mq_proto_rawDescGZIP(), []int{0}
 }
 
 func (x *FindBrokerRequest) GetNamespace() string {
@@ -795,7 +94,7 @@ type FindBrokerResponse struct {
 func (x *FindBrokerResponse) Reset() {
 	*x = FindBrokerResponse{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[12]
+		mi := &file_mq_proto_msgTypes[1]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -808,7 +107,7 @@ func (x *FindBrokerResponse) String() string {
 func (*FindBrokerResponse) ProtoMessage() {}
 
 func (x *FindBrokerResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[12]
+	mi := &file_mq_proto_msgTypes[1]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -821,7 +120,7 @@ func (x *FindBrokerResponse) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use FindBrokerResponse.ProtoReflect.Descriptor instead.
 func (*FindBrokerResponse) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{12}
+	return file_mq_proto_rawDescGZIP(), []int{1}
 }
 
 func (x *FindBrokerResponse) GetBroker() string {
@@ -831,564 +130,32 @@ func (x *FindBrokerResponse) GetBroker() string {
 	return ""
 }
 
-type TopicConfiguration struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	PartitionCount int32                           `protobuf:"varint,1,opt,name=partition_count,json=partitionCount,proto3" json:"partition_count,omitempty"`
-	Collection     string                          `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
-	Replication    string                          `protobuf:"bytes,3,opt,name=replication,proto3" json:"replication,omitempty"`
-	IsTransient    bool                            `protobuf:"varint,4,opt,name=is_transient,json=isTransient,proto3" json:"is_transient,omitempty"`
-	Partitoning    TopicConfiguration_Partitioning `protobuf:"varint,5,opt,name=partitoning,proto3,enum=messaging_pb.TopicConfiguration_Partitioning" json:"partitoning,omitempty"`
-}
-
-func (x *TopicConfiguration) Reset() {
-	*x = TopicConfiguration{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[13]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *TopicConfiguration) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*TopicConfiguration) ProtoMessage() {}
-
-func (x *TopicConfiguration) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[13]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use TopicConfiguration.ProtoReflect.Descriptor instead.
-func (*TopicConfiguration) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{13}
-}
-
-func (x *TopicConfiguration) GetPartitionCount() int32 {
-	if x != nil {
-		return x.PartitionCount
-	}
-	return 0
-}
-
-func (x *TopicConfiguration) GetCollection() string {
-	if x != nil {
-		return x.Collection
-	}
-	return ""
-}
-
-func (x *TopicConfiguration) GetReplication() string {
-	if x != nil {
-		return x.Replication
-	}
-	return ""
-}
-
-func (x *TopicConfiguration) GetIsTransient() bool {
-	if x != nil {
-		return x.IsTransient
-	}
-	return false
-}
-
-func (x *TopicConfiguration) GetPartitoning() TopicConfiguration_Partitioning {
-	if x != nil {
-		return x.Partitoning
-	}
-	return TopicConfiguration_NonNullKeyHash
-}
-
-type SubscriberMessage_InitMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Namespace     string                                      `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"`
-	Topic         string                                      `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"`
-	Partition     int32                                       `protobuf:"varint,3,opt,name=partition,proto3" json:"partition,omitempty"`
-	StartPosition SubscriberMessage_InitMessage_StartPosition `protobuf:"varint,4,opt,name=startPosition,proto3,enum=messaging_pb.SubscriberMessage_InitMessage_StartPosition" json:"startPosition,omitempty"` // Where to begin consuming from
-	TimestampNs   int64                                       `protobuf:"varint,5,opt,name=timestampNs,proto3" json:"timestampNs,omitempty"`                                                                   // timestamp in nano seconds
-	SubscriberId  string                                      `protobuf:"bytes,6,opt,name=subscriber_id,json=subscriberId,proto3" json:"subscriber_id,omitempty"`                                              // uniquely identify a subscriber to track consumption
-}
-
-func (x *SubscriberMessage_InitMessage) Reset() {
-	*x = SubscriberMessage_InitMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[14]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *SubscriberMessage_InitMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*SubscriberMessage_InitMessage) ProtoMessage() {}
-
-func (x *SubscriberMessage_InitMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[14]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use SubscriberMessage_InitMessage.ProtoReflect.Descriptor instead.
-func (*SubscriberMessage_InitMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{0, 0}
-}
-
-func (x *SubscriberMessage_InitMessage) GetNamespace() string {
-	if x != nil {
-		return x.Namespace
-	}
-	return ""
-}
-
-func (x *SubscriberMessage_InitMessage) GetTopic() string {
-	if x != nil {
-		return x.Topic
-	}
-	return ""
-}
-
-func (x *SubscriberMessage_InitMessage) GetPartition() int32 {
-	if x != nil {
-		return x.Partition
-	}
-	return 0
-}
-
-func (x *SubscriberMessage_InitMessage) GetStartPosition() SubscriberMessage_InitMessage_StartPosition {
-	if x != nil {
-		return x.StartPosition
-	}
-	return SubscriberMessage_InitMessage_LATEST
-}
-
-func (x *SubscriberMessage_InitMessage) GetTimestampNs() int64 {
-	if x != nil {
-		return x.TimestampNs
-	}
-	return 0
-}
-
-func (x *SubscriberMessage_InitMessage) GetSubscriberId() string {
-	if x != nil {
-		return x.SubscriberId
-	}
-	return ""
-}
-
-type SubscriberMessage_AckMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	MessageId int64 `protobuf:"varint,1,opt,name=message_id,json=messageId,proto3" json:"message_id,omitempty"`
-}
-
-func (x *SubscriberMessage_AckMessage) Reset() {
-	*x = SubscriberMessage_AckMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[15]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *SubscriberMessage_AckMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*SubscriberMessage_AckMessage) ProtoMessage() {}
-
-func (x *SubscriberMessage_AckMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[15]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use SubscriberMessage_AckMessage.ProtoReflect.Descriptor instead.
-func (*SubscriberMessage_AckMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{0, 1}
-}
-
-func (x *SubscriberMessage_AckMessage) GetMessageId() int64 {
-	if x != nil {
-		return x.MessageId
-	}
-	return 0
-}
-
-type PublishRequest_InitMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` // only needed on the initial request
-	Topic     string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"`         // only needed on the initial request
-	Partition int32  `protobuf:"varint,3,opt,name=partition,proto3" json:"partition,omitempty"`
-}
-
-func (x *PublishRequest_InitMessage) Reset() {
-	*x = PublishRequest_InitMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[17]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *PublishRequest_InitMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*PublishRequest_InitMessage) ProtoMessage() {}
-
-func (x *PublishRequest_InitMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[17]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use PublishRequest_InitMessage.ProtoReflect.Descriptor instead.
-func (*PublishRequest_InitMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{3, 0}
-}
-
-func (x *PublishRequest_InitMessage) GetNamespace() string {
-	if x != nil {
-		return x.Namespace
-	}
-	return ""
-}
-
-func (x *PublishRequest_InitMessage) GetTopic() string {
-	if x != nil {
-		return x.Topic
-	}
-	return ""
-}
-
-func (x *PublishRequest_InitMessage) GetPartition() int32 {
-	if x != nil {
-		return x.Partition
-	}
-	return 0
-}
-
-type PublishResponse_ConfigMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	PartitionCount int32 `protobuf:"varint,1,opt,name=partition_count,json=partitionCount,proto3" json:"partition_count,omitempty"`
-}
-
-func (x *PublishResponse_ConfigMessage) Reset() {
-	*x = PublishResponse_ConfigMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[18]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *PublishResponse_ConfigMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*PublishResponse_ConfigMessage) ProtoMessage() {}
-
-func (x *PublishResponse_ConfigMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[18]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use PublishResponse_ConfigMessage.ProtoReflect.Descriptor instead.
-func (*PublishResponse_ConfigMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{4, 0}
-}
-
-func (x *PublishResponse_ConfigMessage) GetPartitionCount() int32 {
-	if x != nil {
-		return x.PartitionCount
-	}
-	return 0
-}
-
-type PublishResponse_RedirectMessage struct {
-	state         protoimpl.MessageState
-	sizeCache     protoimpl.SizeCache
-	unknownFields protoimpl.UnknownFields
-
-	NewBroker string `protobuf:"bytes,1,opt,name=new_broker,json=newBroker,proto3" json:"new_broker,omitempty"`
-}
-
-func (x *PublishResponse_RedirectMessage) Reset() {
-	*x = PublishResponse_RedirectMessage{}
-	if protoimpl.UnsafeEnabled {
-		mi := &file_mq_proto_msgTypes[19]
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		ms.StoreMessageInfo(mi)
-	}
-}
-
-func (x *PublishResponse_RedirectMessage) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*PublishResponse_RedirectMessage) ProtoMessage() {}
-
-func (x *PublishResponse_RedirectMessage) ProtoReflect() protoreflect.Message {
-	mi := &file_mq_proto_msgTypes[19]
-	if protoimpl.UnsafeEnabled && x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use PublishResponse_RedirectMessage.ProtoReflect.Descriptor instead.
-func (*PublishResponse_RedirectMessage) Descriptor() ([]byte, []int) {
-	return file_mq_proto_rawDescGZIP(), []int{4, 1}
-}
-
-func (x *PublishResponse_RedirectMessage) GetNewBroker() string {
-	if x != nil {
-		return x.NewBroker
-	}
-	return ""
-}
-
 var File_mq_proto protoreflect.FileDescriptor
 
 var file_mq_proto_rawDesc = []byte{
 	0x0a, 0x08, 0x6d, 0x71, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x6d, 0x65, 0x73, 0x73,
-	0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x22, 0x9e, 0x04, 0x0a, 0x11, 0x53, 0x75, 0x62,
-	0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x3f,
-	0x0a, 0x04, 0x69, 0x6e, 0x69, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x6d,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73,
-	0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x49, 0x6e,
-	0x69, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x04, 0x69, 0x6e, 0x69, 0x74, 0x12,
-	0x3c, 0x0a, 0x03, 0x61, 0x63, 0x6b, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x6d,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73,
-	0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x41, 0x63,
-	0x6b, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x03, 0x61, 0x63, 0x6b, 0x12, 0x19, 0x0a,
-	0x08, 0x69, 0x73, 0x5f, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52,
-	0x07, 0x69, 0x73, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x1a, 0xc1, 0x02, 0x0a, 0x0b, 0x49, 0x6e, 0x69,
-	0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65,
-	0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d,
-	0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1c, 0x0a, 0x09,
-	0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52,
-	0x09, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x5f, 0x0a, 0x0d, 0x73, 0x74,
-	0x61, 0x72, 0x74, 0x50, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28,
-	0x0e, 0x32, 0x39, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62,
-	0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61,
-	0x67, 0x65, 0x2e, 0x49, 0x6e, 0x69, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x53,
-	0x74, 0x61, 0x72, 0x74, 0x50, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0d, 0x73, 0x74,
-	0x61, 0x72, 0x74, 0x50, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x20, 0x0a, 0x0b, 0x74,
-	0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x4e, 0x73, 0x18, 0x05, 0x20, 0x01, 0x28, 0x03,
-	0x52, 0x0b, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x4e, 0x73, 0x12, 0x23, 0x0a,
-	0x0d, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x06,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x72,
-	0x49, 0x64, 0x22, 0x38, 0x0a, 0x0d, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x6f, 0x73, 0x69, 0x74,
-	0x69, 0x6f, 0x6e, 0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x41, 0x54, 0x45, 0x53, 0x54, 0x10, 0x00, 0x12,
-	0x0c, 0x0a, 0x08, 0x45, 0x41, 0x52, 0x4c, 0x49, 0x45, 0x53, 0x54, 0x10, 0x01, 0x12, 0x0d, 0x0a,
-	0x09, 0x54, 0x49, 0x4d, 0x45, 0x53, 0x54, 0x41, 0x4d, 0x50, 0x10, 0x02, 0x1a, 0x2b, 0x0a, 0x0a,
-	0x41, 0x63, 0x6b, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x6d, 0x65,
-	0x73, 0x73, 0x61, 0x67, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09,
-	0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x64, 0x22, 0xee, 0x01, 0x0a, 0x07, 0x4d, 0x65,
-	0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x26, 0x0a, 0x0d, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x74,
-	0x69, 0x6d, 0x65, 0x5f, 0x6e, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x03, 0x42, 0x02, 0x30, 0x01,
-	0x52, 0x0b, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x4e, 0x73, 0x12, 0x10, 0x0a,
-	0x03, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12,
-	0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x05,
-	0x76, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x3c, 0x0a, 0x07, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73,
-	0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x22, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69,
-	0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x48, 0x65,
-	0x61, 0x64, 0x65, 0x72, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x07, 0x68, 0x65, 0x61, 0x64,
-	0x65, 0x72, 0x73, 0x12, 0x19, 0x0a, 0x08, 0x69, 0x73, 0x5f, 0x63, 0x6c, 0x6f, 0x73, 0x65, 0x18,
-	0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, 0x69, 0x73, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x1a, 0x3a,
-	0x0a, 0x0c, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10,
-	0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79,
-	0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52,
-	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0x3a, 0x0a, 0x0d, 0x42, 0x72,
-	0x6f, 0x6b, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x29, 0x0a, 0x04, 0x64,
-	0x61, 0x74, 0x61, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x6d, 0x65, 0x73, 0x73,
-	0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
-	0x52, 0x04, 0x64, 0x61, 0x74, 0x61, 0x22, 0xda, 0x01, 0x0a, 0x0e, 0x50, 0x75, 0x62, 0x6c, 0x69,
-	0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3c, 0x0a, 0x04, 0x69, 0x6e, 0x69,
-	0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x28, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x49, 0x6e, 0x69, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x65, 0x52, 0x04, 0x69, 0x6e, 0x69, 0x74, 0x12, 0x29, 0x0a, 0x04, 0x64, 0x61, 0x74, 0x61, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e,
-	0x67, 0x5f, 0x70, 0x62, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x04, 0x64, 0x61,
-	0x74, 0x61, 0x1a, 0x5f, 0x0a, 0x0b, 0x49, 0x6e, 0x69, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x65, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12,
-	0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
-	0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1c, 0x0a, 0x09, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69,
-	0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x09, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74,
-	0x69, 0x6f, 0x6e, 0x22, 0xaa, 0x02, 0x0a, 0x0f, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x43, 0x0a, 0x06, 0x63, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x4d, 0x65, 0x73,
-	0x73, 0x61, 0x67, 0x65, 0x52, 0x06, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x49, 0x0a, 0x08,
-	0x72, 0x65, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2d,
-	0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x75,
-	0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2e, 0x52, 0x65,
-	0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x08, 0x72,
-	0x65, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x73, 0x5f, 0x63, 0x6c,
-	0x6f, 0x73, 0x65, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x08, 0x69, 0x73, 0x43, 0x6c,
-	0x6f, 0x73, 0x65, 0x64, 0x1a, 0x38, 0x0a, 0x0d, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x4d, 0x65,
-	0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x27, 0x0a, 0x0f, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69,
-	0x6f, 0x6e, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0e,
-	0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x1a, 0x30,
-	0x0a, 0x0f, 0x52, 0x65, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x6e, 0x65, 0x77, 0x5f, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x65, 0x77, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72,
-	0x22, 0x48, 0x0a, 0x12, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70,
-	0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73,
-	0x70, 0x61, 0x63, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x22, 0x15, 0x0a, 0x13, 0x44, 0x65,
-	0x6c, 0x65, 0x74, 0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
-	0x65, 0x22, 0x93, 0x01, 0x0a, 0x15, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x54,
-	0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x6e,
-	0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09,
-	0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70,
-	0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12,
-	0x46, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x20, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69,
-	0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x43, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67,
-	0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x18, 0x0a, 0x16, 0x43, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x75, 0x72, 0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
-	0x65, 0x22, 0x52, 0x0a, 0x1c, 0x47, 0x65, 0x74, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x43, 0x6f, 0x6e,
-	0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12,
-	0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
-	0x74, 0x6f, 0x70, 0x69, 0x63, 0x22, 0x67, 0x0a, 0x1d, 0x47, 0x65, 0x74, 0x54, 0x6f, 0x70, 0x69,
-	0x63, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x46, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67,
-	0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x20, 0x2e,
-	0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70,
-	0x69, 0x63, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52,
-	0x0d, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x63,
-	0x0a, 0x11, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63,
-	0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x72, 0x69, 0x74,
-	0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x70, 0x61, 0x72, 0x69, 0x74,
-	0x69, 0x6f, 0x6e, 0x22, 0x2c, 0x0a, 0x12, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65,
-	0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x72, 0x6f,
-	0x6b, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65,
-	0x72, 0x22, 0xb4, 0x02, 0x0a, 0x12, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x43, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27, 0x0a, 0x0f, 0x70, 0x61, 0x72, 0x74,
-	0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28,
-	0x05, 0x52, 0x0e, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x75, 0x6e,
-	0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x69, 0x6f,
-	0x6e, 0x12, 0x20, 0x0a, 0x0b, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74,
-	0x69, 0x6f, 0x6e, 0x12, 0x21, 0x0a, 0x0c, 0x69, 0x73, 0x5f, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x69,
-	0x65, 0x6e, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0b, 0x69, 0x73, 0x54, 0x72, 0x61,
-	0x6e, 0x73, 0x69, 0x65, 0x6e, 0x74, 0x12, 0x4f, 0x0a, 0x0b, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74,
-	0x6f, 0x6e, 0x69, 0x6e, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x2d, 0x2e, 0x6d, 0x65,
-	0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x54, 0x6f, 0x70, 0x69, 0x63,
-	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x50, 0x61,
-	0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x69, 0x6e, 0x67, 0x52, 0x0b, 0x70, 0x61, 0x72, 0x74,
-	0x69, 0x74, 0x6f, 0x6e, 0x69, 0x6e, 0x67, 0x22, 0x3f, 0x0a, 0x0c, 0x50, 0x61, 0x72, 0x74, 0x69,
-	0x74, 0x69, 0x6f, 0x6e, 0x69, 0x6e, 0x67, 0x12, 0x12, 0x0a, 0x0e, 0x4e, 0x6f, 0x6e, 0x4e, 0x75,
-	0x6c, 0x6c, 0x4b, 0x65, 0x79, 0x48, 0x61, 0x73, 0x68, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x4b,
-	0x65, 0x79, 0x48, 0x61, 0x73, 0x68, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x52, 0x6f, 0x75, 0x6e,
-	0x64, 0x52, 0x6f, 0x62, 0x69, 0x6e, 0x10, 0x02, 0x32, 0xad, 0x04, 0x0a, 0x10, 0x53, 0x65, 0x61,
-	0x77, 0x65, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x12, 0x4f, 0x0a,
-	0x09, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x1f, 0x2e, 0x6d, 0x65, 0x73,
-	0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72,
-	0x69, 0x62, 0x65, 0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x1b, 0x2e, 0x6d, 0x65,
-	0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x42, 0x72, 0x6f, 0x6b, 0x65,
-	0x72, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x4c,
-	0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x1c, 0x2e, 0x6d, 0x65, 0x73, 0x73,
-	0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68,
-	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x54, 0x0a, 0x0b,
-	0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x20, 0x2e, 0x6d, 0x65,
-	0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74,
-	0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x21, 0x2e,
-	0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x44, 0x65, 0x6c,
-	0x65, 0x74, 0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x12, 0x5d, 0x0a, 0x0e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x54,
-	0x6f, 0x70, 0x69, 0x63, 0x12, 0x23, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67,
-	0x5f, 0x70, 0x62, 0x2e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x54, 0x6f, 0x70,
-	0x69, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x24, 0x2e, 0x6d, 0x65, 0x73, 0x73,
-	0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75,
-	0x72, 0x65, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
-	0x00, 0x12, 0x72, 0x0a, 0x15, 0x47, 0x65, 0x74, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x43, 0x6f, 0x6e,
-	0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x2a, 0x2e, 0x6d, 0x65, 0x73,
-	0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x47, 0x65, 0x74, 0x54, 0x6f, 0x70,
-	0x69, 0x63, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2b, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69,
-	0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x47, 0x65, 0x74, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x43, 0x6f,
-	0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x51, 0x0a, 0x0a, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f,
-	0x6b, 0x65, 0x72, 0x12, 0x1f, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f,
-	0x70, 0x62, 0x2e, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67,
-	0x5f, 0x70, 0x62, 0x2e, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x4e, 0x0a, 0x0c, 0x73, 0x65, 0x61, 0x77,
-	0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, 0x6d, 0x71, 0x42, 0x10, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
-	0x51, 0x75, 0x65, 0x75, 0x65, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68,
-	0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66,
-	0x2f, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f,
-	0x70, 0x62, 0x2f, 0x6d, 0x71, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x22, 0x63, 0x0a, 0x11, 0x46, 0x69, 0x6e, 0x64,
+	0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a,
+	0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74,
+	0x6f, 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69,
+	0x63, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x72, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20,
+	0x01, 0x28, 0x05, 0x52, 0x08, 0x70, 0x61, 0x72, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x2c, 0x0a,
+	0x12, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x32, 0x65, 0x0a, 0x10, 0x53,
+	0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x12,
+	0x51, 0x0a, 0x0a, 0x46, 0x69, 0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x1f, 0x2e,
+	0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x46, 0x69, 0x6e,
+	0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20,
+	0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x46, 0x69,
+	0x6e, 0x64, 0x42, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x42, 0x4e, 0x0a, 0x0c, 0x73, 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e,
+	0x6d, 0x71, 0x42, 0x10, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x51, 0x75, 0x65, 0x75, 0x65, 0x50,
+	0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d,
+	0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61, 0x77, 0x65,
+	0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x6d, 0x71, 0x5f,
+	0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -1403,62 +170,19 @@ func file_mq_proto_rawDescGZIP() []byte {
 	return file_mq_proto_rawDescData
 }
 
-var file_mq_proto_enumTypes = make([]protoimpl.EnumInfo, 2)
-var file_mq_proto_msgTypes = make([]protoimpl.MessageInfo, 20)
+var file_mq_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
 var file_mq_proto_goTypes = []interface{}{
-	(SubscriberMessage_InitMessage_StartPosition)(0), // 0: messaging_pb.SubscriberMessage.InitMessage.StartPosition
-	(TopicConfiguration_Partitioning)(0),             // 1: messaging_pb.TopicConfiguration.Partitioning
-	(*SubscriberMessage)(nil),                        // 2: messaging_pb.SubscriberMessage
-	(*Message)(nil),                                  // 3: messaging_pb.Message
-	(*BrokerMessage)(nil),                            // 4: messaging_pb.BrokerMessage
-	(*PublishRequest)(nil),                           // 5: messaging_pb.PublishRequest
-	(*PublishResponse)(nil),                          // 6: messaging_pb.PublishResponse
-	(*DeleteTopicRequest)(nil),                       // 7: messaging_pb.DeleteTopicRequest
-	(*DeleteTopicResponse)(nil),                      // 8: messaging_pb.DeleteTopicResponse
-	(*ConfigureTopicRequest)(nil),                    // 9: messaging_pb.ConfigureTopicRequest
-	(*ConfigureTopicResponse)(nil),                   // 10: messaging_pb.ConfigureTopicResponse
-	(*GetTopicConfigurationRequest)(nil),             // 11: messaging_pb.GetTopicConfigurationRequest
-	(*GetTopicConfigurationResponse)(nil),            // 12: messaging_pb.GetTopicConfigurationResponse
-	(*FindBrokerRequest)(nil),                        // 13: messaging_pb.FindBrokerRequest
-	(*FindBrokerResponse)(nil),                       // 14: messaging_pb.FindBrokerResponse
-	(*TopicConfiguration)(nil),                       // 15: messaging_pb.TopicConfiguration
-	(*SubscriberMessage_InitMessage)(nil),            // 16: messaging_pb.SubscriberMessage.InitMessage
-	(*SubscriberMessage_AckMessage)(nil),             // 17: messaging_pb.SubscriberMessage.AckMessage
-	nil,                                              // 18: messaging_pb.Message.HeadersEntry
-	(*PublishRequest_InitMessage)(nil),               // 19: messaging_pb.PublishRequest.InitMessage
-	(*PublishResponse_ConfigMessage)(nil),            // 20: messaging_pb.PublishResponse.ConfigMessage
-	(*PublishResponse_RedirectMessage)(nil),          // 21: messaging_pb.PublishResponse.RedirectMessage
+	(*FindBrokerRequest)(nil),  // 0: messaging_pb.FindBrokerRequest
+	(*FindBrokerResponse)(nil), // 1: messaging_pb.FindBrokerResponse
 }
 var file_mq_proto_depIdxs = []int32{
-	16, // 0: messaging_pb.SubscriberMessage.init:type_name -> messaging_pb.SubscriberMessage.InitMessage
-	17, // 1: messaging_pb.SubscriberMessage.ack:type_name -> messaging_pb.SubscriberMessage.AckMessage
-	18, // 2: messaging_pb.Message.headers:type_name -> messaging_pb.Message.HeadersEntry
-	3,  // 3: messaging_pb.BrokerMessage.data:type_name -> messaging_pb.Message
-	19, // 4: messaging_pb.PublishRequest.init:type_name -> messaging_pb.PublishRequest.InitMessage
-	3,  // 5: messaging_pb.PublishRequest.data:type_name -> messaging_pb.Message
-	20, // 6: messaging_pb.PublishResponse.config:type_name -> messaging_pb.PublishResponse.ConfigMessage
-	21, // 7: messaging_pb.PublishResponse.redirect:type_name -> messaging_pb.PublishResponse.RedirectMessage
-	15, // 8: messaging_pb.ConfigureTopicRequest.configuration:type_name -> messaging_pb.TopicConfiguration
-	15, // 9: messaging_pb.GetTopicConfigurationResponse.configuration:type_name -> messaging_pb.TopicConfiguration
-	1,  // 10: messaging_pb.TopicConfiguration.partitoning:type_name -> messaging_pb.TopicConfiguration.Partitioning
-	0,  // 11: messaging_pb.SubscriberMessage.InitMessage.startPosition:type_name -> messaging_pb.SubscriberMessage.InitMessage.StartPosition
-	2,  // 12: messaging_pb.SeaweedMessaging.Subscribe:input_type -> messaging_pb.SubscriberMessage
-	5,  // 13: messaging_pb.SeaweedMessaging.Publish:input_type -> messaging_pb.PublishRequest
-	7,  // 14: messaging_pb.SeaweedMessaging.DeleteTopic:input_type -> messaging_pb.DeleteTopicRequest
-	9,  // 15: messaging_pb.SeaweedMessaging.ConfigureTopic:input_type -> messaging_pb.ConfigureTopicRequest
-	11, // 16: messaging_pb.SeaweedMessaging.GetTopicConfiguration:input_type -> messaging_pb.GetTopicConfigurationRequest
-	13, // 17: messaging_pb.SeaweedMessaging.FindBroker:input_type -> messaging_pb.FindBrokerRequest
-	4,  // 18: messaging_pb.SeaweedMessaging.Subscribe:output_type -> messaging_pb.BrokerMessage
-	6,  // 19: messaging_pb.SeaweedMessaging.Publish:output_type -> messaging_pb.PublishResponse
-	8,  // 20: messaging_pb.SeaweedMessaging.DeleteTopic:output_type -> messaging_pb.DeleteTopicResponse
-	10, // 21: messaging_pb.SeaweedMessaging.ConfigureTopic:output_type -> messaging_pb.ConfigureTopicResponse
-	12, // 22: messaging_pb.SeaweedMessaging.GetTopicConfiguration:output_type -> messaging_pb.GetTopicConfigurationResponse
-	14, // 23: messaging_pb.SeaweedMessaging.FindBroker:output_type -> messaging_pb.FindBrokerResponse
-	18, // [18:24] is the sub-list for method output_type
-	12, // [12:18] is the sub-list for method input_type
-	12, // [12:12] is the sub-list for extension type_name
-	12, // [12:12] is the sub-list for extension extendee
-	0,  // [0:12] is the sub-list for field type_name
+	0, // 0: messaging_pb.SeaweedMessaging.FindBroker:input_type -> messaging_pb.FindBrokerRequest
+	1, // 1: messaging_pb.SeaweedMessaging.FindBroker:output_type -> messaging_pb.FindBrokerResponse
+	1, // [1:2] is the sub-list for method output_type
+	0, // [0:1] is the sub-list for method input_type
+	0, // [0:0] is the sub-list for extension type_name
+	0, // [0:0] is the sub-list for extension extendee
+	0, // [0:0] is the sub-list for field type_name
 }
 
 func init() { file_mq_proto_init() }
@@ -1468,138 +192,6 @@ func file_mq_proto_init() {
 	}
 	if !protoimpl.UnsafeEnabled {
 		file_mq_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*SubscriberMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Message); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*BrokerMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*PublishRequest); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*PublishResponse); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DeleteTopicRequest); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DeleteTopicResponse); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*ConfigureTopicRequest); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*ConfigureTopicResponse); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetTopicConfigurationRequest); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetTopicConfigurationResponse); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[11].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*FindBrokerRequest); i {
 			case 0:
 				return &v.state
@@ -1611,7 +203,7 @@ func file_mq_proto_init() {
 				return nil
 			}
 		}
-		file_mq_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} {
+		file_mq_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*FindBrokerResponse); i {
 			case 0:
 				return &v.state
@@ -1623,92 +215,19 @@ func file_mq_proto_init() {
 				return nil
 			}
 		}
-		file_mq_proto_msgTypes[13].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*TopicConfiguration); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[14].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*SubscriberMessage_InitMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[15].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*SubscriberMessage_AckMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[17].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*PublishRequest_InitMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[18].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*PublishResponse_ConfigMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
-		file_mq_proto_msgTypes[19].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*PublishResponse_RedirectMessage); i {
-			case 0:
-				return &v.state
-			case 1:
-				return &v.sizeCache
-			case 2:
-				return &v.unknownFields
-			default:
-				return nil
-			}
-		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_mq_proto_rawDesc,
-			NumEnums:      2,
-			NumMessages:   20,
+			NumEnums:      0,
+			NumMessages:   2,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
 		GoTypes:           file_mq_proto_goTypes,
 		DependencyIndexes: file_mq_proto_depIdxs,
-		EnumInfos:         file_mq_proto_enumTypes,
 		MessageInfos:      file_mq_proto_msgTypes,
 	}.Build()
 	File_mq_proto = out.File
diff --git a/weed/pb/mq_pb/mq_grpc.pb.go b/weed/pb/mq_pb/mq_grpc.pb.go
index 3c986fc78..60258e23d 100644
--- a/weed/pb/mq_pb/mq_grpc.pb.go
+++ b/weed/pb/mq_pb/mq_grpc.pb.go
@@ -18,11 +18,6 @@ const _ = grpc.SupportPackageIsVersion7
 //
 // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
 type SeaweedMessagingClient interface {
-	Subscribe(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_SubscribeClient, error)
-	Publish(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishClient, error)
-	DeleteTopic(ctx context.Context, in *DeleteTopicRequest, opts ...grpc.CallOption) (*DeleteTopicResponse, error)
-	ConfigureTopic(ctx context.Context, in *ConfigureTopicRequest, opts ...grpc.CallOption) (*ConfigureTopicResponse, error)
-	GetTopicConfiguration(ctx context.Context, in *GetTopicConfigurationRequest, opts ...grpc.CallOption) (*GetTopicConfigurationResponse, error)
 	FindBroker(ctx context.Context, in *FindBrokerRequest, opts ...grpc.CallOption) (*FindBrokerResponse, error)
 }
 
@@ -34,95 +29,6 @@ func NewSeaweedMessagingClient(cc grpc.ClientConnInterface) SeaweedMessagingClie
 	return &seaweedMessagingClient{cc}
 }
 
-func (c *seaweedMessagingClient) Subscribe(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_SubscribeClient, error) {
-	stream, err := c.cc.NewStream(ctx, &SeaweedMessaging_ServiceDesc.Streams[0], "/messaging_pb.SeaweedMessaging/Subscribe", opts...)
-	if err != nil {
-		return nil, err
-	}
-	x := &seaweedMessagingSubscribeClient{stream}
-	return x, nil
-}
-
-type SeaweedMessaging_SubscribeClient interface {
-	Send(*SubscriberMessage) error
-	Recv() (*BrokerMessage, error)
-	grpc.ClientStream
-}
-
-type seaweedMessagingSubscribeClient struct {
-	grpc.ClientStream
-}
-
-func (x *seaweedMessagingSubscribeClient) Send(m *SubscriberMessage) error {
-	return x.ClientStream.SendMsg(m)
-}
-
-func (x *seaweedMessagingSubscribeClient) Recv() (*BrokerMessage, error) {
-	m := new(BrokerMessage)
-	if err := x.ClientStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
-
-func (c *seaweedMessagingClient) Publish(ctx context.Context, opts ...grpc.CallOption) (SeaweedMessaging_PublishClient, error) {
-	stream, err := c.cc.NewStream(ctx, &SeaweedMessaging_ServiceDesc.Streams[1], "/messaging_pb.SeaweedMessaging/Publish", opts...)
-	if err != nil {
-		return nil, err
-	}
-	x := &seaweedMessagingPublishClient{stream}
-	return x, nil
-}
-
-type SeaweedMessaging_PublishClient interface {
-	Send(*PublishRequest) error
-	Recv() (*PublishResponse, error)
-	grpc.ClientStream
-}
-
-type seaweedMessagingPublishClient struct {
-	grpc.ClientStream
-}
-
-func (x *seaweedMessagingPublishClient) Send(m *PublishRequest) error {
-	return x.ClientStream.SendMsg(m)
-}
-
-func (x *seaweedMessagingPublishClient) Recv() (*PublishResponse, error) {
-	m := new(PublishResponse)
-	if err := x.ClientStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
-
-func (c *seaweedMessagingClient) DeleteTopic(ctx context.Context, in *DeleteTopicRequest, opts ...grpc.CallOption) (*DeleteTopicResponse, error) {
-	out := new(DeleteTopicResponse)
-	err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/DeleteTopic", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *seaweedMessagingClient) ConfigureTopic(ctx context.Context, in *ConfigureTopicRequest, opts ...grpc.CallOption) (*ConfigureTopicResponse, error) {
-	out := new(ConfigureTopicResponse)
-	err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/ConfigureTopic", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
-func (c *seaweedMessagingClient) GetTopicConfiguration(ctx context.Context, in *GetTopicConfigurationRequest, opts ...grpc.CallOption) (*GetTopicConfigurationResponse, error) {
-	out := new(GetTopicConfigurationResponse)
-	err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/GetTopicConfiguration", in, out, opts...)
-	if err != nil {
-		return nil, err
-	}
-	return out, nil
-}
-
 func (c *seaweedMessagingClient) FindBroker(ctx context.Context, in *FindBrokerRequest, opts ...grpc.CallOption) (*FindBrokerResponse, error) {
 	out := new(FindBrokerResponse)
 	err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/FindBroker", in, out, opts...)
@@ -136,11 +42,6 @@ func (c *seaweedMessagingClient) FindBroker(ctx context.Context, in *FindBrokerR
 // All implementations must embed UnimplementedSeaweedMessagingServer
 // for forward compatibility
 type SeaweedMessagingServer interface {
-	Subscribe(SeaweedMessaging_SubscribeServer) error
-	Publish(SeaweedMessaging_PublishServer) error
-	DeleteTopic(context.Context, *DeleteTopicRequest) (*DeleteTopicResponse, error)
-	ConfigureTopic(context.Context, *ConfigureTopicRequest) (*ConfigureTopicResponse, error)
-	GetTopicConfiguration(context.Context, *GetTopicConfigurationRequest) (*GetTopicConfigurationResponse, error)
 	FindBroker(context.Context, *FindBrokerRequest) (*FindBrokerResponse, error)
 	mustEmbedUnimplementedSeaweedMessagingServer()
 }
@@ -149,21 +50,6 @@ type SeaweedMessagingServer interface {
 type UnimplementedSeaweedMessagingServer struct {
 }
 
-func (UnimplementedSeaweedMessagingServer) Subscribe(SeaweedMessaging_SubscribeServer) error {
-	return status.Errorf(codes.Unimplemented, "method Subscribe not implemented")
-}
-func (UnimplementedSeaweedMessagingServer) Publish(SeaweedMessaging_PublishServer) error {
-	return status.Errorf(codes.Unimplemented, "method Publish not implemented")
-}
-func (UnimplementedSeaweedMessagingServer) DeleteTopic(context.Context, *DeleteTopicRequest) (*DeleteTopicResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method DeleteTopic not implemented")
-}
-func (UnimplementedSeaweedMessagingServer) ConfigureTopic(context.Context, *ConfigureTopicRequest) (*ConfigureTopicResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method ConfigureTopic not implemented")
-}
-func (UnimplementedSeaweedMessagingServer) GetTopicConfiguration(context.Context, *GetTopicConfigurationRequest) (*GetTopicConfigurationResponse, error) {
-	return nil, status.Errorf(codes.Unimplemented, "method GetTopicConfiguration not implemented")
-}
 func (UnimplementedSeaweedMessagingServer) FindBroker(context.Context, *FindBrokerRequest) (*FindBrokerResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method FindBroker not implemented")
 }
@@ -180,112 +66,6 @@ func RegisterSeaweedMessagingServer(s grpc.ServiceRegistrar, srv SeaweedMessagin
 	s.RegisterService(&SeaweedMessaging_ServiceDesc, srv)
 }
 
-func _SeaweedMessaging_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(SeaweedMessagingServer).Subscribe(&seaweedMessagingSubscribeServer{stream})
-}
-
-type SeaweedMessaging_SubscribeServer interface {
-	Send(*BrokerMessage) error
-	Recv() (*SubscriberMessage, error)
-	grpc.ServerStream
-}
-
-type seaweedMessagingSubscribeServer struct {
-	grpc.ServerStream
-}
-
-func (x *seaweedMessagingSubscribeServer) Send(m *BrokerMessage) error {
-	return x.ServerStream.SendMsg(m)
-}
-
-func (x *seaweedMessagingSubscribeServer) Recv() (*SubscriberMessage, error) {
-	m := new(SubscriberMessage)
-	if err := x.ServerStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
-
-func _SeaweedMessaging_Publish_Handler(srv interface{}, stream grpc.ServerStream) error {
-	return srv.(SeaweedMessagingServer).Publish(&seaweedMessagingPublishServer{stream})
-}
-
-type SeaweedMessaging_PublishServer interface {
-	Send(*PublishResponse) error
-	Recv() (*PublishRequest, error)
-	grpc.ServerStream
-}
-
-type seaweedMessagingPublishServer struct {
-	grpc.ServerStream
-}
-
-func (x *seaweedMessagingPublishServer) Send(m *PublishResponse) error {
-	return x.ServerStream.SendMsg(m)
-}
-
-func (x *seaweedMessagingPublishServer) Recv() (*PublishRequest, error) {
-	m := new(PublishRequest)
-	if err := x.ServerStream.RecvMsg(m); err != nil {
-		return nil, err
-	}
-	return m, nil
-}
-
-func _SeaweedMessaging_DeleteTopic_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(DeleteTopicRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(SeaweedMessagingServer).DeleteTopic(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/messaging_pb.SeaweedMessaging/DeleteTopic",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(SeaweedMessagingServer).DeleteTopic(ctx, req.(*DeleteTopicRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _SeaweedMessaging_ConfigureTopic_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(ConfigureTopicRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(SeaweedMessagingServer).ConfigureTopic(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/messaging_pb.SeaweedMessaging/ConfigureTopic",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(SeaweedMessagingServer).ConfigureTopic(ctx, req.(*ConfigureTopicRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
-func _SeaweedMessaging_GetTopicConfiguration_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
-	in := new(GetTopicConfigurationRequest)
-	if err := dec(in); err != nil {
-		return nil, err
-	}
-	if interceptor == nil {
-		return srv.(SeaweedMessagingServer).GetTopicConfiguration(ctx, in)
-	}
-	info := &grpc.UnaryServerInfo{
-		Server:     srv,
-		FullMethod: "/messaging_pb.SeaweedMessaging/GetTopicConfiguration",
-	}
-	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
-		return srv.(SeaweedMessagingServer).GetTopicConfiguration(ctx, req.(*GetTopicConfigurationRequest))
-	}
-	return interceptor(ctx, in, info, handler)
-}
-
 func _SeaweedMessaging_FindBroker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(FindBrokerRequest)
 	if err := dec(in); err != nil {
@@ -312,35 +92,10 @@ var SeaweedMessaging_ServiceDesc = grpc.ServiceDesc{
 	HandlerType: (*SeaweedMessagingServer)(nil),
 	Methods: []grpc.MethodDesc{
 		{
-			MethodName: "DeleteTopic",
-			Handler:    _SeaweedMessaging_DeleteTopic_Handler,
-		},
-		{
-			MethodName: "ConfigureTopic",
-			Handler:    _SeaweedMessaging_ConfigureTopic_Handler,
-		},
-		{
-			MethodName: "GetTopicConfiguration",
-			Handler:    _SeaweedMessaging_GetTopicConfiguration_Handler,
-		},
-		{
 			MethodName: "FindBroker",
 			Handler:    _SeaweedMessaging_FindBroker_Handler,
 		},
 	},
-	Streams: []grpc.StreamDesc{
-		{
-			StreamName:    "Subscribe",
-			Handler:       _SeaweedMessaging_Subscribe_Handler,
-			ServerStreams: true,
-			ClientStreams: true,
-		},
-		{
-			StreamName:    "Publish",
-			Handler:       _SeaweedMessaging_Publish_Handler,
-			ServerStreams: true,
-			ClientStreams: true,
-		},
-	},
+	Streams:  []grpc.StreamDesc{},
 	Metadata: "mq.proto",
 }
diff --git a/weed/server/filer_grpc_server_admin.go b/weed/server/filer_grpc_server_admin.go
index df5b8fa1e..21fd8cc19 100644
--- a/weed/server/filer_grpc_server_admin.go
+++ b/weed/server/filer_grpc_server_admin.go
@@ -104,75 +104,3 @@ func (fs *FilerServer) GetFilerConfiguration(ctx context.Context, req *filer_pb.
 
 	return t, nil
 }
-
-func (fs *FilerServer) KeepConnected(stream filer_pb.SeaweedFiler_KeepConnectedServer) error {
-
-	req, err := stream.Recv()
-	if err != nil {
-		return err
-	}
-
-	clientName := util.JoinHostPort(req.Name, int(req.GrpcPort))
-	m := make(map[string]bool)
-	for _, tp := range req.Resources {
-		m[tp] = true
-	}
-	fs.brokersLock.Lock()
-	fs.brokers[clientName] = m
-	glog.V(0).Infof("+ broker %v", clientName)
-	fs.brokersLock.Unlock()
-
-	defer func() {
-		fs.brokersLock.Lock()
-		delete(fs.brokers, clientName)
-		glog.V(0).Infof("- broker %v: %v", clientName, err)
-		fs.brokersLock.Unlock()
-	}()
-
-	for {
-		if err := stream.Send(&filer_pb.KeepConnectedResponse{}); err != nil {
-			glog.V(0).Infof("send broker %v: %+v", clientName, err)
-			return err
-		}
-		// println("replied")
-
-		if _, err := stream.Recv(); err != nil {
-			glog.V(0).Infof("recv broker %v: %v", clientName, err)
-			return err
-		}
-		// println("received")
-	}
-
-}
-
-func (fs *FilerServer) LocateBroker(ctx context.Context, req *filer_pb.LocateBrokerRequest) (resp *filer_pb.LocateBrokerResponse, err error) {
-
-	resp = &filer_pb.LocateBrokerResponse{}
-
-	fs.brokersLock.Lock()
-	defer fs.brokersLock.Unlock()
-
-	var localBrokers []*filer_pb.LocateBrokerResponse_Resource
-
-	for b, m := range fs.brokers {
-		if _, found := m[req.Resource]; found {
-			resp.Found = true
-			resp.Resources = []*filer_pb.LocateBrokerResponse_Resource{
-				{
-					GrpcAddresses: b,
-					ResourceCount: int32(len(m)),
-				},
-			}
-			return
-		}
-		localBrokers = append(localBrokers, &filer_pb.LocateBrokerResponse_Resource{
-			GrpcAddresses: b,
-			ResourceCount: int32(len(m)),
-		})
-	}
-
-	resp.Resources = localBrokers
-
-	return resp, nil
-
-}
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 6ccb33996..31f8c8022 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -90,9 +90,6 @@ type FilerServer struct {
 	knownListenersLock sync.Mutex
 	knownListeners     map[int32]int32
 
-	brokers     map[string]map[string]bool
-	brokersLock sync.Mutex
-
 	inFlightDataSize      int64
 	inFlightDataLimitCond *sync.Cond
 }
@@ -112,7 +109,6 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		option:                option,
 		grpcDialOption:        security.LoadClientTLS(util.GetViper(), "grpc.filer"),
 		knownListeners:        make(map[int32]int32),
-		brokers:               make(map[string]map[string]bool),
 		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
 	}
 	fs.listenersCond = sync.NewCond(&fs.listenersLock)