include key in LogEntry

photosssa · Mar 7, 2024 · 28f8f33 · 28f8f33
1 parent 34f2b60
commit 28f8f33
Show file tree

Hide file tree

Showing 5 changed files with 275 additions and 264 deletions.
diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto
@@ -364,6 +364,7 @@ message LogEntry {
     int64 ts_ns = 1;
     int32 partition_key_hash = 2;
     bytes data = 3;
+    bytes key = 4;
 }
 
 message KeepConnectedRequest {

diff --git a/weed/mq/broker/broker_grpc_sub.go b/weed/mq/broker/broker_grpc_sub.go
@@ -109,11 +109,10 @@ func (b *MessageQueueBroker) SubscribeMessage(req *mq_pb.SubscribeMessageRequest
 		// reset the sleep interval count
 		sleepIntervalCount = 0
 
-		value := logEntry.GetData()
 		if err := stream.Send(&mq_pb.SubscribeMessageResponse{Message: &mq_pb.SubscribeMessageResponse_Data{
 			Data: &mq_pb.DataMessage{
-				Key:   []byte(fmt.Sprintf("key-%d", logEntry.PartitionKeyHash)),
-				Value: value,
+				Key:   logEntry.Key,
+				Value: logEntry.Data,
 				TsNs:  logEntry.TsNs,
 			},
 		}}); err != nil {

diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto
@@ -364,6 +364,7 @@ message LogEntry {
     int64 ts_ns = 1;
     int32 partition_key_hash = 2;
     bytes data = 3;
+    bytes key = 4;
 }
 
 message KeepConnectedRequest {