Fix stop pending producer on multiple mode requests

2024-05-14 19:31:42 +03:00
parent a51156cf18
commit d9fb734c85
1 changed files with 5 additions and 0 deletions
@@ -88,6 +88,11 @@ func (s *Stream) RemoveProducer(prod core.Producer) {
 }

 func (s *Stream) stopProducers() {
+	if s.pending.Load() > 0 {
+		log.Trace().Msg("[streams] skip stop pending producer")
+		return
+	}
+
 	s.mu.Lock()
 producers:
 	for _, producer := range s.producers {