temp storage

Signed-off-by: chyezh <chyezh@outlook.com>

Author: chyezh

mq/factory.go

@@ -22,9 +22,14 @@ func ParsePositionFromCheckpoint(mqType string, messageID []byte) (ifc.MessageID
 	}
 }
 
-func ParseManualMessageID(mqType string, manualID int64) (ifc.MessageID, error) {
+func ParseManualMessageID(mqType string, manualID string) (ifc.MessageID, error) {
 	// pulsar not supported yet
-	return nil, errors.Newf("not supported mq type: %s", mqType)
+	switch mqType {
+	case "pulsar":
+		return pulsar.StringToMsgID(manualID)
+	default:
+		return nil, errors.Newf("not supported mq type: %s", mqType)
+	}
 }
 
 func NewConsumer(mqType, address, channel string, config ifc.MqOption) (ifc.Consumer, error) {

mq/pulsar/message_id_data.pb.go

@@ -0,0 +1,13 @@
+syntax = "proto2";
+
+package milvus.proto.streaming.wal.pulsar;
+
+option go_package = "github.com/milvus-io/milvus/pkg/streaming/walimpls/impls/pulsar";
+
+
+message MessageIdData {
+    required uint64 ledgerId = 1;
+    required uint64 entryId = 2;
+    optional int32 partition = 3;
+    optional int32 batch_index = 4;
+}

mq/pulsar/message_id_data.proto

@@ -2,9 +2,12 @@ package pulsar
 
 import (
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/apache/pulsar-client-go/pulsar"
+	"github.com/cockroachdb/errors"
+	"github.com/gogo/protobuf/proto"
 
 	"github.com/milvus-io/birdwatcher/mq/ifc"
 )
@@ -86,6 +89,40 @@ func msgIDToString(messageID pulsar.MessageID) string {
 }
 
 // StringToMsgID is used to convert a string to message ID
-func stringToMsgID(msgString string) (pulsar.MessageID, error) {
-	return pulsar.DeserializeMessageID([]byte(msgString))
+func StringToMsgID(msgString string) (ifc.MessageID, error) {
+	ids := strings.Split(msgString, ":")
+	if len(ids) != 2 {
+		return nil, errors.Newf("invalid manual id: %s", msgString)
+	}
+	ledgerID, err := strconv.ParseUint(ids[0], 10, 64)
+	if err != nil {
+		return nil, err
+	}
+	entryID, err := strconv.ParseUint(ids[1], 10, 64)
+	if err != nil {
+		return nil, err
+	}
+	msgID := newMessageIDOfPulsar(ledgerID, entryID, 0)
+	return DeserializePulsarMsgID(msgID.Serialize())
+}
+
+// newMessageIDOfPulsar only for test.
+func newMessageIDOfPulsar(ledgerID uint64, entryID uint64, batchIdx int32) pulsar.MessageID {
+	partitionIdx := int32(0)
+	id := &MessageIdData{
+		LedgerId:   &ledgerID,
+		EntryId:    &entryID,
+		BatchIndex: &batchIdx,
+		Partition:  &partitionIdx,
+	}
+
+	msg, err := proto.Marshal(id)
+	if err != nil {
+		panic(err)
+	}
+	msgID, err := pulsar.DeserializeMessageID(msg)
+	if err != nil {
+		panic(err)
+	}
+	return msgID
 }

mq/pulsar/pulsar_id.go

@@ -3,9 +3,11 @@ package states
 import (
 	"context"
 	"fmt"
+	"os"
 	"path"
 
 	"github.com/cockroachdb/errors"
+	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/milvus-io/birdwatcher/framework"
@@ -25,10 +27,19 @@ type ConsumeParam struct {
 	Topic               string `name:"topic" default:"" desc:"topic to consume"`
 	ShardName           string `name:"shard_name" default:"" desc:"shard name(vchannel name) to filter with"`
 	Detail              bool   `name:"detail" default:"false" desc:"print msg detail"`
-	ManualID            int64  `name:"manual_id" default:"0" desc:"manual id"`
+	ManualID            string `name:"manual-id" default:"" desc:"manual id"`
+	OutputFile          string `name:"output-file" default:"" desc:"output file"`
 }
 
 func (s *InstanceState) ConsumeCommand(ctx context.Context, p *ConsumeParam) error {
+	of := os.Stdout
+	if p.OutputFile != "" {
+		var err error
+		if of, err = os.OpenFile(p.OutputFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644); err != nil {
+			return err
+		}
+	}
+
 	var messageID ifc.MessageID
 	switch p.StartPosition {
 	case "cp":
@@ -94,15 +105,20 @@ func (s *InstanceState) ConsumeCommand(ctx context.Context, p *ConsumeParam) err
 			fmt.Printf("%s ", msgType)
 			switch msgType {
 			case commonpb.MsgType_Insert, commonpb.MsgType_Delete:
+				msgID := msg.ID()
 				v, err := ParseMsg(header.GetBase().GetMsgType(), msg.Payload())
 				if err != nil {
 					fmt.Println(err.Error())
 				}
 				if p.ShardName == "" || v.GetShardName() == p.ShardName {
+					fmt.Fprintf(of, "MsgID: %s, Shard: %s\n", msgID.String(), v.GetShardName())
 					if p.Detail {
-						fmt.Print(v)
+						if s, err := protojson.Marshal(v); err == nil {
+							fmt.Fprintln(of, string(s))
+						} else {
+							fmt.Fprintln(of, err.Error())
+						}
 					} else {
-						fmt.Print(v.GetShardName())
 						err := ValidateMsg(msgType, msg.Payload())
 						if err != nil {
 							fmt.Println(err.Error())
@@ -111,7 +127,7 @@ func (s *InstanceState) ConsumeCommand(ctx context.Context, p *ConsumeParam) err
 				}
 			default:
 			}
-			fmt.Println()
+			fmt.Fprintln(of)
 		}
 		if eq, _ := msg.ID().Equal(latestID.Serialize()); eq {
 			break