Skip to content

[feature] expose error to user when producer reconnect to broker failed #648

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
yorkhellen wants to merge 13 commits into
base: master
Choose a base branch
from
Open

[feature] expose error to user when producer reconnect to broker failed #648

Changes from 5 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
3c4bf04
[feature] expose error to user when producer reconnectToBroker
Oct 25, 2021
2e8408a
fix code style
Nov 16, 2021
4ac03e5
add comment when internal reconnect failed and fix code style
Nov 16, 2021
2c2656e
Merge branch 'apache:master' into master
yorkhellen Nov 17, 2021
e090d27
remove metadataError check
Nov 17, 2021
8313763
Merge branch 'apache:master' into master
yorkhellen Dec 25, 2021
8aeaa4b
Merge branch 'apache:master' into master
yorkhellen Dec 27, 2021
d7678ad
merge upstream
Jan 17, 2022
da7dcb5
Merge branch 'apache-master'
Jan 17, 2022
943bc03
Merge branch 'apache:master' into master
billowqiu Jan 17, 2022
5716f38
Merge branch 'apache:master' into master
yorkhellen Jun 30, 2022
388407d
Merge branch 'apache:master' into master
yorkhellen Mar 21, 2023
b33f371
Merge branch 'apache:master' into master
yorkhellen Apr 1, 2024
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
9 changes: 9 additions & 0 deletions pulsar/producer_partition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@ type partitionProducer struct {
topic string
log log.Logger
cnx internal.Connection
err error
Copy link
Contributor

@cckellogg cckellogg Oct 25, 2021

Choose a reason for hiding this comment

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

Why is this needed?

Copy link
Member

@wolfstudy wolfstudy Oct 27, 2021

Choose a reason for hiding this comment

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

hello @cckellogg , It seems that some of our Error messages are not fully exposed, such as sending failures after reaching the Backlog threshold. It seems that @yorkhellen is encapsulating their business based on the Go SDK, and may need to process specific business logic based on these specific errors. It seems that it is OK to expose the error field to the user here. This way the business can be more flexible in error handling.

Copy link
Contributor

@cckellogg cckellogg Oct 28, 2021

Choose a reason for hiding this comment

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

Can you provide some examples on the use case and or what the business logic will do with these errors? I'm just trying to understand better. Also, I think this should not be part of the 0.7.0 release so we can think about it more. If we are going to bubble some more errors up we should come up with a good api.

Copy link
Member

@wolfstudy wolfstudy Nov 1, 2021

Choose a reason for hiding this comment

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

Sure, I will provide a code example later to illustrate this matter.

Yes, we always include this change in 0.7.0


options *ProducerOptions
producerName string
Expand Down Expand Up @@ -350,6 +351,10 @@ func (p *partitionProducer) reconnectToBroker() {
time.Sleep(d)
atomic.AddUint64(&p.epoch, 1)
err := p.grabCnx()
// In reconnection logic, grabCnx maybe return err, but we did not return the error.
// So in partitionProducer struct, we define an err object to make it easier for users to
// determine what caused the grabCnx error.
p.err = err
if err == nil {
// Successfully reconnected
p.log.WithField("cnx", p.cnx.ID()).Info("Reconnected producer to broker")
Expand Down Expand Up @@ -742,6 +747,10 @@ func (p *partitionProducer) internalSendAsync(ctx context.Context, msg *Producer
callback(nil, msg, errProducerClosed)
return
}
if p.err != nil {
Copy link
Contributor

@cckellogg cckellogg Nov 17, 2021 •
edited
Loading

Choose a reason for hiding this comment

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

This now changes the current behavior. Before if there was room the message would be added to the pending queue and sent when a new connection is established. Now the message will not be added. I'm not sure changing the default behavior is the correct approach here. Also, there could be a race condition checking the error and sending it since the the error is set in another go routine?

How do you envision your application handling this error?

callback(nil, msg, p.err)
return
}

sr := &sendRequest{
ctx: ctx,
Expand Down