fix: fix seek and reconnect race

pulsar/consumer_partition.go

@@ -24,6 +24,7 @@ import (
 	"math"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -182,6 +183,8 @@ type partitionConsumer struct {
 	lastMessageInBroker *trackingMessageID
 
 	redirectedClusterURI string
+
+	seeking atomic.Bool
 }
 
 func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -909,11 +912,18 @@ func (pc *partitionConsumer) requestSeekWithoutClear(msgID *messageID) error {
 		MessageId:  id,
 	}
 
+	if !pc.seeking.CompareAndSwap(false, true) {
+		return errors.New("seek operation is already in progress")
+	}
+	defer pc.seeking.Store(false)
+
 	_, err = pc.client.rpcClient.RequestOnCnx(pc._getConn(), requestID, pb.BaseCommand_SEEK, cmdSeek)
 	if err != nil {
+		pc.seeking.Store(false)
 		pc.log.WithError(err).Error("Failed to reset to message id")
 		return err
 	}
+	pc.seeking.Store(false)
 	return nil
 }
 
@@ -936,7 +946,6 @@ func (pc *partitionConsumer) SeekByTime(time time.Time) error {
 
 func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
 	defer close(seek.doneCh)
-
 	state := pc.getConsumerState()
 	if state == consumerClosing || state == consumerClosed {
 		pc.log.WithField("state", pc.state).Error("Failed seekByTime by consumer is closing or has closed")
@@ -950,6 +959,12 @@ func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
 		MessagePublishTime: proto.Uint64(uint64(seek.publishTime.UnixNano() / int64(time.Millisecond))),
 	}
 
+	if !pc.seeking.CompareAndSwap(false, true) {
+		seek.err = errors.New("seek operation is already in progress")
+		return
+	}
+	defer pc.seeking.Store(false)
+
 	_, err := pc.client.rpcClient.RequestOnCnx(pc._getConn(), requestID, pb.BaseCommand_SEEK, cmdSeek)
 	if err != nil {
 		pc.log.WithError(err).Error("Failed to reset to message publish time")
@@ -1594,6 +1609,13 @@ func (pc *partitionConsumer) runEventsLoop() {
 				pc.log.Info("close consumer, exit reconnect")
 				return
 			case connectionClosed := <-pc.connectClosedCh:
+				if pc.seeking.Load() {
+					go func() {
+						<-time.After(100 * time.Millisecond)
+						pc.connectClosedCh <- connectionClosed
+					}()
+					continue
+				}
 				pc.log.Debug("runEventsLoop will reconnect")
 				pc.reconnectToBroker(connectionClosed)
 			}