fix: add locking around broker throttle timer to prevent race condition (IBM#2826)

Fixes IBM#2823

Signed-off-by: shacheng <[email protected]>
Co-authored-by: shacheng <[email protected]>

Author: 2 people

broker.go

@@ -59,7 +59,8 @@ type Broker struct {
 	kerberosAuthenticator               GSSAPIKerberosAuth
 	clientSessionReauthenticationTimeMs int64
 
-	throttleTimer *time.Timer
+	throttleTimer     *time.Timer
+	throttleTimerLock sync.Mutex
 }
 
 // SASLMechanism specifies the SASL mechanism the client uses to authenticate with the broker
@@ -1694,6 +1695,8 @@ func (b *Broker) handleThrottledResponse(resp protocolBody) {
 }
 
 func (b *Broker) setThrottle(throttleTime time.Duration) {
+	b.throttleTimerLock.Lock()
+	defer b.throttleTimerLock.Unlock()
 	if b.throttleTimer != nil {
 		// if there is an existing timer stop/clear it
 		if !b.throttleTimer.Stop() {
@@ -1704,6 +1707,8 @@ func (b *Broker) setThrottle(throttleTime time.Duration) {
 }
 
 func (b *Broker) waitIfThrottled() {
+	b.throttleTimerLock.Lock()
+	defer b.throttleTimerLock.Unlock()
 	if b.throttleTimer != nil {
 		DebugLogger.Printf("broker/%d waiting for throttle timer\n", b.ID())
 		<-b.throttleTimer.C