You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2023/01/03 12:50:51 UTC
[pulsar-client-go] branch master updated: [Issue 918] [Refactor] Remove the clearMessageQueuesCh in partitionConsumer.dispatcher() (#921)
This is an automated email from the ASF dual-hosted git repository.
zike pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new 44dc85c [Issue 918] [Refactor] Remove the clearMessageQueuesCh in partitionConsumer.dispatcher() (#921)
44dc85c is described below
commit 44dc85c9323f5e64d1acf250fcafe8580617ce52
Author: Jiaqi Shen <18...@163.com>
AuthorDate: Tue Jan 3 20:50:45 2023 +0800
[Issue 918] [Refactor] Remove the clearMessageQueuesCh in partitionConsumer.dispatcher() (#921)
Master Issue: #918
### Motivation
The two chanel `clearMessageQueuesCh` and `clearQueueCb` just need to keep one.
For more details please check #918.
This PR does not ony aim to clean up, but also fix the potential bug in `clearMessageQueuesCh`.
For example, the `clearMessageQueuesCh` do the jod including clearing messageCh. But it may cause problem when `SeekByTime` invoked on partition topic.
https://github.com/apache/pulsar-client-go/blob/1d3499a18d526b4b1aef0bdbbc54ac812b8ae0c0/pulsar/consumer_impl.go#L614-L626
https://github.com/apache/pulsar-client-go/blob/1d3499a18d526b4b1aef0bdbbc54ac812b8ae0c0/pulsar/consumer_partition.go#L1168-L1175
When consume the partition topic, all the `partitionConsumer` share the same `messageCh`. After `SeekByTime` on partitioned topic, the messageCh may be cleared more than one time which will cause the messages losing. Suppose there is such a situation, partitionConsumer-1 has cleared its messageCh and queueCh. When partitionConsumer-2 do the clear job, it can also exec this logic.
https://github.com/apache/pulsar-client-go/blob/1d3499a18d526b4b1aef0bdbbc54ac812b8ae0c0/pulsar/consumer_partition.go#L1172-L1174
But `messageCh` is a share chan, partitionConsumer-1 may received new messages and put them to `messageCh` at this moment. There is such a possibility that partitionConsumer-2 **cleared the new messages** from `messageCh`.
### Modifications
- Remove the `clearMessageQueuesCh` in `partitionConsumer.dispatcher()`
- Modify the `clearQueueCb` in `partitionConsumer.dispatcher()`
---
pulsar/consumer_impl.go | 17 ++++++++-
pulsar/consumer_partition.go | 88 ++++++++++++++++++++------------------------
pulsar/reader_impl.go | 6 ++-
3 files changed, 60 insertions(+), 51 deletions(-)
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index 545b57c..e0120ad 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -640,7 +640,16 @@ func (c *consumer) Seek(msgID MessageID) error {
return err
}
- return c.consumers[msgID.PartitionIdx()].Seek(msgID)
+ if err := c.consumers[msgID.PartitionIdx()].Seek(msgID); err != nil {
+ return err
+ }
+
+ // clear messageCh
+ for len(c.messageCh) > 0 {
+ <-c.messageCh
+ }
+
+ return nil
}
func (c *consumer) SeekByTime(time time.Time) error {
@@ -654,6 +663,12 @@ func (c *consumer) SeekByTime(time time.Time) error {
errs = pkgerrors.Wrap(newError(SeekFailed, err.Error()), msg)
}
}
+
+ // clear messageCh
+ for len(c.messageCh) > 0 {
+ <-c.messageCh
+ }
+
return errs
}
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 7f5b6df..5ee09ab 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -145,15 +145,14 @@ type partitionConsumer struct {
// the size of the queue channel for buffering messages
queueSize int32
queueCh chan []*message
- startMessageID trackingMessageID
+ startMessageID atomicMessageID
lastDequeuedMsg trackingMessageID
- eventsCh chan interface{}
- connectedCh chan struct{}
- connectClosedCh chan connectionClosed
- closeCh chan struct{}
- clearQueueCh chan func(id trackingMessageID)
- clearMessageQueuesCh chan chan struct{}
+ eventsCh chan interface{}
+ connectedCh chan struct{}
+ connectClosedCh chan connectionClosed
+ closeCh chan struct{}
+ clearQueueCh chan func(id trackingMessageID)
nackTracker *negativeAcksTracker
dlq *dlqRouter
@@ -212,6 +211,24 @@ func (p *availablePermits) reset() {
atomic.StoreInt32(&p.permits, 0)
}
+// atomicMessageID is a wrapper for trackingMessageID to make get and set atomic
+type atomicMessageID struct {
+ msgID trackingMessageID
+ sync.RWMutex
+}
+
+func (a *atomicMessageID) get() trackingMessageID {
+ a.RLock()
+ defer a.RUnlock()
+ return a.msgID
+}
+
+func (a *atomicMessageID) set(msgID trackingMessageID) {
+ a.Lock()
+ defer a.Unlock()
+ a.msgID = msgID
+}
+
type schemaInfoCache struct {
lock sync.RWMutex
cache map[string]Schema
@@ -277,13 +294,12 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
eventsCh: make(chan interface{}, 10),
queueSize: int32(options.receiverQueueSize),
queueCh: make(chan []*message, options.receiverQueueSize),
- startMessageID: options.startMessageID,
+ startMessageID: atomicMessageID{msgID: options.startMessageID},
connectedCh: make(chan struct{}),
messageCh: messageCh,
connectClosedCh: make(chan connectionClosed, 10),
closeCh: make(chan struct{}),
clearQueueCh: make(chan func(id trackingMessageID)),
- clearMessageQueuesCh: make(chan chan struct{}),
compressionProviders: sync.Map{},
dlq: dlq,
metrics: metrics,
@@ -324,14 +340,14 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
pc.log.Info("Created consumer")
pc.setConsumerState(consumerReady)
- if pc.options.startMessageIDInclusive && pc.startMessageID.equal(lastestMessageID.(messageID)) {
+ if pc.options.startMessageIDInclusive && pc.startMessageID.get().equal(lastestMessageID.(messageID)) {
msgID, err := pc.requestGetLastMessageID()
if err != nil {
pc.nackTracker.Close()
return nil, err
}
if msgID.entryID != noMessageEntry {
- pc.startMessageID = msgID
+ pc.startMessageID.set(msgID)
// use the WithoutClear version because the dispatcher is not started yet
err = pc.requestSeekWithoutClear(msgID.messageID)
@@ -671,7 +687,7 @@ func (pc *partitionConsumer) requestSeek(msgID messageID) error {
if err := pc.requestSeekWithoutClear(msgID); err != nil {
return err
}
- pc.clearMessageChannels()
+ pc.clearReceiverQueue()
return nil
}
@@ -742,13 +758,7 @@ func (pc *partitionConsumer) internalSeekByTime(seek *seekByTimeRequest) {
seek.err = err
return
}
- pc.clearMessageChannels()
-}
-
-func (pc *partitionConsumer) clearMessageChannels() {
- doneCh := make(chan struct{})
- pc.clearMessageQueuesCh <- doneCh
- <-doneCh
+ pc.clearReceiverQueue()
}
func (pc *partitionConsumer) internalAck(req *ackRequest) {
@@ -1057,7 +1067,7 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
}
func (pc *partitionConsumer) messageShouldBeDiscarded(msgID trackingMessageID) bool {
- if pc.startMessageID.Undefined() {
+ if pc.startMessageID.get().Undefined() {
return false
}
// if we start at latest message, we should never discard
@@ -1066,11 +1076,11 @@ func (pc *partitionConsumer) messageShouldBeDiscarded(msgID trackingMessageID) b
}
if pc.options.startMessageIDInclusive {
- return pc.startMessageID.greater(msgID.messageID)
+ return pc.startMessageID.get().greater(msgID.messageID)
}
// Non inclusive
- return pc.startMessageID.greaterEqual(msgID.messageID)
+ return pc.startMessageID.get().greaterEqual(msgID.messageID)
}
// create EncryptionContext from message metadata
@@ -1218,37 +1228,19 @@ func (pc *partitionConsumer) dispatcher() {
go func() {
pc.queueCh <- nil
}()
+
for m := range pc.queueCh {
// the queue has been drained
if m == nil {
break
} else if nextMessageInQueue.Undefined() {
- nextMessageInQueue = m[0].msgID.(trackingMessageID)
+ nextMessageInQueue, _ = toTrackingMessageID(m[0].msgID)
}
}
- clearQueueCb(nextMessageInQueue)
-
- case doneCh := <-pc.clearMessageQueuesCh:
- for len(pc.queueCh) > 0 {
- <-pc.queueCh
- }
- for len(pc.messageCh) > 0 {
- <-pc.messageCh
- }
messages = nil
- // reset available permits
- pc.availablePermits.reset()
- initialPermits := uint32(pc.queueSize)
-
- pc.log.Debugf("dispatcher requesting initial permits=%d", initialPermits)
- // send initial permits
- if err := pc.internalFlow(initialPermits); err != nil {
- pc.log.WithError(err).Error("unable to send initial permits to broker")
- }
-
- close(doneCh)
+ clearQueueCb(nextMessageInQueue)
}
}
}
@@ -1490,10 +1482,10 @@ func (pc *partitionConsumer) grabConn() error {
KeySharedMeta: keySharedMeta,
}
- pc.startMessageID = pc.clearReceiverQueue()
+ pc.startMessageID.set(pc.clearReceiverQueue())
if pc.options.subscriptionMode != durable {
// For regular subscriptions the broker will determine the restarting point
- cmdSubscribe.StartMessageId = convertToMessageIDData(pc.startMessageID)
+ cmdSubscribe.StartMessageId = convertToMessageIDData(pc.startMessageID.get())
}
if len(pc.options.metadata) > 0 {
@@ -1571,8 +1563,8 @@ func (pc *partitionConsumer) clearQueueAndGetNextMessage() trackingMessageID {
func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
nextMessageInQueue := pc.clearQueueAndGetNextMessage()
- if pc.startMessageID.Undefined() {
- return pc.startMessageID
+ if pc.startMessageID.get().Undefined() {
+ return pc.startMessageID.get()
}
if !nextMessageInQueue.Undefined() {
@@ -1583,7 +1575,7 @@ func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
return pc.lastDequeuedMsg
} else {
// No message was received or dequeued by this consumer. Next message would still be the startMessageId
- return pc.startMessageID
+ return pc.startMessageID.get()
}
}
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 3931f36..079754b 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -185,11 +185,13 @@ func (r *reader) hasMoreMessages() bool {
}
if r.pc.options.startMessageIDInclusive {
- return r.lastMessageInBroker.isEntryIDValid() && r.lastMessageInBroker.greaterEqual(r.pc.startMessageID.messageID)
+ return r.lastMessageInBroker.isEntryIDValid() &&
+ r.lastMessageInBroker.greaterEqual(r.pc.startMessageID.get().messageID)
}
// Non-inclusive
- return r.lastMessageInBroker.isEntryIDValid() && r.lastMessageInBroker.greater(r.pc.startMessageID.messageID)
+ return r.lastMessageInBroker.isEntryIDValid() &&
+ r.lastMessageInBroker.greater(r.pc.startMessageID.get().messageID)
}
func (r *reader) Close() {