(activemq-artemis) branch main updated: ARTEMIS-4576 ServerSessionImpl#updateProducerMetrics access large messages after being routed

[clebertsuconic]

This is an automated email from the ASF dual-hosted git repository.

clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git


The following commit(s) were added to refs/heads/main by this push:
     new 6f6c5723d5 ARTEMIS-4576 ServerSessionImpl#updateProducerMetrics access 
large messages after being routed
6f6c5723d5 is described below

commit 6f6c5723d50b179860319d380ab759ea1d65620a
Author: Howard Gao <howard....@gmail.com>
AuthorDate: Thu Jan 18 17:31:50 2024 +0800

    ARTEMIS-4576 ServerSessionImpl#updateProducerMetrics access large messages 
after being routed
---
 .../activemq/artemis/core/server/impl/ServerSessionImpl.java   | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index a180b0b842..3e3f268d64 100644
--- 
a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ 
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -2357,11 +2357,15 @@ public class ServerSessionImpl implements 
ServerSession, FailureListener {
          routingContext.setAddress(art.getName());
          routingContext.setRoutingType(art.getRoutingType());
 
+         // Retrieve message size for metrics update before routing,
+         // since large message backing files may be closed once routing 
completes
+         int mSize = msg instanceof LargeServerMessageImpl ? 
((LargeServerMessageImpl)msg).getBodyBufferSize() : msg.getEncodeSize();
+
          result = postOffice.route(msg, routingContext, direct);
 
          logger.debug("Routing result for {} = {}", msg, result);
 
-         updateProducerMetrics(msg, senderName);
+         updateProducerMetrics(msg, senderName, mSize);
       } finally {
          if (!routingContext.isReusable()) {
             routingContext.clear();
@@ -2520,10 +2524,10 @@ public class ServerSessionImpl implements 
ServerSession, FailureListener {
       return "ServerSession [id=" + getConnectionID() + ":" + getName() + "]";
    }
 
-   private void updateProducerMetrics(Message msg, String senderName) {
+   private void updateProducerMetrics(Message msg, String senderName, int 
mSize) {
       ServerProducer serverProducer = 
serverProducers.getServerProducer(senderName, msg, this);
       if (serverProducer != null) {
-         serverProducer.updateMetrics(msg.getUserID(), msg instanceof 
LargeServerMessageImpl ? ((LargeServerMessageImpl)msg).getBodyBufferSize() : 
msg.getEncodeSize());
+         serverProducer.updateMetrics(msg.getUserID(), mSize);
       }
    }