Skip to content

fix 640,change create way is same as java client #641

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

Open
baomingyu wants to merge 10 commits into
base: master
Choose a base branch
from
Open

fix 640,change create way is same as java client #641

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
e9c219c
fix 640,change create way is same as java client
baomingyu Oct 14, 2021
c7a3467
fix 640,handle test error
baomingyu Oct 14, 2021
84de13c
add handler different err
baomingyu Oct 15, 2021
5085a17
modify test error
baomingyu Oct 15, 2021
d5cf4b6
recover code whit some advice
baomingyu Oct 15, 2021
8df4e6d
only handle server not already and too many request lookup error
baomingyu Oct 15, 2021
94c2db5
modify log text
baomingyu Oct 15, 2021
d018309
check code style
baomingyu Oct 15, 2021
9d864c2
add handle metadataError
baomingyu Oct 15, 2021
9fd5a9e
add handle metadataError
baomingyu Oct 15, 2021
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
41 changes: 24 additions & 17 deletions pulsar/consumer_partition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,33 +205,40 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon

err := pc.grabConn()
if err != nil {
pc.log.WithError(err).Error("Failed to create consumer")
pc.nackTracker.Close()
return nil, err
}
pc.log.Info("Created consumer")
pc.setConsumerState(consumerReady)

if pc.options.startMessageIDInclusive && pc.startMessageID.equal(lastestMessageID.(messageID)) {
msgID, err := pc.requestGetLastMessageID()
if err != nil {
errMsg := err.Error()
if !strings.EqualFold(errMsg, pb.ServerError_ServiceNotReady.String()) &&
!strings.EqualFold(errMsg, pb.ServerError_TooManyRequests.String()) &&
!strings.EqualFold(errMsg, pb.ServerError_MetadataError.String()) {
// when topic is deleted, we should give up reconnection.
pc.log.WithError(err).Error("Failed to create consumer")
pc.nackTracker.Close()
return nil, err
}
if msgID.entryID != noMessageEntry {
pc.startMessageID = msgID
pc.log.WithError(err).Error("Failed to create consumer, it will be retried later!")
pc.nackTracker.Close()
} else {
pc.log.Info("Created consumer")
pc.setConsumerState(consumerReady)

// use the WithoutClear version because the dispatcher is not started yet
err = pc.requestSeekWithoutClear(msgID.messageID)
if pc.options.startMessageIDInclusive && pc.startMessageID.equal(lastestMessageID.(messageID)) {
msgID, err := pc.requestGetLastMessageID()
if err != nil {
pc.nackTracker.Close()
return nil, err
}
if msgID.entryID != noMessageEntry {
pc.startMessageID = msgID

// use the WithoutClear version because the dispatcher is not started yet
err = pc.requestSeekWithoutClear(msgID.messageID)
if err != nil {
pc.nackTracker.Close()
return nil, err
}
}
}
go pc.dispatcher()
}

go pc.dispatcher()

go pc.runEventsLoop()

return pc, nil
Expand Down
32 changes: 19 additions & 13 deletions pulsar/producer_partition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,22 +160,28 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions

err := p.grabCnx()
if err != nil {
logger.WithError(err).Error("Failed to create producer")
return nil, err
}

p.log = p.log.SubLogger(log.Fields{
"producer_name": p.producerName,
"producerID": p.producerID,
})
errMsg := err.Error()
if !strings.EqualFold(errMsg, pb.ServerError_ServiceNotReady.String()) &&
!strings.EqualFold(errMsg, pb.ServerError_TooManyRequests.String()) &&
!strings.EqualFold(errMsg, pb.ServerError_MetadataError.String()) {
// when topic is deleted, we should give up reconnection.
logger.WithError(err).Error("Failed to create producer")
return nil, err
}
logger.WithError(err).Error("Failed to create producer, it will be retried later!")
} else {
p.log = p.log.SubLogger(log.Fields{
"producer_name": p.producerName,
"producerID": p.producerID,
})

p.log.WithField("cnx", p.cnx.ID()).Info("Created producer")
p.setProducerState(producerReady)
p.log.WithField("cnx", p.cnx.ID()).Info("Created producer")
p.setProducerState(producerReady)

if p.options.SendTimeout > 0 {
go p.failTimeoutMessages()
if p.options.SendTimeout > 0 {
go p.failTimeoutMessages()
}
}

go p.runEventsLoop()

return p, nil
Expand Down