Expose the number of unsettled messages in consumer

To make it easy to close a consumer without unsettled messages.

Author: acogoluegnes

src/docs/asciidoc/usage.adoc

@@ -93,7 +93,7 @@ include::{test-examples}/Api.java[tag=target-address-null]
 
 === Consuming
 
-.Consuming from a queue (generic body)
+.Consuming from a queue
 [source,java,indent=0]
 --------
 include::{test-examples}/Api.java[tag=consumer-consume]
@@ -102,6 +102,42 @@ include::{test-examples}/Api.java[tag=consumer-consume]
 <2> Process the message
 <3> Accept (acknowledge) the message
 
+A consumer can accept, discard, or requeue a message.
+We say the consumer _settles_ a message.
+
+Unsettled messages are requeued when a consumer get closed.
+This can lead to duplicate processing of messages.
+Here is an example:
+
+* A consumer executes a database operation for a given message.
+* The consumer gets closed before it accepts (settles) the message.
+* The message is requeued.
+* Another consumer gets the message and executes the database operation again.
+
+It is difficult to completely avoid duplicate messages, this is why processing should be idempotent.
+The consumer API allows nevertheless to _pause_ the delivery of messages, get the number of unsettled messages to make sure it reaches 0 at some point, and then _close_ the consumer.
+This ensures the consumer has finally quiesced and all the received messages have been processed.
+
+Here is an example of a consumer graceful shutdown:
+
+.Closing a consumer gracefully
+[source,java,indent=0]
+--------
+include::{test-examples}/Api.java[tag=consumer-graceful-shutdown]
+--------
+<1> Pause the delivery of messages
+<2> Ensure the number of unsettled messages reaches 0
+<3> Close the consumer
+
+It is also possible to simplify the consumer shutdown by just closing it, but this is likely to requeue unsettled messages.
+
+.Closing a consumer abruptly (can lead to duplicate messages)
+[source,java,indent=0]
+--------
+include::{test-examples}/Api.java[tag=consumer-abrupt-shutdown]
+--------
+<1> Close the consumer with potential unsettled messages
+
 === Resource Management
 
 The `Management` object is the entry point to deal with resources.

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

@@ -21,6 +21,8 @@ public interface Consumer extends AutoCloseable {
 
   void pause();
 
+  long unsettledCount();
+
   void unpause();
 
   @Override

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

@@ -122,6 +122,11 @@ public void unpause() {
     }
   }
 
