do create topic

weed/mq/broker/broker_grpc_create.go

@@ -0,0 +1,43 @@
+package broker
+
+import (
+	"context"
+	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
+	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// CreateTopic Runs on any broker, but proxied to the balancer if not the balancer
+func (broker *MessageQueueBroker) CreateTopic(ctx context.Context, request *mq_pb.CreateTopicRequest) (resp *mq_pb.CreateTopicResponse, 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.CreateTopic(ctx, request)
+			return nil
+		})
+		if proxyErr != nil {
+			return nil, proxyErr
+		}
+		return resp, err
+	}
+
+	ret := &mq_pb.CreateTopicResponse{}
+	ret.BrokerPartitionAssignments, err = broker.Balancer.LookupOrAllocateTopicPartitions(request.Topic, true, request.PartitionCount)
+
+	return ret, err
+}
+
+func (broker *MessageQueueBroker) DoCreateTopic(ctx context.Context, req *mq_pb.DoCreateTopicRequest) (resp *mq_pb.DoCreateTopicResponse, err error) {
+	ret := &mq_pb.DoCreateTopicResponse{}
+	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
+}

weed/mq/broker/broker_grpc_lookup.go

@@ -7,26 +7,6 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-func (broker *MessageQueueBroker) CreateTopic(ctx context.Context, request *mq_pb.CreateTopicRequest) (resp *mq_pb.CreateTopicResponse, 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.CreateTopic(ctx, request)
-			return nil
-		})
-		if proxyErr != nil {
-			return nil, proxyErr
-		}
-		return resp, err
-	}
-
-	ret := &mq_pb.CreateTopicResponse{}
-	ret.BrokerPartitionAssignments, err = broker.Balancer.LookupOrAllocateTopicPartitions(request.Topic, true, request.PartitionCount)
-	return ret, err
-}
-
 // FindTopicBrokers returns the brokers that are serving the topic
 //
 //  1. lock the topic

weed/pb/mq.proto

@@ -28,6 +28,8 @@ service SeaweedMessaging {
     }
     rpc CreateTopic (CreateTopicRequest) returns (CreateTopicResponse) {
     }
+    rpc DoCreateTopic (DoCreateTopicRequest) returns (DoCreateTopicResponse) {
+    }
     // a pub client will call this to get the topic partitions assignment
     rpc RequestTopicPartitions (RequestTopicPartitionsRequest) returns (RequestTopicPartitionsResponse) {
     }
