Use AMQP outcomes in publisher callback

Author: acogoluegnes

src/docs/asciidoc/usage.adoc

@@ -132,7 +132,8 @@ The following metrics are recorded:
 * the number of open consumers
 * the total number of published messages
 * the total number of accepted published messages
-* the total number of failed published messages
+* the total number of rejected published messages
+* the total number of released published messages
 * the total number of consumed messages
 * the total number of accepted consumed messages
 * the total number of requeued consumed messages
@@ -167,9 +168,12 @@ rabbitmq_amqp_published_total 2.0
 # HELP rabbitmq_amqp_published_accepted_total
 # TYPE rabbitmq_amqp_published_accepted_total counter
 rabbitmq_amqp_published_accepted_total 1.0
-# HELP rabbitmq_amqp_published_failed_total
-# TYPE rabbitmq_amqp_published_failed_total counter
-rabbitmq_amqp_published_failed_total 1.0
+# HELP rabbitmq_amqp_published_rejected_total
+# TYPE rabbitmq_amqp_published_rejected_total counter
+rabbitmq_amqp_published_rejected_total 1.0
+# HELP rabbitmq_amqp_published_released_total
+# TYPE rabbitmq_amqp_published_released_total counter
+rabbitmq_amqp_published_released_total 1.0
 # HELP rabbitmq_amqp_publishers
 # TYPE rabbitmq_amqp_publishers gauge
 rabbitmq_amqp_publishers 1.0

src/main/java/com/rabbitmq/client/amqp/Publisher.java

@@ -102,7 +102,17 @@ interface Context {
   enum Status {
     /** The message has been accepted by the broker. */
     ACCEPTED,
-    /** The broker could not handle the message properly. */
-    FAILED
+    /**
+     * At least one queue the message was routed to rejected the message. This happens when the
+     * queue length is exceeded and the queue's overflow behaviour is set to reject-publish or when
+     * a target classic queue is unavailable.
+     */
+    REJECTED,
+    /**
+     * The broker could not route the message to any queue.
+     *
+     * <p>This is likely to be due to a topology misconfiguration.
+     */
+    RELEASED
   }
 }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpPublisher.java

@@ -113,21 +113,42 @@ public void publish(Message message, Callback callback) {
           try {
             // FIXME set a timeout for publishing settlement
             tracker.settlementFuture().get();
-            status =
-                tracker.remoteState() == DeliveryState.accepted() ? Status.ACCEPTED : Status.FAILED;
+            status = mapDeliveryState(tracker.remoteState());
           } catch (InterruptedException | ExecutionException e) {
-            status = Status.FAILED;
+            status = Status.REJECTED;
           }
           DefaultContext defaultContext = new DefaultContext(message, status);
-          this.metricsCollector.publishDisposition(
-              status == Status.ACCEPTED
-                  ? MetricsCollector.PublishDisposition.ACCEPTED
-                  : MetricsCollector.PublishDisposition.FAILED);
+          this.metricsCollector.publishDisposition(mapToPublishDisposition(status));
           callback.handle(defaultContext);
         });
     this.metricsCollector.publish();
   }
 
