Repository: apache/pulsar-client-go
Pull request #1334 (2025-02-20 12:59:44)

chore(deps): bump golang.org/x/net from 0.23.0 to 0.33.0
Bumps [golang.org/x/net](https://github.com/golang/net) from 0.23.0 to 0.33.0.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/golang/net/commit/dfc720dfe0cfc125116068c20efcdcb5e4eab464"><code>dfc720d</code></a> go.mod: update golang.org/x dependencies</li>
<li><a href="https://github.com/golang/net/commit/8e66b04771e35c4e4125e8c60334b34e2423effb"><code>8e66b04</code></a> html: use strings.EqualFold instead of lowering ourselves</li>
<li><a href="https://github.com/golang/net/commit/b935f7b5d723c82894e1a1fd936a94dd2d1eae46"><code>b935f7b</code></a> html: avoid endless loop on error token</li>
<li><a href="https://github.com/golang/net/commit/9af49ef148d7d8b3e4cbbd9cc0cd37f2a520a4a3"><code>9af49ef</code></a> route: remove unused sizeof* consts</li>
<li><a href="https://github.com/golang/net/commit/6705db9a4df8f2cf16aed83e773e7a0213788b7a"><code>6705db9</code></a> quic: clean up crypto streams when dropping packet protection keys</li>
<li><a href="https://github.com/golang/net/commit/4ef7588d2b3f83775099797baac43c34e2e23200"><code>4ef7588</code></a> quic: handle ACK frame in packet which drops number space</li>
<li><a href="https://github.com/golang/net/commit/552d8ac903a11a9fde71a88732f5b58b6b394178"><code>552d8ac</code></a> Revert &quot;route: change from syscall to x/sys/unix&quot;</li>
<li><a href="https://github.com/golang/net/commit/13a7c0108bd38aad013797cdb95e6bfca0bbcec6"><code>13a7c01</code></a> Revert &quot;route: remove unused sizeof* consts on freebsd&quot;</li>
<li><a href="https://github.com/golang/net/commit/285e1cf6650f407805ea8af9255624961b768479"><code>285e1cf</code></a> go.mod: update golang.org/x dependencies</li>
<li><a href="https://github.com/golang/net/commit/d0a1049b7e05d3d6f62570a8381a22301a97ae73"><code>d0a1049</code></a> route: remove unused sizeof* consts on freebsd</li>
<li>Additional commits viewable in <a href="https://github.com/golang/net/compare/v0.23.0...v0.33.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=golang.org/x/net&package-manager=go_modules&previous-version=0.23.0&new-version=0.33.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
You can disable automated security fix PRs for this repo from the [Security Alerts page](https://github.com/apache/pulsar-client-go/network/alerts).

</details>


Patch:
diff --git a/go.mod b/go.mod
index ab20b16..b8ac4c6 100644
--- a/go.mod
+++ b/go.mod
@@ -94,7 +94,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	golang.org/x/crypto v0.31.0 // indirect
-	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/term v0.27.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
diff --git a/go.sum b/go.sum
index 8820f07..98e4af3 100644
--- a/go.sum
+++ b/go.sum
@@ -269,8 +269,8 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
-golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
 golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 7aee964..d611c69 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -261,6 +261,7 @@ type ConsumerOptions struct {
 	SubscriptionMode SubscriptionMode
 
 	// StartMessageIDInclusive, if true, the consumer will start at the `StartMessageID`, included.
+	// Note: This configuration also affects the seek operation.
 	// Default is `false` and the consumer will start from the "next" message
 	StartMessageIDInclusive bool
 
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index aa922af..cc2554c 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -190,9 +190,13 @@ type partitionConsumer struct {
 	backoffPolicyFunc    func() backoff.Policy
 
 	dispatcherSeekingControlCh chan struct{}
-	isSeeking                  atomic.Bool
-	ctx                        context.Context
-	cancelFunc                 context.CancelFunc
+	// handle to the dispatcher goroutine
+	isSeeking atomic.Bool
+	// After executing seekByTime, the client is unaware of the startMessageId.
+	// Use this flag to compare markDeletePosition with BrokerLastMessageId when checking hasMoreMessages.
+	hasSoughtByTime atomic.Bool
+	ctx             context.Context
+	cancelFunc      context.CancelFunc
 }
 
 // pauseDispatchMessage used to discard the message in the dispatcher goroutine.
@@ -429,11 +433,12 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 
 	startingMessageID := pc.startMessageID.get()
 	if pc.options.startMessageIDInclusive && startingMessageID != nil && startingMessageID.equal(latestMessageID) {
-		msgID, err := pc.requestGetLastMessageID()
+		msgIDResp, err := pc.requestGetLastMessageID()
 		if err != nil {
 			pc.Close()
 			return nil, err
 		}
+		msgID := convertToMessageID(msgIDResp.GetLastMessageId())
 		if msgID.entryID != noMessageEntry {
 			pc.startMessageID.set(msgID)
 
@@ -616,18 +621,27 @@ func (pc *partitionConsumer) internalUnsubscribe(unsub *unsubscribeRequest) {
 }
 
 func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
+	res, err := pc.getLastMessageIDAndMarkDeletePosition()
+	if err != nil {
+		return nil, err
+	}
+	return res.msgID, err
+}
+
+func (pc *partitionConsumer) getLastMessageIDAndMarkDeletePosition() (*getLastMsgIDResult, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to getLastMessageID for the closing or closed consumer")
 		return nil, errors.New("failed to getLastMessageID for the closing or closed consumer")
 	}
 	bo := pc.backoffPolicyFunc()
-	request := func() (*trackingMessageID, error) {
+	request := func() (*getLastMsgIDResult, error) {
 		req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
 		pc.eventsCh <- req
 
 		// wait for the request to complete
 		<-req.doneCh
-		return req.msgID, req.err
+		res := &getLastMsgIDResult{req.msgID, req.markDeletePosition}
+		return res, req.err
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), pc.client.operationTimeout)
@@ -647,10 +661,16 @@ func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
 
 func (pc *partitionConsumer) internalGetLastMessageID(req *getLastMsgIDRequest) {
 	defer close(req.doneCh)
-	req.msgID, req.err = pc.requestGetLastMessageID()
+	rsp, err := pc.requestGetLastMessageID()
+	if err != nil {
+		req.err = err
+		return
+	}
+	req.msgID = convertToMessageID(rsp.GetLastMessageId())
+	req.markDeletePosition = convertToMessageID(rsp.GetConsumerMarkDeletePosition())
 }
 
-func (pc *partitionConsumer) requestGetLastMessageID() (*trackingMessageID, error) {
+func (pc *partitionConsumer) requestGetLastMessageID() (*pb.CommandGetLastMessageIdResponse, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to getLastMessageID closing or closed consumer")
 		return nil, errors.New("failed to getLastMessageID closing or closed consumer")
@@ -667,8 +687,7 @@ func (pc *partitionConsumer) requestGetLastMessageID() (*trackingMessageID, erro
 		pc.log.WithError(err).Error("Failed to get last message id")
 		return nil, err
 	}
-	id := res.Response.GetLastMessageIdResponse.GetLastMessageId()
-	return convertToMessageID(id), nil
+	return res.Response.GetLastMessageIdResponse, nil
 }
 
 func (pc *partitionConsumer) sendIndividualAck(msgID MessageID) *ackRequest {
@@ -997,7 +1016,15 @@ func (pc *partitionConsumer) requestSeek(msgID *messageID) error {
 	if err := pc.requestSeekWithoutClear(msgID); err != nil {
 		return err
 	}
-	pc.clearReceiverQueue()
+	// When the seek operation is successful, it indicates:
+	// 1. The broker has reset the cursor and sent a request to close the consumer on the client side.
+	//    Since this method is in the same goroutine as the reconnectToBroker,
+	//    we can safely clear the messages in the queue (at this point, it won't contain messages after the seek).
+	// 2. The startMessageID is reset to ensure accurate judgment when calling hasNext next time.
+	//    Since the messages in the queue are cleared here reconnection won't reset startMessageId.
+	pc.lastDequeuedMsg = nil
+	pc.startMessageID.set(toTrackingMessageID(msgID))
+	pc.clearQueueAndGetNextMessage()
 	return nil
 }
 
@@ -1069,7 +1096,9 @@ func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
 		seek.err = err
 		return
 	}
-	pc.clearReceiverQueue()
+	pc.lastDequeuedMsg = nil
+	pc.hasSoughtByTime.Store(true)
+	pc.clearQueueAndGetNextMessage()
 }
 
 func (pc *partitionConsumer) internalAck(req *ackRequest) {
@@ -1451,10 +1480,6 @@ func (pc *partitionConsumer) messageShouldBeDiscarded(msgID *trackingMessageID)
 	if pc.startMessageID.get() == nil {
 		return false
 	}
-	// if we start at latest message, we should never discard
-	if pc.options.startMessageID != nil && pc.options.startMessageID.equal(latestMessageID) {
-		return false
-	}
 
 	if pc.options.startMessageIDInclusive {
 		return pc.startMessageID.get().greater(msgID.messageID)
@@ -1709,9 +1734,15 @@ type redeliveryRequest struct {
 }
 
 type getLastMsgIDRequest struct {
-	doneCh chan struct{}
-	msgID  *trackingMessageID
-	err    error
+	doneCh             chan struct{}
+	msgID              *trackingMessageID
+	markDeletePosition *trackingMessageID
+	err                error
+}
+
+type getLastMsgIDResult struct {
+	msgID              *trackingMessageID
+	markDeletePosition *trackingMessageID
 }
 
 type seekRequest struct {
@@ -2200,6 +2231,25 @@ func (pc *partitionConsumer) discardCorruptedMessage(msgID *pb.MessageIdData,
 }
 
 func (pc *partitionConsumer) hasNext() bool {
+
+	// If a seek by time has been performed, then the `startMessageId` becomes irrelevant.
+	// We need to compare `markDeletePosition` and `lastMessageId`,
+	// and then reset `startMessageID` to `markDeletePosition`.
+	if pc.lastDequeuedMsg == nil && pc.hasSoughtByTime.CompareAndSwap(true, false) {
+		res, err := pc.getLastMessageIDAndMarkDeletePosition()
+		if err != nil {
+			pc.log.WithError(err).Error("Failed to get last message id")
+			pc.hasSoughtByTime.CompareAndSwap(false, true)
+			return false
+		}
+		pc.lastMessageInBroker = res.msgID
+		pc.startMessageID.set(res.markDeletePosition)
+		// We only care about comparing ledger ids and entry ids as mark delete position
+		// doesn't have other ids such as batch index
+		compareResult := pc.lastMessageInBroker.messageID.compareLedgerAndEntryID(pc.startMessageID.get().messageID)
+		return compareResult > 0 || (pc.options.startMessageIDInclusive && compareResult == 0)
+	}
+
 	if pc.lastMessageInBroker != nil && pc.hasMoreMessages() {
 		return true
 	}
@@ -2261,12 +2311,14 @@ func convertToMessageID(id *pb.MessageIdData) *trackingMessageID {
 
 	msgID := &trackingMessageID{
 		messageID: &messageID{
-			ledgerID: int64(*id.LedgerId),
-			entryID:  int64(*id.EntryId),
+			ledgerID:  int64(id.GetLedgerId()),
+			entryID:   int64(id.GetEntryId()),
+			batchIdx:  id.GetBatchIndex(),
+			batchSize: id.GetBatchSize(),
 		},
 	}
-	if id.BatchIndex != nil {
-		msgID.batchIdx = *id.BatchIndex
+	if msgID.batchIdx == -1 {
+		msgID.batchIdx = 0
 	}
 
 	return msgID
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 0acd782..f32cc7f 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -18,6 +18,7 @@
 package pulsar
 
 import (
+	"cmp"
 	"errors"
 	"fmt"
 	"math"
@@ -147,6 +148,13 @@ func (id *messageID) equal(other *messageID) bool {
 		id.batchIdx == other.batchIdx
 }
 
+func (id *messageID) compareLedgerAndEntryID(other *messageID) int {
+	if result := cmp.Compare(id.ledgerID, other.ledgerID); result != 0 {
+		return result
+	}
+	return cmp.Compare(id.entryID, other.entryID)
+}
+
 func (id *messageID) greaterEqual(other *messageID) bool {
 	return id.equal(other) || id.greater(other)
 }
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 57fc724..2ad1acb 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -18,6 +18,7 @@
 package internal
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"errors"
@@ -78,7 +79,7 @@ type ConnectionListener interface {
 type Connection interface {
 	SendRequest(requestID uint64, req *pb.BaseCommand, callback func(*pb.BaseCommand, error))
 	SendRequestNoWait(req *pb.BaseCommand) error
-	WriteData(data Buffer)
+	WriteData(ctx context.Context, data Buffer)
 	RegisterListener(id uint64, listener ConnectionListener) error
 	UnregisterListener(id uint64)
 	AddConsumeHandler(id uint64, handler ConsumerHandler) error
@@ -129,6 +130,11 @@ type request struct {
 	callback func(command *pb.BaseCommand, err error)
 }
 
+type dataRequest struct {
+	ctx  context.Context
+	data Buffer
+}
+
 type connection struct {
 	started           int32
 	connectionTimeout time.Duration
@@ -157,7 +163,7 @@ type connection struct {
 	incomingRequestsCh chan *request
 	closeCh            chan struct{}
 	readyCh            chan struct{}
-	writeRequestsCh    chan Buffer
+	writeRequestsCh    chan *dataRequest
 
 	pendingLock sync.Mutex
 	pendingReqs map[uint64]*request
@@ -209,7 +215,7 @@ func newConnection(opts connectionOptions) *connection {
 		// partition produces writing on a single connection. In general it's
 		// good to keep this above the number of partition producers assigned
 		// to a single connection.
-		writeRequestsCh:  make(chan Buffer, 256),
+		writeRequestsCh:  make(chan *dataRequest, 256),
 		listeners:        make(map[uint64]ConnectionListener),
 		consumerHandlers: make(map[uint64]ConsumerHandler),
 		metrics:          opts.metrics,
@@ -421,11 +427,11 @@ func (c *connection) run() {
 				return // TODO: this never gonna be happen
 			}
 			c.internalSendRequest(req)
-		case data := <-c.writeRequestsCh:
-			if data == nil {
+		case req := <-c.writeRequestsCh:
+			if req == nil {
 				return
 			}
-			c.internalWriteData(data)
+			c.internalWriteData(req.ctx, req.data)
 
 		case <-pingSendTicker.C:
 			c.sendPing()
@@ -450,22 +456,26 @@ func (c *connection) runPingCheck(pingCheckTicker *time.Ticker) {
 	}
 }
 
-func (c *connection) WriteData(data Buffer) {
+func (c *connection) WriteData(ctx context.Context, data Buffer) {
 	select {
-	case c.writeRequestsCh <- data:
+	case c.writeRequestsCh <- &dataRequest{ctx: ctx, data: data}:
 		// Channel is not full
 		return
-
+	case <-ctx.Done():
+		c.log.Debug("Write data context cancelled")
+		return
 	default:
 		// Channel full, fallback to probe if connection is closed
 	}
 
 	for {
 		select {
-		case c.writeRequestsCh <- data:
+		case c.writeRequestsCh <- &dataRequest{ctx: ctx, data: data}:
 			// Successfully wrote on the channel
 			return
-
+		case <-ctx.Done():
+			c.log.Debug("Write data context cancelled")
+			return
 		case <-time.After(100 * time.Millisecond):
 			// The channel is either:
 			// 1. blocked, in which case we need to wait until we have space
@@ -481,11 +491,17 @@ func (c *connection) WriteData(data Buffer) {
 
 }
 
-func (c *connection) internalWriteData(data Buffer) {
+func (c *connection) internalWriteData(ctx context.Context, data Buffer) {
 	c.log.Debug("Write data: ", data.ReadableBytes())
-	if _, err := c.cnx.Write(data.ReadableSlice()); err != nil {
-		c.log.WithError(err).Warn("Failed to write on connection")
-		c.Close()
+
+	select {
+	case <-ctx.Done():
+		return
+	default:
+		if _, err := c.cnx.Write(data.ReadableSlice()); err != nil {
+			c.log.WithError(err).Warn("Failed to write on connection")
+			c.Close()
+		}
 	}
 }
 
@@ -510,7 +526,7 @@ func (c *connection) writeCommand(cmd *pb.BaseCommand) {
 	}
 
 	c.writeBuffer.WrittenBytes(cmdSize)
-	c.internalWriteData(c.writeBuffer)
+	c.internalWriteData(context.Background(), c.writeBuffer)
 }
 
 func (c *connection) receivedCommand(cmd *pb.BaseCommand, headersAndPayload Buffer) {
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 448f780..f652312 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -394,7 +394,7 @@ func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
 			pi.sentAt = time.Now()
 			pi.Unlock()
 			p.pendingQueue.Put(pi)
-			p._getConn().WriteData(pi.buffer)
+			p._getConn().WriteData(pi.ctx, pi.buffer)
 
 			if pi == lastViewItem {
 				break
@@ -837,6 +837,8 @@ func (p *partitionProducer) internalSingleSend(
 
 type pendingItem struct {
 	sync.Mutex
+	ctx           context.Context
+	cancel        context.CancelFunc
 	buffer        internal.Buffer
 	sequenceID    uint64
 	createdAt     time.Time
@@ -895,14 +897,17 @@ func (p *partitionProducer) writeData(buffer internal.Buffer, sequenceID uint64,
 		return
 	default:
 		now := time.Now()
+		ctx, cancel := context.WithCancel(context.Background())
 		p.pendingQueue.Put(&pendingItem{
+			ctx:          ctx,
+			cancel:       cancel,
 			createdAt:    now,
 			sentAt:       now,
 			buffer:       buffer,
 			sequenceID:   sequenceID,
 			sendRequests: callbacks,
 		})
-		p._getConn().WriteData(buffer)
+		p._getConn().WriteData(ctx, buffer)
 	}
 }
 
@@ -1579,14 +1584,14 @@ type sendRequest struct {
 	uuid             string
 	chunkRecorder    *chunkRecorder
 
-	/// resource management
+	// resource management
 
 	memLimit          internal.MemoryLimitController
 	reservedMem       int64
 	semaphore         internal.Semaphore
 	reservedSemaphore int
 
-	/// convey settable state
+	// convey settable state
 
 	sendAsBatch         bool
 	transaction         *transaction
@@ -1659,7 +1664,7 @@ func (sr *sendRequest) done(msgID MessageID, err error) {
 }
 
 func (p *partitionProducer) blockIfQueueFull() bool {
-	//DisableBlockIfQueueFull == false means enable block
+	// DisableBlockIfQueueFull == false means enable block
 	return !p.options.DisableBlockIfQueueFull
 }
 
@@ -1741,6 +1746,10 @@ func (i *pendingItem) done(err error) {
 	if i.flushCallback != nil {
 		i.flushCallback(err)
 	}
+
+	if i.cancel != nil {
+		i.cancel()
+	}
 }
 
 // _setConn sets the internal connection field of this partition producer atomically.
diff --git a/pulsar/reader.go b/pulsar/reader.go
index 98bde4e..95aed0e 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -53,6 +53,7 @@ type ReaderOptions struct {
 	StartMessageID MessageID
 
 	// StartMessageIDInclusive, if true, the reader will start at the `StartMessageID`, included.
+	// Note: This configuration also affects the seek operation.
 	// Default is `false` and the reader will start from the "next" message
 	StartMessageIDInclusive bool
 
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index f76255e..55b0503 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -196,19 +196,6 @@ func (r *reader) Close() {
 	r.metrics.ReadersClosed.Inc()
 }
 
-func (r *reader) messageID(msgID MessageID) *trackingMessageID {
-	mid := toTrackingMessageID(msgID)
-
-	partition := int(mid.partitionIdx)
-	// did we receive a valid partition index?
-	if partition < 0 {
-		r.log.Warnf("invalid partition index %d expected", partition)
-		return nil
-	}
-
-	return mid
-}
-
 func (r *reader) Seek(msgID MessageID) error {
 	r.Lock()
 	defer r.Unlock()
@@ -218,9 +205,12 @@ func (r *reader) Seek(msgID MessageID) error {
 		return fmt.Errorf("invalid message id type %T", msgID)
 	}
 
-	mid := r.messageID(msgID)
-	if mid == nil {
-		return nil
+	mid := toTrackingMessageID(msgID)
+
+	partition := int(mid.partitionIdx)
+	if partition < 0 {
+		r.log.Warnf("invalid partition index %d expected", partition)
+		return fmt.Errorf("seek msgId must include partitoinIndex")
 	}
 
 	return r.c.Seek(mid)


Test patch:
diff --git a/pulsar/consumer_multitopic_test.go b/pulsar/consumer_multitopic_test.go
index cd236ec..30ae5cc 100644
--- a/pulsar/consumer_multitopic_test.go
+++ b/pulsar/consumer_multitopic_test.go
@@ -18,18 +18,20 @@
 package pulsar
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
-	"github.com/stretchr/testify/assert"
 )
 
 func TestMultiTopicConsumerReceive(t *testing.T) {
@@ -330,7 +332,7 @@ func (dummyConnection) SendRequestNoWait(_ *pb.BaseCommand) error {
 	return nil
 }
 
-func (dummyConnection) WriteData(_ internal.Buffer) {
+func (dummyConnection) WriteData(_ context.Context, _ internal.Buffer) {
 }
 
 func (dummyConnection) RegisterListener(_ uint64, _ internal.ConnectionListener) error {
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index edef440..43ce3c7 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -1262,8 +1262,9 @@ func TestConsumerSeek(t *testing.T) {
 	defer producer.Close()
 
 	consumer, err := client.Subscribe(ConsumerOptions{
-		Topic:            topicName,
-		SubscriptionName: "sub-1",
+		Topic:                   topicName,
+		SubscriptionName:        "sub-1",
+		StartMessageIDInclusive: true,
 	})
 	assert.Nil(t, err)
 	defer consumer.Close()
diff --git a/pulsar/consumer_zero_queue_test.go b/pulsar/consumer_zero_queue_test.go
index 34e9df9..2e7c4b2 100644
--- a/pulsar/consumer_zero_queue_test.go
+++ b/pulsar/consumer_zero_queue_test.go
@@ -474,6 +474,7 @@ func TestZeroQueueConsumer_Seek(t *testing.T) {
 		Topic:                   topicName,
 		EnableZeroQueueConsumer: true,
 		SubscriptionName:        "sub-1",
+		StartMessageIDInclusive: true,
 	})
 	assert.Nil(t, err)
 	_, ok := consumer.(*zeroQueueConsumer)
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index 8365357..2fbe89e 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -1070,3 +1070,170 @@ func TestReaderNextReturnsOnClosedConsumer(t *testing.T) {
 	assert.ErrorAs(t, err, &e)
 	assert.Equal(t, ConsumerClosed, e.Result())
 }
+
+func testReaderSeekByIDWithHasNext(t *testing.T, startMessageID MessageID, startMessageIDInclusive bool) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// send 100 messages
+	var lastMsgID MessageID
+	for i := 0; i < 10; i++ {
+		lastMsgID, err = producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.NoError(t, err)
+		assert.NotNil(t, lastMsgID)
+	}
+
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:                   topic,
+		StartMessageID:          startMessageID,
+		StartMessageIDInclusive: startMessageIDInclusive,
+	})
+	assert.Nil(t, err)
+	defer reader.Close()
+
+	// Seek to last message ID
+	err = reader.Seek(lastMsgID)
+	assert.NoError(t, err)
+
+	if startMessageIDInclusive {
+		assert.True(t, reader.HasNext())
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		msg, err := reader.Next(ctx)
+		assert.NoError(t, err)
+		assert.NotNil(t, msg)
+		assert.True(t, messageIDCompare(lastMsgID, msg.ID()) == 0)
+		cancel()
+	} else {
+		assert.False(t, reader.HasNext())
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		msg, err := reader.Next(ctx)
+		assert.Error(t, err)
+		assert.Nil(t, msg)
+		cancel()
+	}
+
+}
+
+func TestReaderWithSeekByID(t *testing.T) {
+	params := []struct {
+		messageID               MessageID
+		startMessageIDInclusive bool
+	}{
+		{EarliestMessageID(), false},
+		{EarliestMessageID(), true},
+		{LatestMessageID(), false},
+		{LatestMessageID(), true},
+	}
+
+	for _, c := range params {
+		t.Run(fmt.Sprintf("TestReaderSeekByID_%v_%v", c.messageID, c.startMessageIDInclusive),
+			func(t *testing.T) {
+				testReaderSeekByIDWithHasNext(t, c.messageID, c.startMessageIDInclusive)
+			})
+	}
+}
+
+func testReaderSeekByTimeWithHasNext(t *testing.T, startMessageID MessageID) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// 1. send 10 messages
+	var lastMsgID MessageID
+	for i := 0; i < 10; i++ {
+		lastMsgID, err = producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.NoError(t, err)
+
+		assert.NotNil(t, lastMsgID)
+	}
+
+	// 2. create reader
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:                   topic,
+		StartMessageID:          startMessageID,
+		StartMessageIDInclusive: false,
+	})
+	assert.Nil(t, err)
+	defer reader.Close()
+
+	// 3. Seek time to now
+	reader.SeekByTime(time.Now())
+
+	// 4. Should not receive msg
+	{
+		assert.False(t, reader.HasNext())
+		timeoutCtx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		msg, err := reader.Next(timeoutCtx)
+		assert.Error(t, err)
+		assert.Nil(t, msg)
+		cancel()
+	}
+
+	// 5. send more 10 messages
+	for i := 0; i < 10; i++ {
+		lastMsgID, err = producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello2-%d", i)),
+		})
+		assert.NoError(t, err)
+		assert.NotNil(t, lastMsgID)
+	}
+
+	// 6. Assert these messages are received
+	for i := 0; i < 10; i++ {
+		assert.True(t, reader.HasNext())
+		msg, err := reader.Next(context.Background())
+		assert.NoError(t, err)
+		assert.Equal(t, fmt.Sprintf("hello2-%d", i), string(msg.Payload()))
+	}
+
+	// assert not more msg
+	{
+		assert.False(t, reader.HasNext())
+		timeoutCtx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		msg, err := reader.Next(timeoutCtx)
+		assert.Error(t, err)
+		assert.Nil(t, msg)
+		cancel()
+	}
+}
+func TestReaderWithSeekByTime(t *testing.T) {
+	startMessageIDs := []MessageID{EarliestMessageID(), LatestMessageID()}
+	for _, startMsgID := range startMessageIDs {
+		t.Run(fmt.Sprintf("TestReaderSeekByTime_%v", startMsgID), func(t *testing.T) {
+			testReaderSeekByTimeWithHasNext(t, startMsgID)
+		})
+	}
+}