Added measurement point to consumer profiling (#1869)

allegro · Jun 14, 2024 · a478894 · a478894
1 parent b594e7e
commit a478894
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 0 deletions.
diff --git a/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/ConsumerMessageSender.java b/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/ConsumerMessageSender.java
@@ -118,6 +118,7 @@ public void sendAsync(Message message, ConsumerProfiler profiler) {
     }
 
     private void sendAsync(Message message, int delayMillis, ConsumerProfiler profiler) {
+        profiler.measure(Measurement.SCHEDULE_MESSAGE_SENDING);
         retrySingleThreadExecutor.schedule(() -> sendMessage(message, profiler), delayMillis, TimeUnit.MILLISECONDS);
     }
 

diff --git a/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/profiling/Measurement.java b/...sumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/profiling/Measurement.java
@@ -8,6 +8,7 @@ public class Measurement {
     public static final String MESSAGE_CONVERSION = "messageConverter.convert";
     public static final String OFFER_INFLIGHT_OFFSET = "offsetQueue.offerInflightOffset";
     public static final String TRACKERS_LOG_INFLIGHT = "trackers.logInflight";
+    public static final String SCHEDULE_MESSAGE_SENDING = "retrySingleThreadExecutor.schedule";
     public static final String ACQUIRE_RATE_LIMITER = "acquireRateLimiter";
     public static final String MESSAGE_SENDER_SEND = "messageSender.send";
     public static final String HANDLERS = "handlers";

diff --git a/...c/integrationTest/java/pl/allegro/tech/hermes/integrationtests/ConsumerProfilingTest.java b/...c/integrationTest/java/pl/allegro/tech/hermes/integrationtests/ConsumerProfilingTest.java
@@ -148,6 +148,7 @@ public void shouldProfileSuccessfulMessageProcessing() {
                     Measurement.MESSAGE_CONVERSION,
                     Measurement.OFFER_INFLIGHT_OFFSET,
                     Measurement.TRACKERS_LOG_INFLIGHT,
+                    Measurement.SCHEDULE_MESSAGE_SENDING,
                     Measurement.ACQUIRE_RATE_LIMITER,
                     Measurement.MESSAGE_SENDER_SEND,
                     Measurement.HANDLERS,
@@ -183,6 +184,7 @@ public void shouldProfileDiscardedMessageProcessing() {
                     Measurement.MESSAGE_CONVERSION,
                     Measurement.OFFER_INFLIGHT_OFFSET,
                     Measurement.TRACKERS_LOG_INFLIGHT,
+                    Measurement.SCHEDULE_MESSAGE_SENDING,
                     Measurement.ACQUIRE_RATE_LIMITER,
                     Measurement.MESSAGE_SENDER_SEND,
                     Measurement.HANDLERS,
@@ -218,6 +220,7 @@ public void shouldProfileRetriedMessageProcessing() {
                     Measurement.MESSAGE_CONVERSION,
                     Measurement.OFFER_INFLIGHT_OFFSET,
                     Measurement.TRACKERS_LOG_INFLIGHT,
+                    Measurement.SCHEDULE_MESSAGE_SENDING,
                     Measurement.ACQUIRE_RATE_LIMITER,
                     Measurement.MESSAGE_SENDER_SEND,
                     Measurement.HANDLERS,