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

fixed panic on calling updateMetadata on closed client #1531

Merged
merged 3 commits into from
Nov 12, 2019
Merged

fixed panic on calling updateMetadata on closed client #1531

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a5a3dd2
fixed panic on calling updateMetadata on closed client
roblaszczak Nov 1, 2019
93f4001
fixed race conditions and nil panic on calling registerBroker after c…
roblaszczak Nov 2, 2019
e56a4a9
added comments for handleError
roblaszczak Nov 12, 2019
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
12 changes: 12 additions & 0 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,9 @@ func (client *client) Close() error {
}

func (client *client) Closed() bool {
client.lock.RLock()
defer client.lock.RUnlock()

return client.brokers == nil
}

Expand Down Expand Up @@ -529,6 +532,11 @@ func (client *client) RefreshCoordinator(consumerGroup string) error {
// in the brokers map. It returns the broker that is registered, which may be the provided broker,
// or a previously registered Broker instance. You must hold the write lock before calling this function.
func (client *client) registerBroker(broker *Broker) {
if client.brokers == nil {
Copy link

Choose a reason for hiding this comment

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

this is a race condition, no?

It checks for client.brokers != nil
and then uses client.brokers

it could have been set to nil between the if client.brokers == nil and if client.brokers[broker.ID()] == nil

Logger.Printf("cannot register broker #%d at %s, client already closed", broker.ID(), broker.Addr())
return
}

if client.brokers[broker.ID()] == nil {
client.brokers[broker.ID()] = broker
Logger.Printf("client/brokers registered new broker #%d at %s", broker.ID(), broker.Addr())
Expand Down Expand Up @@ -833,6 +841,10 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int,

// if no fatal error, returns a list of topics that need retrying due to ErrLeaderNotAvailable
func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bool) (retry bool, err error) {
if client.Closed() {
return
}

client.lock.Lock()
defer client.lock.Unlock()

Expand Down
30 changes: 18 additions & 12 deletions consumer_group.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -417,12 +417,6 @@ func (c *consumerGroup) leave() error {
}

func (c *consumerGroup) handleError(err error, topic string, partition int32) {
select {
case <-c.closed:
return
default:
}

if _, ok := err.(*ConsumerError); !ok && topic != "" && partition > -1 {
err = &ConsumerError{
Topic: topic,
Expand All @@ -431,13 +425,25 @@ func (c *consumerGroup) handleError(err error, topic string, partition int32) {
}
}

if c.config.Consumer.Return.Errors {
select {
case c.errors <- err:
default:
}
} else {
if !c.config.Consumer.Return.Errors {
Logger.Println(err)
return
}

c.lock.Lock()
defer c.lock.Unlock()

select {
case <-c.closed:
//consumer is closed
return
default:
}

select {
case c.errors <- err:
default:
// no error listener
}
}

Expand Down