weed/pb/mq.proto


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244

syntax = "proto3";

package messaging_pb;

option go_package = "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb";
option java_package = "seaweedfs.mq";
option java_outer_classname = "MessagQueueProto";

//////////////////////////////////////////////////

service SeaweedMessaging {

    // control plane
    rpc FindBrokerLeader (FindBrokerLeaderRequest) returns (FindBrokerLeaderResponse) {
    }
    rpc AssignSegmentBrokers (AssignSegmentBrokersRequest) returns (AssignSegmentBrokersResponse) {
    }
    rpc CheckSegmentStatus (CheckSegmentStatusRequest) returns (CheckSegmentStatusResponse) {
    }
    rpc CheckBrokerLoad (CheckBrokerLoadRequest) returns (CheckBrokerLoadResponse) {
    }

    // control plane for balancer
    rpc ConnectToBalancer (stream ConnectToBalancerRequest) returns (stream ConnectToBalancerResponse) {
    }
    // control plane for topic partitions
    rpc LookupTopicBrokers (LookupTopicBrokersRequest) returns (LookupTopicBrokersResponse) {
    }
    rpc CreateTopic (CreateTopicRequest) returns (CreateTopicResponse) {
    }
    rpc DoCreateTopic (DoCreateTopicRequest) returns (DoCreateTopicResponse) {
    }
    rpc ListTopics (ListTopicsRequest) returns (ListTopicsResponse) {
    }
    // a pub client will call this to get the topic partitions assignment
    rpc RequestTopicPartitions (RequestTopicPartitionsRequest) returns (RequestTopicPartitionsResponse) {
    }
    rpc AssignTopicPartitions (AssignTopicPartitionsRequest) returns (AssignTopicPartitionsResponse) {
    }
    rpc CheckTopicPartitionsStatus (CheckTopicPartitionsStatusRequest) returns (CheckTopicPartitionsStatusResponse) {
    }

    // data plane
    rpc Publish (stream PublishRequest) returns (stream PublishResponse) {
    }
    rpc Subscribe (SubscribeRequest) returns (stream SubscribeResponse) {
    }
}

//////////////////////////////////////////////////
message SegmentInfo {
    Segment segment = 1;
    int64 start_ts_ns = 2;
    repeated string brokers = 3;
    int64 stop_ts_ns = 4;
    repeated int32 previous_segments = 5;
    repeated int32 next_segments = 6;
}

//////////////////////////////////////////////////

message FindBrokerLeaderRequest {
    string filer_group = 1;
}

message FindBrokerLeaderResponse {
    string broker = 1;
}

message Topic {
    string namespace = 1;
    string name = 2;
}
message Partition {
    int32 ring_size = 1;
    int32 range_start = 2;
    int32 range_stop = 3;
}

message Segment {
    string namespace = 1;
    string topic = 2;
    int32 id = 3;
    Partition partition = 4;
}

message AssignSegmentBrokersRequest {
    Segment segment = 1;
}

message AssignSegmentBrokersResponse {
    repeated string brokers = 1;
}

message CheckSegmentStatusRequest {
    Segment segment = 1;
}

message CheckSegmentStatusResponse {
    bool is_active = 1;
}

message CheckBrokerLoadRequest {
}

message CheckBrokerLoadResponse {
    int64 message_count = 1;
    int64 bytes_count = 2;

}

//////////////////////////////////////////////////
message BrokerStats {
    int32 cpu_usage_percent = 1;
    map<string, TopicPartitionStats> stats = 2;
}
message TopicPartitionStats {
    Topic topic = 1;
    Partition partition = 2;
    int32 consumer_count = 3;
    bool is_leader = 4;
}


message ConnectToBalancerRequest {
    message InitMessage {
        string broker = 1;
    }
    oneof message {
        InitMessage init = 1;
        BrokerStats stats = 2;
    }
}
message ConnectToBalancerResponse {
}
//////////////////////////////////////////////////
message CreateTopicRequest {
    Topic topic = 1;
    int32 partition_count = 2;
}
message CreateTopicResponse {
    repeated BrokerPartitionAssignment broker_partition_assignments = 2;
}
message DoCreateTopicRequest {
    Topic topic = 1;
    Partition partition = 2;
}
message DoCreateTopicResponse {
}
message ListTopicsRequest {
}
message ListTopicsResponse {
    repeated Topic topics = 1;
}
message LookupTopicBrokersRequest {
    Topic topic = 1;
    bool is_for_publish = 2;
}
message LookupTopicBrokersResponse {
    Topic topic = 1;
    repeated BrokerPartitionAssignment broker_partition_assignments = 2;
}
message BrokerPartitionAssignment {
    Partition partition = 1;
    string leader_broker = 2;
    repeated string follower_brokers = 3;
}

message RequestTopicPartitionsRequest {
    Topic topic = 1;
    int32 partition_count = 2;
}
message RequestTopicPartitionsResponse {
    repeated BrokerPartitionAssignment broker_partition_assignments = 1;
}

message AssignTopicPartitionsRequest {
    Topic topic = 1;
    repeated BrokerPartitionAssignment broker_partition_assignments = 2;
    bool is_leader = 3;
}
message AssignTopicPartitionsResponse {
}

message CheckTopicPartitionsStatusRequest {
    string namespace = 1;
    string topic = 2;
    BrokerPartitionAssignment broker_partition_assignment = 3;
    bool should_cancel_if_not_match = 4;
}
message CheckTopicPartitionsStatusResponse {
    repeated BrokerPartitionAssignment broker_partition_assignments = 1;
}

//////////////////////////////////////////////////
message DataMessage {
    bytes key = 1;
    bytes value = 2;
}
message PublishRequest {
    message InitMessage {
        Topic topic = 1;
        Partition partition = 2;
        int32 ack_interval = 3;
    }
    oneof message {
        InitMessage init = 1;
        DataMessage data = 2;
    }
    int64 sequence = 3;
}
message PublishResponse {
    int64 ack_sequence = 1;
    string error = 2;
    string redirect_to_broker = 3;
}
message SubscribeRequest {
    message Consumer {
        string consumer_group = 1;
        string consumer_id = 2;
        string client_id = 3;
    }
    message Cursor {
        Topic topic = 1;
        Partition partition = 2;
        oneof offset {
            int64 start_offset = 3;
            int64 start_timestamp_ns = 4;
        }
        string filter = 5;
    }
    Consumer consumer = 1;
    Cursor cursor = 2;
}
message SubscribeResponse {
    message CtrlMessage {
        string error = 1;
        string redirect_to_broker = 2;
    }
    oneof message {
        CtrlMessage ctrl = 1;
        DataMessage data = 2;
    }
}