fix Send callback stuck

billowqiu · Oct 17, 2022 · fc0a56d · fc0a56d
1 parent 858b1e9
commit fc0a56d
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
@@ -742,8 +742,15 @@ func (p *partitionProducer) Send(ctx context.Context, msg *ProducerMessage) (Mes
 	}, true)
 
 	// wait for send request to finish
-	<-doneCh
-	return msgID, err
+	for {
+		select {
+		case <-ctx.Done():
+			isDone.Store(true)
+			return msgID, ctx.Err()
+		case <-doneCh:
+			return msgID, err
+		}
+	}
 }
 
 func (p *partitionProducer) SendAsync(ctx context.Context, msg *ProducerMessage,