[fix][broker] Record GeoPersistentReplicator.msgOut before producer#s…

…endAsync (#21673) Signed-off-by: Zixuan Liu <nodeces@gmail.com>
apache · Dec 8, 2023 · c4196fb · c4196fb
1 parent 2393ca7
commit c4196fb
Showing 1 changed file with 1 addition and 3 deletions.
diff --git a/...er/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/...er/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java
@@ -149,9 +149,6 @@ protected boolean replicateEntries(List<Entry> entries) {
                 }
 
                 dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(1, entry.getLength()));
-
-                msgOut.recordEvent(headersAndPayload.readableBytes());
-
                 msg.setReplicatedFrom(localCluster);
 
                 headersAndPayload.retain();
@@ -181,6 +178,7 @@ protected boolean replicateEntries(List<Entry> entries) {
                     msg.setSchemaInfoForReplicator(schemaFuture.get());
                     msg.getMessageBuilder().clearTxnidMostBits();
                     msg.getMessageBuilder().clearTxnidLeastBits();
+                    msgOut.recordEvent(headersAndPayload.readableBytes());
                     // Increment pending messages for messages produced locally
                     PENDING_MESSAGES_UPDATER.incrementAndGet(this);
                     producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg));