Distinguish nack metrics by requeue flag

ohksj77 · acogoluegnes · commit 0bfd4a61d84e · 2024-11-12T14:35:12.000+01:00
(cherry picked from commit 92b944d) Conflicts: src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java
diff --git a/src/main/java/com/rabbitmq/client/MetricsCollector.java b/src/main/java/com/rabbitmq/client/MetricsCollector.java
@@ -58,9 +58,9 @@ default void basicPublishUnrouted(Channel channel) {
 
     void basicAck(Channel channel, long deliveryTag, boolean multiple);
 
-    void basicNack(Channel channel, long deliveryTag);
+    void basicNack(Channel channel, long deliveryTag, boolean requeue);
 
-    void basicReject(Channel channel, long deliveryTag);
+    void basicReject(Channel channel, long deliveryTag, boolean requeue);
 
     void basicConsume(Channel channel, String consumerTag, boolean autoAck);
 
diff --git a/src/main/java/com/rabbitmq/client/NoOpMetricsCollector.java b/src/main/java/com/rabbitmq/client/NoOpMetricsCollector.java
@@ -46,12 +46,12 @@ public void basicAck(Channel channel, long deliveryTag, boolean multiple) {
     }
 
     @Override
-    public void basicNack(Channel channel, long deliveryTag) {
+    public void basicNack(Channel channel, long deliveryTag, boolean requeue) {
 
     }
 
     @Override
-    public void basicReject(Channel channel, long deliveryTag) {
+    public void basicReject(Channel channel, long deliveryTag, boolean requeue) {
 
     }
 
diff --git a/src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java b/src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java
@@ -42,7 +42,7 @@ public abstract class AbstractMetricsCollector implements MetricsCollector {
 
     private final Runnable markAcknowledgedMessageAction = () -> markAcknowledgedMessage();
 
-    private final Runnable markRejectedMessageAction = () -> markRejectedMessage();
+    private final Function<Boolean, Runnable> markRejectedMessageAction = requeue -> () -> markRejectedMessage(requeue);
 
     @Override
     public void newConnection(final Connection connection) {
@@ -226,18 +226,18 @@ public void basicAck(Channel channel, long deliveryTag, boolean multiple) {
     }
 
     @Override
-    public void basicNack(Channel channel, long deliveryTag) {
+    public void basicNack(Channel channel, long deliveryTag, boolean requeue) {
         try {
-            updateChannelStateAfterAckReject(channel, deliveryTag, true, markRejectedMessageAction);
+            updateChannelStateAfterAckReject(channel, deliveryTag, true, markRejectedMessageAction.apply(requeue));
         } catch(Exception e) {
             LOGGER.info("Error while computing metrics in basicNack: " + e.getMessage());
         }
     }
 
     @Override
-    public void basicReject(Channel channel, long deliveryTag) {
+    public void basicReject(Channel channel, long deliveryTag, boolean requeue) {
         try {
-            updateChannelStateAfterAckReject(channel, deliveryTag, false, markRejectedMessageAction);
+            updateChannelStateAfterAckReject(channel, deliveryTag, false, markRejectedMessageAction.apply(requeue));
         } catch(Exception e) {
             LOGGER.info("Error while computing metrics in basicReject: " + e.getMessage());
         }
@@ -395,7 +395,7 @@ private ChannelState(Channel channel) {
     /**
      * Marks the event of a rejected message.
      */
-    protected abstract void markRejectedMessage();
+    protected abstract void markRejectedMessage(boolean requeue);
 
     /**
      * Marks the event of a message publishing acknowledgement.
diff --git a/src/main/java/com/rabbitmq/client/impl/ChannelN.java b/src/main/java/com/rabbitmq/client/impl/ChannelN.java
@@ -1215,7 +1215,7 @@ public void basicNack(long deliveryTag, boolean multiple, boolean requeue)
         throws IOException
     {
         transmit(new Basic.Nack(deliveryTag, multiple, requeue));
-        metricsCollector.basicNack(this, deliveryTag);
+        metricsCollector.basicNack(this, deliveryTag, requeue);
     }
 
     /** Public API - {@inheritDoc} */
@@ -1224,7 +1224,7 @@ public void basicReject(long deliveryTag, boolean requeue)
         throws IOException
     {
         transmit(new Basic.Reject(deliveryTag, requeue));
-        metricsCollector.basicReject(this, deliveryTag);
+        metricsCollector.basicReject(this, deliveryTag, requeue);
     }
 
     /** Public API - {@inheritDoc} */
diff --git a/src/main/java/com/rabbitmq/client/impl/MicrometerMetricsCollector.java b/src/main/java/com/rabbitmq/client/impl/MicrometerMetricsCollector.java
@@ -63,6 +63,8 @@ public class MicrometerMetricsCollector extends AbstractMetricsCollector {
 
     private final Counter rejectedMessages;
 
+    private final Counter requeuedPublishedMessages;
+
     public MicrometerMetricsCollector(MeterRegistry registry) {
         this(registry, "rabbitmq");
     }
@@ -90,6 +92,7 @@ public MicrometerMetricsCollector(Function<Metrics, Object> metricsCreator) {
         this.ackedPublishedMessages = (Counter) metricsCreator.apply(ACKED_PUBLISHED_MESSAGES);
         this.nackedPublishedMessages = (Counter) metricsCreator.apply(NACKED_PUBLISHED_MESSAGES);
         this.unroutedPublishedMessages = (Counter) metricsCreator.apply(UNROUTED_PUBLISHED_MESSAGES);
+        this.requeuedPublishedMessages = (Counter) metricsCreator.apply(REQUEUED_PUBLISHED_MESSAGES);
     }
 
     @Override
@@ -133,7 +136,10 @@ protected void markAcknowledgedMessage() {
     }
 
     @Override
-    protected void markRejectedMessage() {
+    protected void markRejectedMessage(boolean requeue) {
+        if (requeue) {
+            requeuedPublishedMessages.increment();
+        }
         rejectedMessages.increment();
     }
 
@@ -252,6 +258,12 @@ Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
             Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
                 return registry.counter(prefix + ".unrouted_published", tags);
             }
+        },
+        REQUEUED_PUBLISHED_MESSAGES {
+            @Override
+            Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
+                return registry.counter(prefix + ".requeued_published", tags);
+            }
         };
 
         /**
diff --git a/src/main/java/com/rabbitmq/client/impl/OpenTelemetryMetricsCollector.java b/src/main/java/com/rabbitmq/client/impl/OpenTelemetryMetricsCollector.java
@@ -46,6 +46,7 @@ public class OpenTelemetryMetricsCollector extends AbstractMetricsCollector {
     private final LongCounter ackedPublishedMessagesCounter;
     private final LongCounter nackedPublishedMessagesCounter;
     private final LongCounter unroutedPublishedMessagesCounter;
+    private final LongCounter requeuedPublishedMessagesCounter;
 
     public OpenTelemetryMetricsCollector(OpenTelemetry openTelemetry) {
         this(openTelemetry, "rabbitmq");
@@ -123,6 +124,12 @@ public OpenTelemetryMetricsCollector(final OpenTelemetry openTelemetry, final St
             .setUnit("{messages}")
             .setDescription("The number of un-routed published messages to the RabbitMQ server")
             .build();
+
+        // requeuedPublishedMessages
+        this.requeuedPublishedMessagesCounter = meter.counterBuilder(prefix + ".requeued_published")
+                .setUnit("{messages}")
+                .setDescription("The number of re-queued published messages to the RabbitMQ server")
+                .build();
     }
 
     @Override
@@ -166,7 +173,10 @@ protected void markAcknowledgedMessage() {
     }
 
     @Override
-    protected void markRejectedMessage() {
+    protected void markRejectedMessage(boolean requeue) {
+        if (requeue) {
+            requeuedPublishedMessagesCounter.add(1L, attributes);
+        }
         rejectedMessagesCounter.add(1L, attributes);
     }
 
diff --git a/src/main/java/com/rabbitmq/client/impl/StandardMetricsCollector.java b/src/main/java/com/rabbitmq/client/impl/StandardMetricsCollector.java
@@ -45,6 +45,7 @@ public class StandardMetricsCollector extends AbstractMetricsCollector {
     private final Meter publishAcknowledgedMessages;
     private final Meter publishNacknowledgedMessages;
     private final Meter publishUnroutedMessages;
+    private final Meter requeuedPublishedMessages;
 
 
     public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {
@@ -59,6 +60,7 @@ public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {
         this.consumedMessages = registry.meter(metricsPrefix+".consumed");
         this.acknowledgedMessages = registry.meter(metricsPrefix+".acknowledged");
         this.rejectedMessages = registry.meter(metricsPrefix+".rejected");
+        this.requeuedPublishedMessages = registry.meter(metricsPrefix+".requeued_published");
     }
 
     public StandardMetricsCollector() {
@@ -110,7 +112,7 @@ protected void markAcknowledgedMessage() {
     }
 
     @Override
-    protected void markRejectedMessage() {
+    protected void markRejectedMessage(boolean requeue) {
         rejectedMessages.mark();
     }
 
diff --git a/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java b/src/main/java/com/rabbitmq/client/impl/recovery/RecoveryAwareChannelN.java
@@ -108,15 +108,15 @@ public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throw
             return;
         }
         transmit(new Basic.Nack(realTag, multiple, requeue));
-        metricsCollector.basicNack(this, deliveryTag);
+        metricsCollector.basicNack(this, deliveryTag, requeue);
     }
 
     @Override
     public void basicReject(long deliveryTag, boolean requeue) throws IOException {
         long realTag = deliveryTag - activeDeliveryTagOffset;
         if (realTag > 0) {
             transmit(new Basic.Reject(realTag, requeue));
-            metricsCollector.basicReject(this, deliveryTag);
+            metricsCollector.basicReject(this, deliveryTag, requeue);
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -46,12 +46,12 @@ public void basicAck(Channel channel, long deliveryTag, boolean multiple) {`
`46`	`46`	`}`
`47`	`47`
`48`	`48`	`@Override`
`49`		`- public void basicNack(Channel channel, long deliveryTag) {`
	`49`	`+ public void basicNack(Channel channel, long deliveryTag, boolean requeue) {`
`50`	`50`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`@Override`
`54`		`- public void basicReject(Channel channel, long deliveryTag) {`
	`54`	`+ public void basicReject(Channel channel, long deliveryTag, boolean requeue) {`
`55`	`55`
`56`	`56`	`}`
`57`	`57`
Original file line number	Diff line number	Diff line change
`@@ -1215,7 +1215,7 @@ public void basicNack(long deliveryTag, boolean multiple, boolean requeue)`
`1215`	`1215`	`throws IOException`
`1216`	`1216`	`{`
`1217`	`1217`	`transmit(new Basic.Nack(deliveryTag, multiple, requeue));`
`1218`		`- metricsCollector.basicNack(this, deliveryTag);`
	`1218`	`+ metricsCollector.basicNack(this, deliveryTag, requeue);`
`1219`	`1219`	`}`
`1220`	`1220`
`1221`	`1221`	`/** Public API - {@inheritDoc} */`
`@@ -1224,7 +1224,7 @@ public void basicReject(long deliveryTag, boolean requeue)`
`1224`	`1224`	`throws IOException`
`1225`	`1225`	`{`
`1226`	`1226`	`transmit(new Basic.Reject(deliveryTag, requeue));`
`1227`		`- metricsCollector.basicReject(this, deliveryTag);`
	`1227`	`+ metricsCollector.basicReject(this, deliveryTag, requeue);`
`1228`	`1228`	`}`
`1229`	`1229`
`1230`	`1230`	`/** Public API - {@inheritDoc} */`
Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,7 @@ public class StandardMetricsCollector extends AbstractMetricsCollector {`
`45`	`45`	`private final Meter publishAcknowledgedMessages;`
`46`	`46`	`private final Meter publishNacknowledgedMessages;`
`47`	`47`	`private final Meter publishUnroutedMessages;`
	`48`	`+ private final Meter requeuedPublishedMessages;`
`48`	`49`
`49`	`50`
`50`	`51`	`public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {`
`@@ -59,6 +60,7 @@ public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {`
`59`	`60`	`this.consumedMessages = registry.meter(metricsPrefix+".consumed");`
`60`	`61`	`this.acknowledgedMessages = registry.meter(metricsPrefix+".acknowledged");`
`61`	`62`	`this.rejectedMessages = registry.meter(metricsPrefix+".rejected");`
	`63`	`+ this.requeuedPublishedMessages = registry.meter(metricsPrefix+".requeued_published");`
`62`	`64`	`}`
`63`	`65`
`64`	`66`	`public StandardMetricsCollector() {`
`@@ -110,7 +112,7 @@ protected void markAcknowledgedMessage() {`
`110`	`112`	`}`
`111`	`113`
`112`	`114`	`@Override`
`113`		`- protected void markRejectedMessage() {`
	`115`	`+ protected void markRejectedMessage(boolean requeue) {`
`114`	`116`	`rejectedMessages.mark();`
`115`	`117`	`}`
`116`	`118`
Original file line number	Diff line number	Diff line change
`@@ -108,15 +108,15 @@ public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throw`
`108`	`108`	`return;`
`109`	`109`	`}`
`110`	`110`	`transmit(new Basic.Nack(realTag, multiple, requeue));`
`111`		`- metricsCollector.basicNack(this, deliveryTag);`
	`111`	`+ metricsCollector.basicNack(this, deliveryTag, requeue);`
`112`	`112`	`}`
`113`	`113`
`114`	`114`	`@Override`
`115`	`115`	`public void basicReject(long deliveryTag, boolean requeue) throws IOException {`
`116`	`116`	`long realTag = deliveryTag - activeDeliveryTagOffset;`
`117`	`117`	`if (realTag > 0) {`
`118`	`118`	`transmit(new Basic.Reject(realTag, requeue));`
`119`		`- metricsCollector.basicReject(this, deliveryTag);`
	`119`	`+ metricsCollector.basicReject(this, deliveryTag, requeue);`
`120`	`120`	`}`
`121`	`121`	`}`
`122`	`122`