Fix flush lock keeps locked problem

Trendyol · Jul 21, 2023 · f5b2043 · f5b2043
1 parent b6a3184
commit f5b2043
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/kafka/producer/producer_batch.go b/kafka/producer/producer_batch.go
@@ -88,6 +88,8 @@ func (b *Batch) PrepareEndRebalancing() {
 func (b *Batch) AddMessages(ctx *models.ListenerContext, messages []kafka.Message, eventTime time.Time) {
 	b.flushLock.Lock()
 	if b.isDcpRebalancing {
+		b.errorLogger.Printf("could not add new message to batch while rebalancing")
+		b.flushLock.Unlock()
 		return
 	}
 	b.messages = append(b.messages, messages...)