Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[refactor]: Refactor the toTrackingMessageID() #972

Merged
merged 2 commits into from
Mar 7, 2023
Merged

[refactor]: Refactor the toTrackingMessageID() #972

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
29b70f1
sync: rebase to master
Gleiphir2769 Mar 2, 2023
df9b721
fix: fix some bugs
Gleiphir2769 Mar 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions pulsar/consumer_impl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -531,10 +531,17 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
if delay < 0 {
delay = 0
}

if !checkMessageIDType(msg.ID()) {
c.log.Warnf("invalid message id type %T", msg.ID())
return
}

msgID := c.messageID(msg.ID())
if msgID == nil {
return
}

props := make(map[string]string)
for k, v := range msg.Properties() {
props[k] = v
Expand Down Expand Up @@ -580,6 +587,10 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
}

func (c *consumer) Nack(msg Message) {
if !checkMessageIDType(msg.ID()) {
c.log.Warnf("invalid message id type %T", msg.ID())
return
}
if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
mid := c.messageID(msg.ID())
if mid == nil {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Even if we check the type of msgID, we can't remove this check, because there is other logic that returns nil

pulsar-client-go/pulsar/consumer_impl.go

Lines 755 to 759 in 48b7d01

if partition < 0 || partition >= len(c.consumers) {
c.log.Warnf("invalid partition index %d expected a partition between [0-%d]",
partition, len(c.consumers))
return nil
}

We can add checkMessageIdType to c. messageID method, and keep this check.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great suggestion! I will fix it.

Expand Down Expand Up @@ -745,10 +756,6 @@ func toProtoInitialPosition(p SubscriptionInitialPosition) pb.CommandSubscribe_I

func (c *consumer) messageID(msgID MessageID) *trackingMessageID {
mid := toTrackingMessageID(msgID)
if mid == nil {
c.log.Warnf("invalid message id type %T", msgID)
return nil
}

partition := int(mid.partitionIdx)
// did we receive a valid partition index?
Expand Down
16 changes: 8 additions & 8 deletions pulsar/consumer_multitopic.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,11 @@ func (c *multiTopicConsumer) Ack(msg Message) error {

// AckID the consumption of a single message, identified by its MessageID
func (c *multiTopicConsumer) AckID(msgID MessageID) error {
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return errors.New("invalid message id type in multi_consumer")
}
mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
Expand All @@ -152,11 +152,11 @@ func (c *multiTopicConsumer) AckCumulative(msg Message) error {
// AckIDCumulative the reception of all the messages in the stream up to (and including)
// the provided message, identified by its MessageID
func (c *multiTopicConsumer) AckIDCumulative(msgID MessageID) error {
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return errors.New("invalid message id type in multi_consumer")
}
mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
Expand Down Expand Up @@ -203,11 +203,11 @@ func (c *multiTopicConsumer) ReconsumeLaterWithCustomProperties(msg Message, cus
func (c *multiTopicConsumer) Nack(msg Message) {
if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
msgID := msg.ID()
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return
}
mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
Expand All @@ -221,11 +221,11 @@ func (c *multiTopicConsumer) Nack(msg Message) {
}

func (c *multiTopicConsumer) NackID(msgID MessageID) {
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return
}
mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
Expand Down
44 changes: 31 additions & 13 deletions pulsar/consumer_partition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,9 +465,6 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
}

trackingID := toTrackingMessageID(msgID)
if trackingID == nil {
return errors.New("failed to convert trackingMessageID")
}

if trackingID != nil && trackingID.ack() {
pc.metrics.AcksCounter.Inc()
Expand Down Expand Up @@ -501,18 +498,34 @@ func (pc *partitionConsumer) sendIndividualAck(msgID MessageID) *ackRequest {
}

func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
if !checkMessageIDType(msgID) {
pc.log.Errorf("invalid message id type %T", msgID)
return fmt.Errorf("invalid message id type %T", msgID)
}
return pc.ackID(msgID, true)
}

func (pc *partitionConsumer) AckID(msgID MessageID) error {
if !checkMessageIDType(msgID) {
pc.log.Errorf("invalid message id type %T", msgID)
return fmt.Errorf("invalid message id type %T", msgID)
}
return pc.ackID(msgID, false)
}

func (pc *partitionConsumer) AckIDCumulative(msgID MessageID) error {
if !checkMessageIDType(msgID) {
pc.log.Errorf("invalid message id type %T", msgID)
return fmt.Errorf("invalid message id type %T", msgID)
}
return pc.internalAckIDCumulative(msgID, false)
}

func (pc *partitionConsumer) AckIDWithResponseCumulative(msgID MessageID) error {
if !checkMessageIDType(msgID) {
pc.log.Errorf("invalid message id type %T", msgID)
return fmt.Errorf("invalid message id type %T", msgID)
}
return pc.internalAckIDCumulative(msgID, true)
}

Expand Down Expand Up @@ -574,15 +587,17 @@ func (pc *partitionConsumer) sendCumulativeAck(msgID MessageID) *ackRequest {
}

func (pc *partitionConsumer) NackID(msgID MessageID) {
if !checkMessageIDType(msgID) {
pc.log.Warnf("invalid message id type %T", msgID)
return
}

if cmid, ok := msgID.(*chunkMessageID); ok {
pc.unAckChunksTracker.nack(cmid)
return
}

trackingID := toTrackingMessageID(msgID)
if trackingID == nil {
return
}

pc.nackTracker.Add(trackingID.messageID)
pc.metrics.NacksCounter.Inc()
Expand Down Expand Up @@ -665,16 +680,20 @@ func (pc *partitionConsumer) Seek(msgID MessageID) error {
pc.log.WithField("state", state).Error("Failed to seek by closing or closed consumer")
return errors.New("failed to seek by closing or closed consumer")
}

if !checkMessageIDType(msgID) {
pc.log.Errorf("invalid message id type %T", msgID)
return fmt.Errorf("invalid message id type %T", msgID)
}

req := &seekRequest{
doneCh: make(chan struct{}),
}
if cmid, ok := msgID.(*chunkMessageID); ok {
req.msgID = cmid.firstChunkID
} else if tmid := toTrackingMessageID(msgID); tmid != nil {
req.msgID = tmid.messageID
} else {
// will never reach
return errors.New("unhandled messageID type")
tmid := toTrackingMessageID(msgID)
req.msgID = tmid.messageID
}

pc.ackGroupingTracker.flushAndClean()
Expand Down Expand Up @@ -1812,9 +1831,8 @@ func (c *chunkedMsgCtx) discard(pc *partitionConsumer) {
continue
}
pc.log.Info("Removing chunk message-id", mid.String())
if tmid := toTrackingMessageID(mid); tmid != nil {
pc.AckID(tmid)
}
tmid := toTrackingMessageID(mid)
pc.AckID(tmid)
}
}

Expand Down
23 changes: 13 additions & 10 deletions pulsar/consumer_regex.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,12 +174,13 @@ func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customPr

// AckID the consumption of a single message, identified by its MessageID
func (c *regexConsumer) AckID(msgID MessageID) error {
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return errors.New("invalid message id type")
return fmt.Errorf("invalid message id type %T", msgID)
}

mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
return errors.New("consumer is nil in consumer_regex")
Expand All @@ -201,12 +202,13 @@ func (c *regexConsumer) AckCumulative(msg Message) error {
// AckIDCumulative the reception of all the messages in the stream up to (and including)
// the provided message, identified by its MessageID
func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return errors.New("invalid message id type")
return fmt.Errorf("invalid message id type %T", msgID)
}

mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
return errors.New("unable to ack message because consumer is nil")
Expand All @@ -222,11 +224,11 @@ func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
func (c *regexConsumer) Nack(msg Message) {
if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
msgID := msg.ID()
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return
}
mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
Expand All @@ -240,12 +242,13 @@ func (c *regexConsumer) Nack(msg Message) {
}

func (c *regexConsumer) NackID(msgID MessageID) {
mid := toTrackingMessageID(msgID)
if mid == nil {
if !checkMessageIDType(msgID) {
c.log.Warnf("invalid message id type %T", msgID)
return
}

mid := toTrackingMessageID(msgID)

if mid.consumer == nil {
c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
return
Expand Down
44 changes: 29 additions & 15 deletions pulsar/impl_message.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,16 @@ func newMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx in
}
}

func fromMessageID(msgID MessageID) *messageID {
return &messageID{
ledgerID: msgID.LedgerID(),
entryID: msgID.EntryID(),
batchIdx: msgID.BatchIdx(),
partitionIdx: msgID.PartitionIdx(),
batchSize: msgID.BatchSize(),
}
}

func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32, batchSize int32,
tracker *ackTracker) *trackingMessageID {
return &trackingMessageID{
Expand All @@ -228,22 +238,26 @@ func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partiti
}
}

func toTrackingMessageID(msgID MessageID) *trackingMessageID {
if mid, ok := msgID.(*messageID); ok {
return &trackingMessageID{
messageID: mid,
receivedTime: time.Now(),
}
} else if mid, ok := msgID.(*trackingMessageID); ok {
// checkMessageIDType checks if the MessageID is user-defined
func checkMessageIDType(msgID MessageID) (valid bool) {
switch msgID.(type) {
case *trackingMessageID:
return true
case *chunkMessageID:
return true
case *messageID:
return true
default:
return false
}
}

func toTrackingMessageID(msgID MessageID) (trackingMsgID *trackingMessageID) {
if mid, ok := msgID.(*trackingMessageID); ok {
return mid
} else if cmid, ok := msgID.(*chunkMessageID); ok {
return &trackingMessageID{
messageID: cmid.messageID,
receivedTime: cmid.receivedTime,
consumer: cmid.consumer,
}
} else {
return nil
}
return &trackingMessageID{
messageID: fromMessageID(msgID),
}
}

Expand Down
30 changes: 14 additions & 16 deletions pulsar/reader_impl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,19 +51,18 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
return nil, newError(InvalidConfiguration, "StartMessageID is required")
}

startMessageID := toTrackingMessageID(options.StartMessageID)
if startMessageID == nil {
var startMessageID *trackingMessageID
if !checkMessageIDType(options.StartMessageID) {
// a custom type satisfying MessageID may not be a messageID or trackingMessageID
// so re-create messageID using its data
deserMsgID, err := deserializeMessageID(options.StartMessageID.Serialize())
if err != nil {
return nil, err
}
// de-serialized MessageID is a messageID
startMessageID = &trackingMessageID{
messageID: deserMsgID.(*messageID),
receivedTime: time.Now(),
}
startMessageID = toTrackingMessageID(deserMsgID)
} else {
startMessageID = toTrackingMessageID(options.StartMessageID)
}

subscriptionName := options.SubscriptionName
Expand Down Expand Up @@ -148,12 +147,10 @@ func (r *reader) Next(ctx context.Context) (Message, error) {
// Acknowledge message immediately because the reader is based on non-durable subscription. When it reconnects,
// it will specify the subscription position anyway
msgID := cm.Message.ID()
if mid := toTrackingMessageID(msgID); mid != nil {
r.pc.lastDequeuedMsg = mid
r.pc.AckID(mid)
return cm.Message, nil
}
return nil, newError(InvalidMessage, fmt.Sprintf("invalid message id type %T", msgID))
mid := toTrackingMessageID(msgID)
r.pc.lastDequeuedMsg = mid
r.pc.AckID(mid)
return cm.Message, nil
case <-ctx.Done():
return nil, ctx.Err()
}
Expand Down Expand Up @@ -202,10 +199,6 @@ func (r *reader) Close() {

func (r *reader) messageID(msgID MessageID) *trackingMessageID {
mid := toTrackingMessageID(msgID)
if mid == nil {
r.log.Warnf("invalid message id type %T", msgID)
return nil
}

partition := int(mid.partitionIdx)
// did we receive a valid partition index?
Expand All @@ -221,6 +214,11 @@ func (r *reader) Seek(msgID MessageID) error {
r.Lock()
defer r.Unlock()

if !checkMessageIDType(msgID) {
r.log.Warnf("invalid message id type %T", msgID)
return fmt.Errorf("invalid message id type %T", msgID)
}

mid := r.messageID(msgID)
if mid == nil {
return nil
Expand Down