diff --git a/pkg/stream/producer_unconfirmed.go b/pkg/stream/producer_unconfirmed.go
index 9ba1aef7..215b349a 100644
--- a/pkg/stream/producer_unconfirmed.go
+++ b/pkg/stream/producer_unconfirmed.go
@@ -137,7 +137,7 @@ func (u *unConfirmed) maybeUnLock() {
 	if len(u.messages) < u.maxSize {
 		logs.LogDebug("unConfirmed size: %d back to normal, producer unblocked", u.maxSize)
 		u.blockSignal.L.Lock()
-		u.blockSignal.Signal()
+		u.blockSignal.Broadcast()
 		u.blockSignal.L.Unlock()
 	}
 }