fixup

Signed-off-by: tison <wander4096@gmail.com>
apache · Oct 24, 2023 · dea241a · dea241a
1 parent 1f62b38
commit dea241a
Showing 1 changed file with 2 additions and 9 deletions.
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
@@ -1140,12 +1140,6 @@ func (p *partitionProducer) prepareTransaction(sr *sendRequest) error {
 	}
 
 	sr.transaction = txn
-	callback := sr.callback
-	sr.callback = func(id MessageID, producerMessage *ProducerMessage, err error) {
-		runCallback(callback, id, producerMessage, err)
-		txn.endSendOrAckOp(err)
-	}
-
 	return nil
 }
 
@@ -1177,13 +1171,12 @@ func (p *partitionProducer) internalSendAsync(
 		closeBlockChOnce: &sync.Once{},
 	}
 	if err := p.prepareTransaction(sr); err != nil {
-		runCallback(sr.callback, nil, msg, err)
+		sr.done(nil, err)
 		return
 	}
 
 	if p.getProducerState() != producerReady {
-		// Producer is closing
-		runCallback(sr.callback, nil, msg, errProducerClosed)
+		sr.done(nil, errProducerClosed)
 		return
 	}