Refine producer retryOnRecovery support

References #642

Author: acogoluegnes

src/docs/asciidoc/api.adoc

@@ -467,8 +467,8 @@ Set the value to `Duration.ZERO` if there should be no timeout.
 |10 seconds
 
 |`retryOnRecovery`
-|Re-publish unconfirmed messages after recovering a connection.
-Set to false if do not want to re-publish unconfirmed messages after recovering a connection.
+|Whether to republish unconfirmed messages after recovery.
+Set to `false` to not republish unconfirmed messages and get a negative `ConfirmationStatus` for unconfirmed messages.
 |true
 |===
 

src/main/java/com/rabbitmq/stream/ProducerBuilder.java

@@ -134,13 +134,17 @@ public interface ProducerBuilder {
   ProducerBuilder enqueueTimeout(Duration timeout);
 
   /**
-   * Re-publish unconfirmed messages after recovering a connection.
+   * Whether to republish unconfirmed messages after recovery.
    *
-   * <p>Default is true.</p>
+   * <p>Default is <code>true</code> (unconfirmed messages are republished after recovery).
    *
-   * <p>Set to false if do not want to re-publish unconfirmed messages after recovering a connection.</p>
+   * <p>Set to <code>false</code> to not republish unconfirmed messages and get a negative {@link
+   * ConfirmationStatus} for unconfirmed messages.
    *
-   * @param retryOnRecovery
+   * <p>Note setting this flag to <code>false</code> translates to at-most-once semantics, that is
+   * published messages may be lost, unless the publishing application retries publishing them.
+   *
+   * @param retryOnRecovery retry flag
    * @return this builder instance
    */
   ProducerBuilder retryOnRecovery(boolean retryOnRecovery);

src/main/java/com/rabbitmq/stream/impl/StreamProducer.java

@@ -507,34 +507,23 @@ void unavailable() {
 
   void running() {
     synchronized (this) {
-      if (!this.retryOnRecovery) {
-        LOGGER.debug(
-            "Skip to republish {} unconfirmed message(s) and re-publishing {} accumulated message(s)",
-            this.unconfirmedMessages.size(),
-            this.accumulator.size());
-
-        this.unconfirmedMessages.clear();
-        int toRelease = maxUnconfirmedMessages - unconfirmedMessagesSemaphore.availablePermits();
-        if (toRelease > 0) {
-          unconfirmedMessagesSemaphore.release(toRelease);
-        }
-
-        publishBatch(false);
-      } else {
-        LOGGER.debug(
-            "Re-publishing {} unconfirmed message(s) and {} accumulated message(s)",
-            this.unconfirmedMessages.size(),
-            this.accumulator.size());
+      LOGGER.debug(
+          "Recovering producer with {} unconfirmed message(s) and {} accumulated message(s)",
+          this.unconfirmedMessages.size(),
+          this.accumulator.size());
+      if (this.retryOnRecovery) {
+        LOGGER.debug("Re-publishing {} unconfirmed message(s)", this.unconfirmedMessages.size());
         if (!this.unconfirmedMessages.isEmpty()) {
           Map<Long, AccumulatedEntity> messagesToResend = new TreeMap<>(this.unconfirmedMessages);
           this.unconfirmedMessages.clear();
           Iterator<Entry<Long, AccumulatedEntity>> resendIterator =
-            messagesToResend.entrySet().iterator();
+              messagesToResend.entrySet().iterator();
           while (resendIterator.hasNext()) {
             List<Object> messages = new ArrayList<>(this.batchSize);
             int batchCount = 0;
             while (batchCount != this.batchSize) {
-              Object accMessage = resendIterator.hasNext() ? resendIterator.next().getValue() : null;
+              Object accMessage =
+                  resendIterator.hasNext() ? resendIterator.next().getValue() : null;
               if (accMessage == null) {
                 break;
               }
@@ -549,18 +538,34 @@ void running() {
                 this.publishSequenceFunction);
           }
         }
-        publishBatch(false);
-
-        int toRelease = maxUnconfirmedMessages - unconfirmedMessagesSemaphore.availablePermits();
-        if (toRelease > 0) {
-          unconfirmedMessagesSemaphore.release(toRelease);
-          if (!unconfirmedMessagesSemaphore.tryAcquire(this.unconfirmedMessages.size())) {
-            LOGGER.debug(
-                "Could not acquire {} permit(s) for message republishing",
-                this.unconfirmedMessages.size());
+      } else {
+        LOGGER.debug(
+            "Skipping republishing of {} unconfirmed messages", this.unconfirmedMessages.size());
+        Map<Long, AccumulatedEntity> messagesToFail = new TreeMap<>(this.unconfirmedMessages);
+        this.unconfirmedMessages.clear();
+        for (AccumulatedEntity accumulatedEntity : messagesToFail.values()) {
+          try {
+            int permits =
+                accumulatedEntity
+                    .confirmationCallback()
+                    .handle(false, CODE_PUBLISH_CONFIRM_TIMEOUT);
+            this.unconfirmedMessagesSemaphore.release(permits);
+          } catch (Exception e) {
+            LOGGER.debug("Error while nack-ing outbound message: {}", e.getMessage());
+            this.unconfirmedMessagesSemaphore.release(1);
           }
         }
       }
+      publishBatch(false);
+      int toRelease = maxUnconfirmedMessages - unconfirmedMessagesSemaphore.availablePermits();
+      if (toRelease > 0) {
+        unconfirmedMessagesSemaphore.release(toRelease);
+        if (!unconfirmedMessagesSemaphore.tryAcquire(this.unconfirmedMessages.size())) {
+          LOGGER.debug(
+              "Could not acquire {} permit(s) for message republishing",
+              this.unconfirmedMessages.size());
+        }
+      }
     }
     this.status = Status.RUNNING;
   }

src/test/java/com/rabbitmq/stream/impl/ProducersCoordinatorTest.java

@@ -37,6 +37,7 @@
 import com.rabbitmq.stream.StreamDoesNotExistException;
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Utils.ClientFactory;
+import io.netty.channel.ConnectTimeoutException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -46,8 +47,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.IntStream;
-
-import io.netty.channel.ConnectTimeoutException;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -309,8 +308,7 @@ void shouldRecoverOnConnectionTimeout() throws Exception {
     when(environment.scheduledExecutorService()).thenReturn(scheduledExecutorService);
     Duration retryDelay = Duration.ofMillis(50);
     when(environment.recoveryBackOffDelayPolicy()).thenReturn(BackOffDelayPolicy.fixed(retryDelay));
-    when(locator.metadata("stream"))
-        .thenReturn(metadata(leader(), replicas()));
+    when(locator.metadata("stream")).thenReturn(metadata(leader(), replicas()));
 
     when(clientFactory.client(any()))
         .thenReturn(client)

src/test/java/com/rabbitmq/stream/impl/StreamProducerTest.java

@@ -243,7 +243,7 @@ void sendToNonExistingStreamShouldReturnUnconfirmedStatus() throws Exception {
   @TestUtils.DisabledIfRabbitMqCtlNotSet
   void shouldRecoverAfterConnectionIsKilled(int subEntrySize) throws Exception {
     Producer producer =
-        environment.producerBuilder().subEntrySize(subEntrySize).retryOnRecovery(true).stream(stream).build();
+        environment.producerBuilder().subEntrySize(subEntrySize).stream(stream).build();
 
     AtomicInteger published = new AtomicInteger(0);
     AtomicInteger confirmed = new AtomicInteger(0);