aboutsummaryrefslogtreecommitdiff
path: root/weed/mq/broker/broker_grpc_create.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/mq/broker/broker_grpc_create.go')
-rw-r--r--weed/mq/broker/broker_grpc_create.go72
1 files changed, 0 insertions, 72 deletions
diff --git a/weed/mq/broker/broker_grpc_create.go b/weed/mq/broker/broker_grpc_create.go
deleted file mode 100644
index 15b3efd26..000000000
--- a/weed/mq/broker/broker_grpc_create.go
+++ /dev/null
@@ -1,72 +0,0 @@
-package broker
-
-import (
- "context"
- "fmt"
- "github.com/seaweedfs/seaweedfs/weed/mq/balancer"
- "github.com/seaweedfs/seaweedfs/weed/mq/topic"
- "github.com/seaweedfs/seaweedfs/weed/pb"
- "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
- "google.golang.org/grpc/codes"
- "google.golang.org/grpc/status"
-)
-
-// ConfigureTopic Runs on any broker, but proxied to the balancer if not the balancer
-func (broker *MessageQueueBroker) ConfigureTopic(ctx context.Context, request *mq_pb.ConfigureTopicRequest) (resp *mq_pb.ConfigureTopicResponse, err error) {
- if broker.currentBalancer == "" {
- return nil, status.Errorf(codes.Unavailable, "no balancer")
- }
- if !broker.lockAsBalancer.IsLocked() {
- proxyErr := broker.withBrokerClient(false, broker.currentBalancer, func(client mq_pb.SeaweedMessagingClient) error {
- resp, err = client.ConfigureTopic(ctx, request)
- return nil
- })
- if proxyErr != nil {
- return nil, proxyErr
- }
- return resp, err
- }
-
- ret := &mq_pb.ConfigureTopicResponse{}
- ret.BrokerPartitionAssignments, err = broker.Balancer.LookupOrAllocateTopicPartitions(request.Topic, true, request.PartitionCount)
-
- for _, bpa := range ret.BrokerPartitionAssignments {
- // fmt.Printf("create topic %s on %s\n", request.Topic, bpa.LeaderBroker)
- if doCreateErr := broker.withBrokerClient(false, pb.ServerAddress(bpa.LeaderBroker), func(client mq_pb.SeaweedMessagingClient) error {
- _, doCreateErr := client.DoConfigureTopic(ctx, &mq_pb.DoConfigureTopicRequest{
- Topic: request.Topic,
- Partition: bpa.Partition,
- })
- if doCreateErr != nil {
- return fmt.Errorf("do create topic %s on %s: %v", request.Topic, bpa.LeaderBroker, doCreateErr)
- }
- brokerStats, found := broker.Balancer.Brokers.Get(bpa.LeaderBroker)
- if !found {
- brokerStats = balancer.NewBrokerStats()
- if !broker.Balancer.Brokers.SetIfAbsent(bpa.LeaderBroker, brokerStats) {
- brokerStats, _ = broker.Balancer.Brokers.Get(bpa.LeaderBroker)
- }
- }
- brokerStats.RegisterAssignment(request.Topic, bpa.Partition)
- return nil
- }); doCreateErr != nil {
- return nil, doCreateErr
- }
- }
-
- // TODO revert if some error happens in the middle of the assignments
-
- return ret, err
-}
-
-func (broker *MessageQueueBroker) DoConfigureTopic(ctx context.Context, req *mq_pb.DoConfigureTopicRequest) (resp *mq_pb.DoConfigureTopicResponse, err error) {
- ret := &mq_pb.DoConfigureTopicResponse{}
- t, p := topic.FromPbTopic(req.Topic), topic.FromPbPartition(req.Partition)
- localTopicPartition := broker.localTopicManager.GetTopicPartition(t, p)
- if localTopicPartition == nil {
- localTopicPartition = topic.NewLocalPartition(t, p, true, nil)
- broker.localTopicManager.AddTopicPartition(t, localTopicPartition)
- }
-
- return ret, err
-}