@@ -137,6 +139,12 @@ message CreateTopicRequest {
 message CreateTopicResponse {
     repeated BrokerPartitionAssignment broker_partition_assignments = 2;
 }
+message DoCreateTopicRequest {
+    Topic topic = 1;
+    Partition partition = 2;
+}
+message DoCreateTopicResponse {
+}
 message LookupTopicBrokersRequest {
     Topic topic = 1;
     bool is_for_publish = 2;

weed/pb/mq_pb/mq_grpc.pb.go

@@ -32,6 +32,7 @@ type SeaweedMessagingClient interface {
 	// control plane for topic partitions
 	LookupTopicBrokers(ctx context.Context, in *LookupTopicBrokersRequest, opts ...grpc.CallOption) (*LookupTopicBrokersResponse, error)
 	CreateTopic(ctx context.Context, in *CreateTopicRequest, opts ...grpc.CallOption) (*CreateTopicResponse, error)
+	DoCreateTopic(ctx context.Context, in *DoCreateTopicRequest, opts ...grpc.CallOption) (*DoCreateTopicResponse, error)
 	// a pub client will call this to get the topic partitions assignment
 	RequestTopicPartitions(ctx context.Context, in *RequestTopicPartitionsRequest, opts ...grpc.CallOption) (*RequestTopicPartitionsResponse, error)
 	AssignTopicPartitions(ctx context.Context, in *AssignTopicPartitionsRequest, opts ...grpc.CallOption) (*AssignTopicPartitionsResponse, error)
@@ -134,6 +135,15 @@ func (c *seaweedMessagingClient) CreateTopic(ctx context.Context, in *CreateTopi
 	return out, nil
 }
 
+func (c *seaweedMessagingClient) DoCreateTopic(ctx context.Context, in *DoCreateTopicRequest, opts ...grpc.CallOption) (*DoCreateTopicResponse, error) {
+	out := new(DoCreateTopicResponse)
+	err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/DoCreateTopic", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *seaweedMessagingClient) RequestTopicPartitions(ctx context.Context, in *RequestTopicPartitionsRequest, opts ...grpc.CallOption) (*RequestTopicPartitionsResponse, error) {
 	out := new(RequestTopicPartitionsResponse)
 	err := c.cc.Invoke(ctx, "/messaging_pb.SeaweedMessaging/RequestTopicPartitions", in, out, opts...)
@@ -238,6 +248,7 @@ type SeaweedMessagingServer interface {
 	// control plane for topic partitions
 	LookupTopicBrokers(context.Context, *LookupTopicBrokersRequest) (*LookupTopicBrokersResponse, error)
 	CreateTopic(context.Context, *CreateTopicRequest) (*CreateTopicResponse, error)
+	DoCreateTopic(context.Context, *DoCreateTopicRequest) (*DoCreateTopicResponse, error)
 	// a pub client will call this to get the topic partitions assignment
 	RequestTopicPartitions(context.Context, *RequestTopicPartitionsRequest) (*RequestTopicPartitionsResponse, error)
 	AssignTopicPartitions(context.Context, *AssignTopicPartitionsRequest) (*AssignTopicPartitionsResponse, error)
@@ -273,6 +284,9 @@ func (UnimplementedSeaweedMessagingServer) LookupTopicBrokers(context.Context, *
 func (UnimplementedSeaweedMessagingServer) CreateTopic(context.Context, *CreateTopicRequest) (*CreateTopicResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method CreateTopic not implemented")
 }
+func (UnimplementedSeaweedMessagingServer) DoCreateTopic(context.Context, *DoCreateTopicRequest) (*DoCreateTopicResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method DoCreateTopic not implemented")
+}
 func (UnimplementedSeaweedMessagingServer) RequestTopicPartitions(context.Context, *RequestTopicPartitionsRequest) (*RequestTopicPartitionsResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method RequestTopicPartitions not implemented")
 }
@@ -435,6 +449,24 @@ func _SeaweedMessaging_CreateTopic_Handler(srv interface{}, ctx context.Context,
 	return interceptor(ctx, in, info, handler)
 }
 
+func _SeaweedMessaging_DoCreateTopic_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DoCreateTopicRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SeaweedMessagingServer).DoCreateTopic(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/messaging_pb.SeaweedMessaging/DoCreateTopic",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SeaweedMessagingServer).DoCreateTopic(ctx, req.(*DoCreateTopicRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _SeaweedMessaging_RequestTopicPartitions_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(RequestTopicPartitionsRequest)
 	if err := dec(in); err != nil {
@@ -567,6 +599,10 @@ var SeaweedMessaging_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "CreateTopic",
 			Handler:    _SeaweedMessaging_CreateTopic_Handler,
 		},
+		{
+			MethodName: "DoCreateTopic",
+			Handler:    _SeaweedMessaging_DoCreateTopic_Handler,
+		},
 		{
 			MethodName: "RequestTopicPartitions",
 			Handler:    _SeaweedMessaging_RequestTopicPartitions_Handler,

weed/shell/command_mq_topic_create.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"context"
+	"encoding/json"
 	"flag"
 	"fmt"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -58,7 +59,8 @@ func (c *commandMqTopicCreate) Do(args []string, commandEnv *CommandEnv, writer
 		if err != nil {
 			return err
 		}
-		fmt.Fprintf(writer, "response: %+v\n", resp)
+		output, _ := json.MarshalIndent(resp, "", "  ")
+		fmt.Fprintf(writer, "response:\n%+v\n", string(output))
 		return nil
 	})