Description
Description
When the Kafka broker is throttled, the producer is easily blocked.
I find data race in setThrottle and waitIfThrottled.
Versions
| Sarama | Kafka | Go |
|---|---|---|
| v1.43.0 | 2.4.1 | 1.21.5 |
Configuration
Logs
WARNING: DATA RACE
Write at 0x00c000598f80 by goroutine 93:
github.com/IBM/sarama.(*Broker).setThrottle()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:1706 +0xe4
github.com/IBM/sarama.(*Broker).handleThrottledResponse()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:1695 +0x208
github.com/IBM/sarama.(*Broker).AsyncProduce.func1()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:467 +0x117
github.com/IBM/sarama.(*responsePromise).handle()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:140 +0x148
github.com/IBM/sarama.(*Broker).responseReceiver()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:1210 +0x584
github.com/IBM/sarama.(*Broker).responseReceiver-fm()
:1 +0x33
github.com/IBM/sarama.withRecover()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/utils.go:43 +0x41
github.com/IBM/sarama.(*Broker).Open.func1.5()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:258 +0x33
Previous read at 0x00c000598f80 by goroutine 77:
github.com/IBM/sarama.(*Broker).waitIfThrottled()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:1710 +0x44
github.com/IBM/sarama.(*Broker).sendInternal()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:1010 +0x307
github.com/IBM/sarama.(*Broker).sendWithPromise()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:994 +0xdd
github.com/IBM/sarama.(*Broker).AsyncProduce()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:473 +0x3a5
github.com/IBM/sarama.(*asyncProducer).newBrokerProducer.func1()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/async_producer.go:829 +0x264
github.com/IBM/sarama.withRecover()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/utils.go:43 +0x41
github.com/IBM/sarama.(*asyncProducer).newBrokerProducer.func4()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/async_producer.go:794 +0x33
Goroutine 93 (running) created at:
github.com/IBM/sarama.(*Broker).Open.func1()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:258 +0x1dea
github.com/IBM/sarama.withRecover()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/utils.go:43 +0x41
github.com/IBM/sarama.(*Broker).Open.func2()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/broker.go:184 +0x33
Goroutine 77 (running) created at:
github.com/IBM/sarama.(*asyncProducer).newBrokerProducer()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/async_producer.go:794 +0x690
github.com/IBM/sarama.(*asyncProducer).getBrokerProducer()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/async_producer.go:1343 +0x133
github.com/IBM/sarama.(*partitionProducer).dispatch()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/async_producer.go:634 +0x1a9
github.com/IBM/sarama.(*partitionProducer).dispatch-fm()
:1 +0x33
github.com/IBM/sarama.withRecover()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/utils.go:43 +0x41
github.com/IBM/sarama.(*asyncProducer).newPartitionProducer.func1()
/data/gopath/pkg/mod/github.com/!i!b!m/[email protected]/async_producer.go:600 +0x33
logs: CLICK ME
Additional Context
i can resove this bug by add lock
