Add ConsumeAfterSeek interface (#11818)

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

internal/msgstream/mq_msgstream.go

@@ -529,7 +529,7 @@ func (ms *mqMsgStream) Seek(msgPositions []*internalpb.MsgPosition) error {
 			return err
 		}
 		log.Debug("MsgStream seek finished", zap.Any("MessageID", messageID))
-		if _, ok := consumer.(*mqclient.PulsarConsumer); ok {
+		if consumer.ConsumeAfterSeek() {
 			log.Debug("MsgStream start to pop one message after seek")
 			msg, ok := <-consumer.Chan()
 			if !ok {
@@ -885,9 +885,9 @@ func (ms *MqTtMsgStream) Seek(msgPositions []*internalpb.MsgPosition) error {
 		// rmq seek behavior (position, ...)
 		// pulsar seek behavior [position, ...)
 		// skip one tt for pulsar
-		_, ok := consumer.(*mqclient.RmqConsumer)
+
 		runLoop := false
-		if !ok {
+		if consumer.ConsumeAfterSeek() {
 			runLoop = true
 		}
 		for runLoop {

internal/util/mqclient/consumer.go

@@ -85,6 +85,9 @@ type Consumer interface {
 	// Make sure that msg is received. Only used in pulsar
 	Ack(ConsumerMessage)
 
+	// ConsumeAfterSeek defines the behavior whether to consume after seeking is done
+	ConsumeAfterSeek() bool
+
 	// Close consumer
 	Close()
 }

internal/util/mqclient/pulsar_consumer.go

@@ -80,6 +80,11 @@ func (pc *PulsarConsumer) Seek(id MessageID) error {
 	return err
 }
 
+// ConsumeAfterSeek defines pulsar consumer SHOULD consume after seek
+func (pc *PulsarConsumer) ConsumeAfterSeek() bool {
+	return true
+}
+
 func (pc *PulsarConsumer) Ack(message ConsumerMessage) {
 	pm := message.(*pulsarMessage)
 	pc.c.Ack(pm.msg)

internal/util/mqclient/rmq_consumer.go

@@ -61,6 +61,11 @@ func (rc *RmqConsumer) Seek(id MessageID) error {
 	return rc.c.Seek(msgID)
 }
 
+// ConsumeAfterSeek defines rmq consumer should NOT consume after seek
+func (rc *RmqConsumer) ConsumeAfterSeek() bool {
+	return false
+}
+
 // Ack is used to ask a rocksmq message
 func (rc *RmqConsumer) Ack(message ConsumerMessage) {
 }