Polish support for new "requeued" count metric

Ensure backward compatibility, use "requeued" name, update copyright
year, add test.

Author: acogoluegnes

src/main/java/com/rabbitmq/client/MetricsCollector.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -58,9 +58,17 @@ default void basicPublishUnrouted(Channel channel) {
 
     void basicAck(Channel channel, long deliveryTag, boolean multiple);
 
-    void basicNack(Channel channel, long deliveryTag, boolean requeue);
+    void basicNack(Channel channel, long deliveryTag);
 
-    void basicReject(Channel channel, long deliveryTag, boolean requeue);
+    default void basicNack(Channel channel, long deliveryTag, boolean requeue) {
+        this.basicNack(channel, deliveryTag);
+    }
+
+    void basicReject(Channel channel, long deliveryTag);
+
+    default void basicReject(Channel channel, long deliveryTag, boolean requeue) {
+        this.basicReject(channel, deliveryTag);
+    }
 
     void basicConsume(Channel channel, String consumerTag, boolean autoAck);
 

src/main/java/com/rabbitmq/client/NoOpMetricsCollector.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -45,11 +45,21 @@ public void basicAck(Channel channel, long deliveryTag, boolean multiple) {
 
     }
 
+    @Override
+    public void basicNack(Channel channel, long deliveryTag) {
+
+    }
+
     @Override
     public void basicNack(Channel channel, long deliveryTag, boolean requeue) {
 
     }
 
+    @Override
+    public void basicReject(Channel channel, long deliveryTag) {
+
+    }
+
     @Override
     public void basicReject(Channel channel, long deliveryTag, boolean requeue) {
 

src/main/java/com/rabbitmq/client/impl/AbstractMetricsCollector.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -43,7 +43,7 @@ public abstract class AbstractMetricsCollector implements MetricsCollector {
 
     private final Runnable markAcknowledgedMessageAction = () -> markAcknowledgedMessage();
 
-    private final Function<Boolean, Runnable> markRejectedMessageAction = requeue -> () -> markRejectedMessage(requeue);
+    private final Function<Boolean, Runnable> markRejectedMessageAction;
 
     private final Runnable markMessagePublishAcknowledgedAction = () -> markMessagePublishAcknowledged();
 
@@ -53,6 +53,12 @@ public abstract class AbstractMetricsCollector implements MetricsCollector {
 
     private static final Function<ChannelState, Set<Long>> GET_UNCONFIRMED_DTAGS = channelState -> channelState.unconfirmedMessageDeliveryTags;
 
+    public AbstractMetricsCollector() {
+       Runnable rejectRequeue = () -> markRejectedMessage(true);
+       Runnable rejectNoRequeue = () -> markRejectedMessage(false);
+       this.markRejectedMessageAction = requeue -> requeue ? rejectRequeue : rejectNoRequeue;
+    }
+
     @Override
     public void newConnection(final Connection connection) {
         try {
@@ -235,6 +241,11 @@ public void basicAck(Channel channel, long deliveryTag, boolean multiple) {
         }
     }
 
+    @Override
+    public void basicNack(Channel channel, long deliveryTag) {
+        // replaced by #basicNack(Channel, long, boolean)
+    }
+
     @Override
     public void basicNack(Channel channel, long deliveryTag, boolean requeue) {
         try {
@@ -244,6 +255,11 @@ public void basicNack(Channel channel, long deliveryTag, boolean requeue) {
         }
     }
 
+    @Override
+    public void basicReject(Channel channel, long deliveryTag) {
+        // replaced by #basicReject(Channel, long, boolean)
+    }
+
     @Override
     public void basicReject(Channel channel, long deliveryTag, boolean requeue) {
         try {
@@ -406,10 +422,19 @@ private ChannelState(Channel channel) {
      */
     protected abstract void markAcknowledgedMessage();
 
+    /**
+     * Marks the event of a rejected message.
+     *
+     * @deprecated Use {@link #markRejectedMessage(boolean)} instead
+     */
+    protected abstract void markRejectedMessage();
+
     /**
      * Marks the event of a rejected message.
      */
-    protected abstract void markRejectedMessage(boolean requeue);
+    protected void markRejectedMessage(boolean requeue) {
+        this.markRejectedMessage();
+    }
 
     /**
      * Marks the event of a message publishing acknowledgement.

src/main/java/com/rabbitmq/client/impl/MicrometerMetricsCollector.java

@@ -63,7 +63,7 @@ public class MicrometerMetricsCollector extends AbstractMetricsCollector {
 
     private final Counter rejectedMessages;
 
-    private final Counter requeuedPublishedMessages;
+    private final Counter requeuedMessages;
 
     public MicrometerMetricsCollector(MeterRegistry registry) {
         this(registry, "rabbitmq");
@@ -92,7 +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);
+        this.requeuedMessages = (Counter) metricsCreator.apply(REQUEUED_MESSAGES);
     }
 
     @Override
@@ -135,10 +135,16 @@ protected void markAcknowledgedMessage() {
         acknowledgedMessages.increment();
     }
 
+    @Override
+    @SuppressWarnings("deprecation")
+    protected void markRejectedMessage() {
+
+    }
+
     @Override
     protected void markRejectedMessage(boolean requeue) {
         if (requeue) {
-            requeuedPublishedMessages.increment();
+            requeuedMessages.increment();
         }
         rejectedMessages.increment();
     }
@@ -198,6 +204,10 @@ public Counter getRejectedMessages() {
         return rejectedMessages;
     }
 
+    public Counter getRequeuedMessages() {
+        return requeuedMessages;
+    }
+
     public enum Metrics {
         CONNECTIONS {
             @Override
@@ -235,6 +245,12 @@ Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
                 return registry.counter(prefix + ".rejected", tags);
             }
         },
+        REQUEUED_MESSAGES {
+            @Override
+            Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
+                return registry.counter(prefix + ".requeued", tags);
+            }
+        },
         FAILED_TO_PUBLISH_MESSAGES {
             @Override
             Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags) {
@@ -258,12 +274,6 @@ 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);
-            }
         };
 
         abstract Object create(MeterRegistry registry, String prefix, Iterable<Tag> tags);

src/main/java/com/rabbitmq/client/impl/OpenTelemetryMetricsCollector.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2023-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -46,7 +46,7 @@ public class OpenTelemetryMetricsCollector extends AbstractMetricsCollector {
     private final LongCounter ackedPublishedMessagesCounter;
     private final LongCounter nackedPublishedMessagesCounter;
     private final LongCounter unroutedPublishedMessagesCounter;
-    private final LongCounter requeuedPublishedMessagesCounter;
+    private final LongCounter requeuedMessagesCounter;
 
     public OpenTelemetryMetricsCollector(OpenTelemetry openTelemetry) {
         this(openTelemetry, "rabbitmq");
@@ -101,6 +101,12 @@ public OpenTelemetryMetricsCollector(final OpenTelemetry openTelemetry, final St
             .setDescription("The number of messages rejected from the RabbitMQ server")
             .build();
 
+        // requeuedPublishedMessages
+        this.requeuedMessagesCounter = meter.counterBuilder(prefix + ".requeued")
+            .setUnit("{messages}")
+            .setDescription("The number of re-queued messages to the RabbitMQ server")
+            .build();
+
         // failedToPublishMessages
         this.failedToPublishMessagesCounter = meter.counterBuilder(prefix + ".failed_to_publish")
             .setUnit("{messages}")
@@ -124,12 +130,6 @@ 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
@@ -172,10 +172,16 @@ protected void markAcknowledgedMessage() {
         acknowledgedMessagesCounter.add(1L, attributes);
     }
 
+    @Override
+    @SuppressWarnings("deprecation")
+    protected void markRejectedMessage() {
+
+    }
+
     @Override
     protected void markRejectedMessage(boolean requeue) {
         if (requeue) {
-            requeuedPublishedMessagesCounter.add(1L, attributes);
+            requeuedMessagesCounter.add(1L, attributes);
         }
         rejectedMessagesCounter.add(1L, attributes);
     }

src/main/java/com/rabbitmq/client/impl/StandardMetricsCollector.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+// Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -41,12 +41,11 @@ public class StandardMetricsCollector extends AbstractMetricsCollector {
     private final Meter consumedMessages;
     private final Meter acknowledgedMessages;
     private final Meter rejectedMessages;
+    private final Meter requeuedMessages;
     private final Meter failedToPublishMessages;
     private final Meter publishAcknowledgedMessages;
     private final Meter publishNacknowledgedMessages;
     private final Meter publishUnroutedMessages;
-    private final Meter requeuedPublishedMessages;
-
 
     public StandardMetricsCollector(MetricRegistry registry, String metricsPrefix) {
         this.registry = registry;
@@ -60,7 +59,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");
+        this.requeuedMessages = registry.meter(metricsPrefix+".requeued");
     }
 
     public StandardMetricsCollector() {
@@ -111,8 +110,17 @@ protected void markAcknowledgedMessage() {
         acknowledgedMessages.mark();
     }
 
+    @Override
+    @SuppressWarnings("deprecation")
+    protected void markRejectedMessage() {
+
+    }
+
     @Override
     protected void markRejectedMessage(boolean requeue) {
+        if (requeue) {
+            requeuedMessages.mark();
+        }
         rejectedMessages.mark();
     }
 
@@ -159,6 +167,10 @@ public Meter getRejectedMessages() {
         return rejectedMessages;
     }
 
+    public Meter getRequeuedMessages() {
+        return this.requeuedMessages;
+    }
+
     public Meter getFailedToPublishMessages() {
         return failedToPublishMessages;
     }