+  private Status mapDeliveryState(DeliveryState in) {
+    if (in.isAccepted()) {
+      return Status.ACCEPTED;
+    } else if (in.getType() == DeliveryState.Type.REJECTED) {
+      return Status.REJECTED;
+    } else if (in.getType() == DeliveryState.Type.RELEASED) {
+      return Status.RELEASED;
+    } else {
+      LOGGER.warn("Delivery state not supported: " + in.getType());
+      throw new IllegalStateException("This delivery state is not supported: " + in.getType());
+    }
+  }
+
+  private static MetricsCollector.PublishDisposition mapToPublishDisposition(Status status) {
+    if (status == Status.ACCEPTED) {
+      return MetricsCollector.PublishDisposition.ACCEPTED;
+    } else if (status == Status.REJECTED) {
+      return MetricsCollector.PublishDisposition.REJECTED;
+    } else if (status == Status.RELEASED) {
+      return MetricsCollector.PublishDisposition.RELEASED;
+    } else {
+      return null;
+    }
+  }
+
   void recoverAfterConnectionFailure() {
     this.connectionInfo = new Utils.ObservationConnectionInfo(this.connection.connectionAddress());
     this.sender =

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

@@ -41,7 +41,8 @@ public interface MetricsCollector {
 
   enum PublishDisposition {
     ACCEPTED,
-    FAILED
+    REJECTED,
+    RELEASED
   }
 
   enum ConsumeDisposition {

src/main/java/com/rabbitmq/client/amqp/metrics/MicrometerMetricsCollector.java

@@ -29,7 +29,7 @@ public class MicrometerMetricsCollector implements MetricsCollector {
   private final AtomicLong connections;
   private final AtomicLong publishers;
   private final AtomicLong consumers;
-  private final Counter publish, publishAccepted, publishFailed;
+  private final Counter publish, publishAccepted, publishRejected, publishReleased;
   private final Counter consume, consumeAccepted, consumeRequeued, consumeDiscarded;
 
   public MicrometerMetricsCollector(MeterRegistry registry) {
@@ -52,7 +52,8 @@ public MicrometerMetricsCollector(
     this.consumers = registry.gauge(prefix + ".consumers", tags, new AtomicLong(0));
     this.publish = registry.counter(prefix + ".published", tags);
     this.publishAccepted = registry.counter(prefix + ".published_accepted", tags);
-    this.publishFailed = registry.counter(prefix + ".published_failed", tags);
+    this.publishRejected = registry.counter(prefix + ".published_rejected", tags);
+    this.publishReleased = registry.counter(prefix + ".published_released", tags);
     this.consume = registry.counter(prefix + ".consumed", tags);
     this.consumeAccepted = registry.counter(prefix + ".consumed_accepted", tags);
     this.consumeRequeued = registry.counter(prefix + ".consumed_requeued", tags);
@@ -100,8 +101,11 @@ public void publishDisposition(PublishDisposition disposition) {
       case ACCEPTED:
         this.publishAccepted.increment();
         break;
-      case FAILED:
-        this.publishFailed.increment();
+      case REJECTED:
+        this.publishRejected.increment();
+        break;
+      case RELEASED:
+        this.publishReleased.increment();
         break;
       default:
         break;

src/main/qpid/org/apache/qpid/protonj2/client/impl/ClientDeliveryState.java

@@ -217,7 +217,7 @@ public ClientRejected(String condition, String description, Map<String, Object>
 
         @Override
         public Type getType() {
-            return Type.RELEASED;
+            return Type.REJECTED;
         }
 
         @Override

src/test/java/com/rabbitmq/client/amqp/impl/AddressFormatTest.java

@@ -47,7 +47,7 @@ void exchangeKeyInAddress(TestInfo info) {
       publisher.publish(
           publisher.message(),
           ctx -> {
-            if (ctx.status() == Publisher.Status.FAILED) {
+            if (ctx.status() == Publisher.Status.RELEASED) {
               failedLatch.countDown();
             }
           });
@@ -88,7 +88,7 @@ void exchangeInAddress(TestInfo info) {
       publisher.publish(
           publisher.message(),
           ctx -> {
-            if (ctx.status() == Publisher.Status.FAILED) {
+            if (ctx.status() == Publisher.Status.RELEASED) {
               failedLatch.countDown();
             }
           });
@@ -158,14 +158,14 @@ void exchangeKeyInToField(TestInfo info) {
       publisher.publish(
           publisher.message().toAddress().exchange(e).message(),
           ctx -> {
-            if (ctx.status() == Publisher.Status.FAILED) {
+            if (ctx.status() == Publisher.Status.RELEASED) {
               failedLatch.countDown();
             }
           });
       publisher.publish(
           publisher.message().toAddress().exchange(e).key("foo").message(),
           ctx -> {
-            if (ctx.status() == Publisher.Status.FAILED) {
+            if (ctx.status() == Publisher.Status.RELEASED) {
               failedLatch.countDown();
             }
           });

src/test/java/com/rabbitmq/client/amqp/impl/AmqpTest.java

@@ -667,6 +667,33 @@ void declareQueueWithUnsupportedArgument() {
         });
   }
 
+  @Test
+  void publishedMessageShouldBeRejectedWhenQueueLimitIsReached(TestInfo info) {
+    Management management = connection.management();
+    String q = TestUtils.name(info);
+    int maxLength = 10;
+    try {
+      management
+          .queue(q)
+          .maxLength(maxLength)
+          .overflowStrategy(Management.OverFlowStrategy.REJECT_PUBLISH)
+          .declare();
+      CountDownLatch rejectedLatch = new CountDownLatch(1);
+      Publisher.Callback callback =
+          context -> {
+            if (context.status() == Publisher.Status.REJECTED) {
+              rejectedLatch.countDown();
+            }
+          };
+      Publisher publisher = connection.publisherBuilder().queue(q).build();
+      IntStream.range(0, maxLength + 1)
+          .forEach(ignored -> publisher.publish(publisher.message(), callback));
+      Assertions.assertThat(rejectedLatch).completes();
+    } finally {
+      management.queueDeletion().delete(q);
+    }
+  }
+
   private static String uuid() {
     return UUID.randomUUID().toString();
   }

src/test/java/com/rabbitmq/client/amqp/impl/MetricsCollectorTest.java

@@ -18,7 +18,7 @@
 package com.rabbitmq.client.amqp.impl;
 
 import static com.rabbitmq.client.amqp.metrics.MetricsCollector.ConsumeDisposition.*;
-import static com.rabbitmq.client.amqp.metrics.MetricsCollector.PublishDisposition.FAILED;
+import static com.rabbitmq.client.amqp.metrics.MetricsCollector.PublishDisposition.RELEASED;
 import static java.lang.String.format;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
@@ -113,8 +113,8 @@ void metricsShouldBeCollected() {
           disposed(disposed));
       verify(metricsCollector, times(2)).publish();
       Assertions.assertThat(disposed).completes();
-      // the last message could not be routed, so its disposition state is failed
-      verify(metricsCollector, times(1)).publishDisposition(FAILED);
+      // the last message could not be routed, so its disposition state is RELEASED
+      verify(metricsCollector, times(1)).publishDisposition(RELEASED);
       verify(metricsCollector, times(2)).publishDisposition(any());
 
       verify(metricsCollector, never()).consume();

src/test/java/com/rabbitmq/client/amqp/impl/ResourceListenerTest.java

@@ -17,7 +17,7 @@
 // [email protected].
 package com.rabbitmq.client.amqp.impl;
 
-import static com.rabbitmq.client.amqp.Publisher.Status.FAILED;
+import static com.rabbitmq.client.amqp.Publisher.Status.RELEASED;
 import static org.assertj.core.api.Assertions.anyOf;
 import static org.assertj.core.api.Assertions.fail;
 
@@ -95,7 +95,7 @@ void publisherIsClosedOnExchangeDeletion(boolean toExchange, TestInfo info)
         .is(
             anyOf(
                 new Condition<>(s -> outboundMessageStatus.isEmpty(), "no status"),
-                new Condition<>(s -> outboundMessageStatus.contains(FAILED), "only failed")));
+                new Condition<>(s -> outboundMessageStatus.contains(RELEASED), "only released")));
     //    Assertions.assertThat(outboundMessageStatus).containsOnly(Publisher.Status.FAILED);
     Assertions.assertThat(closedCause.get()).isNotNull().isInstanceOf(AmqpException.class);
   }