Fixed deadlock in DLQ ack processing (#375)

apache · Oct 1, 2020 · 6f646aa · 6f646aa
1 parent 7dd67c9
commit 6f646aa
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
@@ -101,7 +101,11 @@ func (r *dlqRouter) run() {
 				ReplicationClusters: msg.replicationClusters,
 			}, func(MessageID, *ProducerMessage, error) {
 				r.log.WithField("msgID", msgID).Debug("Sent message to DLQ")
-				cm.Consumer.AckID(msgID)
+
+				// The Producer ack might be coming from the connection go-routine that
+				// is also used by the consumer. In that case we would get a dead-lock
+				// if we'd try to ack.
+				go cm.Consumer.AckID(msgID)
 			})
 
 		case <-r.closeCh: