1 files changed, 28 insertions, 13 deletions

diff --git a/weed/pb/mq.proto b/weed/pb/mq.proto
index 5c309aaf1..285063b43 100644
--- a/weed/pb/mq.proto
+++ b/weed/pb/mq.proto
@@ -41,9 +41,9 @@ service SeaweedMessaging {
     }
 
     // data plane for each topic partition
-    rpc Publish (stream PublishRequest) returns (stream PublishResponse) {
+    rpc PublishMessage (stream PublishMessageRequest) returns (stream PublishMessageResponse) {
     }
-    rpc Subscribe (SubscribeRequest) returns (stream SubscribeResponse) {
+    rpc SubscribeMessage (SubscribeMessageRequest) returns (stream SubscribeMessageResponse) {
     }
 }
 
@@ -68,6 +68,24 @@ message Partition {
     int64 unix_time_ns = 4;
 }
 
+message Offset {
+    Topic topic = 1;
+    repeated PartitionOffset partition_offsets = 2;
+}
+
+enum PartitionOffsetStartType {
+    EARLIEST = 0;
+    EARLIEST_IN_MEMORY = 1;
+    LATEST = 2;
+}
+
+message PartitionOffset {
+    Partition partition = 1;
+    int64 start_ts_ns = 2;
+    int64 stop_ts_ns = 3;
+    PartitionOffsetStartType start_type = 4;
+}
+
 //////////////////////////////////////////////////
 message BrokerStats {
     int32 cpu_usage_percent = 1;
@@ -137,7 +155,7 @@ message AssignTopicPartitionsResponse {
 message SubscriberToSubCoordinatorRequest {
     message InitMessage {
         string consumer_group = 1;
-        string consumer_instance_id = 2;
+        string consumer_group_instance_id = 2;
         Topic topic = 3;
     }
     message AckMessage {
@@ -153,6 +171,7 @@ message SubscriberToSubCoordinatorResponse {
     message AssignedPartition {
         Partition partition = 1;
         int64 ts_ns = 2;
+        string broker = 3;
     }
     message Assignment {
         int64 generation = 1;
@@ -169,7 +188,7 @@ message DataMessage {
     bytes value = 2;
     int64 ts_ns = 3;
 }
-message PublishRequest {
+message PublishMessageRequest {
     message InitMessage {
         Topic topic = 1;
         Partition partition = 2;
@@ -181,23 +200,19 @@ message PublishRequest {
     }
     int64 sequence = 3;
 }
-message PublishResponse {
+message PublishMessageResponse {
     int64 ack_sequence = 1;
     string error = 2;
     bool should_close = 3;
 }
-message SubscribeRequest {
+message SubscribeMessageRequest {
     message InitMessage {
         string consumer_group = 1;
         string consumer_id = 2;
         string client_id = 3;
         Topic topic = 4;
-        Partition partition = 5;
-        oneof offset {
-            int64 start_offset = 6;
-            int64 start_timestamp_ns = 7;
-        }
-        string filter = 8;
+        PartitionOffset partition_offset = 5;
+        string filter = 6;
     }
     message AckMessage {
         int64 sequence = 1;
@@ -207,7 +222,7 @@ message SubscribeRequest {
         AckMessage ack = 2;
     }
 }
-message SubscribeResponse {
+message SubscribeMessageResponse {
     message CtrlMessage {
         string error = 1;
         bool is_end_of_stream = 2;