Fix seek method maybe lose messages.

pulsar/consumer_impl.go

@@ -653,35 +653,39 @@ func (c *consumer) Seek(msgID MessageID) error {
 		return err
 	}
 
-	if err := c.consumers[msgID.PartitionIdx()].Seek(msgID); err != nil {
-		return err
-	}
-
 	// clear messageCh
+	// We need clear messageCh before execution SeekByTime. Otherwise, messages of after seek maybe will be clear.
+	// Note: Cleaning up messageCh will also drain the queueCh of the sub-consumers
 	for len(c.messageCh) > 0 {
 		<-c.messageCh
 	}
 
+	if err := c.consumers[msgID.PartitionIdx()].Seek(msgID); err != nil {
+		return err
+	}
+
 	return nil
 }
 
 func (c *consumer) SeekByTime(time time.Time) error {
 	c.Lock()
 	defer c.Unlock()
 	var errs error
+
+	// clear messageCh
+	// We need clear messageCh before execution SeekByTime. Otherwise, messages of after seek maybe will be clear.
+	// Note: Cleaning up messageCh will also drain the queueCh of the sub-consumers
+	for len(c.messageCh) > 0 {
+		<-c.messageCh
+	}
+
 	// run SeekByTime on every partition of topic
 	for _, cons := range c.consumers {
 		if err := cons.SeekByTime(time); err != nil {
 			msg := fmt.Sprintf("unable to SeekByTime for topic=%s subscription=%s", c.topic, c.Subscription())
 			errs = pkgerrors.Wrap(newError(SeekFailed, err.Error()), msg)
 		}
 	}
-
-	// clear messageCh
-	for len(c.messageCh) > 0 {
-		<-c.messageCh
-	}
-
 	return errs
 }