+  @Override
+  public long unsettledCount() {
+    return unsettledCount.get();
+  }
+
   @Override
   public void close() {
     this.close(null);

src/test/java/com/rabbitmq/client/amqp/docs/Api.java

@@ -139,7 +139,7 @@ void targetAddressNull() {
   void consuming() {
     Connection connection = null;
     // tag::consumer-consume[]
-    connection.consumerBuilder()
+    Consumer consumer = connection.consumerBuilder()
         .queue("some-queue")
         .messageHandler((context, message) -> {
           byte[] body = message.body(); // <1>
@@ -148,6 +148,16 @@ void consuming() {
         })
         .build();
     // end::consumer-consume[]
+
+    // tag::consumer-graceful-shutdown[]
+    consumer.pause(); // <1>
+    long unsettledCount = consumer.unsettledCount(); // <2>
+    consumer.close(); // <3>
+    // end::consumer-graceful-shutdown[]
+
+    // tag::consumer-abrupt-shutdown[]
+    consumer.close(); // <1>
+    // end::consumer-abrupt-shutdown[]
   }
 
   void management() {

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

@@ -35,10 +35,7 @@
 import com.rabbitmq.client.amqp.impl.TestUtils.DisabledIfAddressV1Permitted;
 import com.rabbitmq.client.amqp.impl.TestUtils.Sync;
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -446,7 +443,7 @@ void consumerShouldGetClosedWhenQueueIsDeleted() {
   }
 
   @Test
-  void consumerShouldNotCloseUntilAllMessagesAreSettled() {
+  void consumerPauseThenClose() {
     connection.management().queue(name).exclusive(true).declare();
     int messageCount = 100;
     int initialCredits = messageCount / 10;
@@ -477,14 +474,82 @@ void consumerShouldNotCloseUntilAllMessagesAreSettled() {
     int unsettledCount = waitUntilStable(unsettledMessages::size);
     assertThat(unsettledCount).isNotZero();
     consumer.pause();
-    int receivedCountBeforePausing = receivedCount.get();
+    int receivedCountAfterPausing = receivedCount.get();
     unsettledMessages.forEach(com.rabbitmq.client.amqp.Consumer.Context::accept);
     consumer.close();
-    assertThat(receivedCount).hasValue(receivedCountBeforePausing);
+    assertThat(receivedCount).hasValue(receivedCountAfterPausing);
     assertThat(connection.management().queueInfo(name))
         .hasMessageCount(messageCount - receivedCount.get());
   }
 
+  @Test
+  void consumerGracefulShutdownExample() {
+    connection.management().queue(name).exclusive(true).declare();
+    int messageCount = 100;
+    int initialCredits = messageCount / 10;
+    Publisher publisher = connection.publisherBuilder().queue(name).build();
+    Sync publishSync = sync(messageCount);
+    range(0, messageCount)
+        .forEach(ignored -> publisher.publish(publisher.message(), ctx -> publishSync.down()));
+    assertThat(publishSync).completes();
+
+    AtomicInteger receivedCount = new AtomicInteger(0);
+    Random random = new Random();
+    com.rabbitmq.client.amqp.Consumer consumer =
+        connection
+            .consumerBuilder()
+            .queue(name)
+            .initialCredits(initialCredits)
+            .messageHandler(
+                (ctx, msg) -> {
+                  receivedCount.incrementAndGet();
+                  int processTime = random.nextInt(10) + 1;
+                  TestUtils.simulateActivity(processTime);
+                  ctx.accept();
+                })
+            .build();
+
+    waitAtMost(() -> receivedCount.get() > initialCredits * 2);
+    consumer.pause();
+    waitAtMost(() -> consumer.unsettledCount() == 0);
+    consumer.close();
+    assertThat(connection.management().queueInfo(name))
+        .hasMessageCount(messageCount - receivedCount.get());
+  }
+
+  @Test
+  void consumerUnsettledMessagesGoBackToQueueAfterClosing() {
+    connection.management().queue(name).exclusive(true).declare();
+    int messageCount = 100;
+    int initialCredits = messageCount / 10;
+    int settledCount = initialCredits * 2;
+    Publisher publisher = connection.publisherBuilder().queue(name).build();
+    Sync publishSync = sync(messageCount);
+    range(0, messageCount)
+        .forEach(ignored -> publisher.publish(publisher.message(), ctx -> publishSync.down()));
+    assertThat(publishSync).completes();
+
+    AtomicInteger receivedCount = new AtomicInteger(0);
+    com.rabbitmq.client.amqp.Consumer consumer =
+        connection
+            .consumerBuilder()
+            .queue(name)
+            .initialCredits(initialCredits)
+            .messageHandler(
+                (ctx, msg) -> {
+                  receivedCount.incrementAndGet();
+                  if (receivedCount.get() <= settledCount) {
+                    ctx.accept();
+                  }
+                })
+            .build();
+
+    waitAtMost(() -> receivedCount.get() > settledCount);
+    consumer.close();
+    assertThat(connection.management().queueInfo(name))
+        .hasMessageCount(messageCount - settledCount);
+  }
+
   static <T> T waitUntilStable(Supplier<T> call) {
     Duration timeout = Duration.ofSeconds(10);
     Duration waitTime = Duration.ofMillis(200);

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

@@ -129,6 +129,14 @@ public static Duration waitAtMost(
     }
   }
 
+  static void simulateActivity(long timeInMs) {
+    try {
+      Thread.sleep(timeInMs);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   public static class CountDownLatchReferenceConditions {
 
     public static Condition<AtomicReference<CountDownLatch>> completed() {