Add single active consumer API to Consumer

WIP.

References rabbitmq/rabbitmq-server#3753

Author: acogoluegnes

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

@@ -64,6 +64,10 @@ public interface ConsumerBuilder {
    */
   ConsumerBuilder name(String name);
 
+  ConsumerBuilder singleActiveConsumer();
+
+  ConsumerBuilder consumerUpdateListener(ConsumerUpdateListener consumerUpdateListener);
+
   /**
    * Callback on subscription.
    *

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

@@ -0,0 +1,34 @@
+// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.stream;
+
+public interface ConsumerUpdateListener {
+
+  OffsetSpecification update(Context context);
+
+  interface Context {
+
+    Consumer consumer();
+
+    Status status();
+
+    Status previousStatus();
+  }
+
+  enum Status {
+    STARTING,
+    ACTIVE,
+    PASSIVE
+  }
+}

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

@@ -24,9 +24,9 @@ public interface SubscriptionListener {
   /**
    * Callback called before the subscription is created.
    *
-   * <p>The method is called when a {@link Consumer} is created and it registers to broker, and also
-   * when the subscription must be re-created (after a disconnection or when the subscription must
-   * moved because the stream member it was connection becomes unavailable).
+   * <p>The method is called when a {@link Consumer} is created and it registers to the broker, and
+   * also when the subscription must be re-created (after a disconnection or when the subscription
+   * must moved because the stream member it was connected to becomes unavailable).
    *
    * <p>Application code can set the {@link OffsetSpecification} that will be used with the {@link
    * SubscriptionContext#offsetSpecification(OffsetSpecification)} method.

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

@@ -121,8 +121,8 @@ AsyncRetryBuilder<V> retry(Predicate<Exception> predicate) {
       return this;
     }
 
-    AsyncRetryBuilder<V> description(String description) {
-      this.description = description;
+    AsyncRetryBuilder<V> description(String description, Object... args) {
+      this.description = String.format(description, args);
       return this;
     }
 

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

@@ -14,6 +14,7 @@
 package com.rabbitmq.stream.impl;
 
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
+import static com.rabbitmq.stream.impl.Utils.isSac;
 
 import com.rabbitmq.stream.BackOffDelayPolicy;
 import com.rabbitmq.stream.Constants;
@@ -26,6 +27,7 @@
 import com.rabbitmq.stream.SubscriptionListener;
 import com.rabbitmq.stream.SubscriptionListener.SubscriptionContext;
 import com.rabbitmq.stream.impl.Client.ChunkListener;
+import com.rabbitmq.stream.impl.Client.ConsumerUpdateListener;
 import com.rabbitmq.stream.impl.Client.CreditNotification;
 import com.rabbitmq.stream.impl.Client.MessageListener;
 import com.rabbitmq.stream.impl.Client.MetadataListener;
@@ -35,7 +37,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -86,7 +87,8 @@ Runnable subscribe(
       OffsetSpecification offsetSpecification,
       String trackingReference,
       SubscriptionListener subscriptionListener,
-      MessageHandler messageHandler) {
+      MessageHandler messageHandler,
+      Map<String, String> subscriptionProperties) {
     // FIXME fail immediately if there's no locator (can provide a supplier that does not retry)
     List<Client.Broker> candidates = findBrokersForStream(stream);
     Client.Broker newNode = pickBroker(candidates);
@@ -103,7 +105,8 @@ Runnable subscribe(
             offsetSpecification,
             trackingReference,
             subscriptionListener,
-            messageHandler);
+            messageHandler,
+            subscriptionProperties);
 
     String key = keyForClientSubscription(newNode);
 
@@ -221,6 +224,7 @@ private static class SubscriptionTracker {
     private final MessageHandler messageHandler;
     private final StreamConsumer consumer;
     private final SubscriptionListener subscriptionListener;
+    private final Map<String, String> subscriptionProperties;
     private volatile long offset;
     private volatile boolean hasReceivedSomething = false;
     private volatile byte subscriptionIdInClient;
@@ -233,13 +237,22 @@ private SubscriptionTracker(
         OffsetSpecification initialOffsetSpecification,
         String offsetTrackingReference,
         SubscriptionListener subscriptionListener,
-        MessageHandler messageHandler) {
+        MessageHandler messageHandler,
+        Map<String, String> subscriptionProperties) {
       this.consumer = consumer;
       this.stream = stream;
       this.initialOffsetSpecification = initialOffsetSpecification;
       this.offsetTrackingReference = offsetTrackingReference;
       this.subscriptionListener = subscriptionListener;
       this.messageHandler = messageHandler;
+      if (this.offsetTrackingReference == null) {
+        this.subscriptionProperties = subscriptionProperties;
+      } else {
+        Map<String, String> properties = new ConcurrentHashMap<>(subscriptionProperties.size() + 1);
+        properties.putAll(subscriptionProperties);
+        properties.put("name", this.offsetTrackingReference);
+        this.subscriptionProperties = Collections.unmodifiableMap(properties);
+      }
     }
 
     synchronized void cancel() {
@@ -518,6 +531,25 @@ private ClientSubscriptionsManager(
                       });
             }
           };
+      ConsumerUpdateListener consumerUpdateListener =
+          (client, subscriptionId, active) -> {
+            OffsetSpecification result = null;
+            SubscriptionTracker subscriptionTracker =
+                subscriptionTrackers.get(subscriptionId & 0xFF);
+            if (subscriptionTracker != null) {
+              if (isSac(subscriptionTracker.subscriptionProperties)) {
+                result = subscriptionTracker.consumer.consumerUpdate(active);
+              } else {
+                LOGGER.debug(
+                    "Subscription {} is not a single active consumer, nothing to do.",
+                    subscriptionId);
+              }
+            } else {
+              LOGGER.debug(
+                  "Could not find stream subscription {} for consumer update", subscriptionId);
+            }
+            return result;
+          };
       ClientFactoryContext clientFactoryContext =
           ClientFactoryContext.fromParameters(
                   clientParameters
@@ -526,7 +558,8 @@ private ClientSubscriptionsManager(
                       .creditNotification(creditNotification)
                       .messageListener(messageListener)
                       .shutdownListener(shutdownListener)
-                      .metadataListener(metadataListener))
+                      .metadataListener(metadataListener)
+                      .consumerUpdateListener(consumerUpdateListener))
               .key(owner.name);
       this.client = clientFactory.client(clientFactoryContext);
       clientInitializedInManager.set(true);
@@ -631,10 +664,11 @@ synchronized void add(
       List<SubscriptionTracker> previousSubscriptions = this.subscriptionTrackers;
 
       LOGGER.debug(
-          "Subscribing to {}, requested offset specification is {}, offset tracking reference is {}",
+          "Subscribing to {}, requested offset specification is {}, offset tracking reference is {}, properties are {}",
           subscriptionTracker.stream,
           offsetSpecification == null ? DEFAULT_OFFSET_SPECIFICATION : offsetSpecification,
-          subscriptionTracker.offsetTrackingReference);
+          subscriptionTracker.offsetTrackingReference,
+          subscriptionTracker.subscriptionProperties);
       try {
         // updating data structures before subscribing
         // (to make sure they are up-to-date in case message would arrive super fast)
@@ -671,12 +705,6 @@ synchronized void add(
         offsetSpecification =
             offsetSpecification == null ? DEFAULT_OFFSET_SPECIFICATION : offsetSpecification;
 
-        Map<String, String> subscriptionProperties = Collections.emptyMap();
-        if (subscriptionTracker.offsetTrackingReference != null) {
-          subscriptionProperties = new HashMap<>(1);
-          subscriptionProperties.put("name", subscriptionTracker.offsetTrackingReference);
-        }
-
         SubscriptionContext subscriptionContext =
             new DefaultSubscriptionContext(offsetSpecification);
         subscriptionTracker.subscriptionListener.preSubscribe(subscriptionContext);
@@ -692,7 +720,7 @@ synchronized void add(
                 subscriptionTracker.stream,
                 subscriptionContext.offsetSpecification(),
                 10,
-                subscriptionProperties);
+                subscriptionTracker.subscriptionProperties);
         if (!subscribeResponse.isOk()) {
           String message =
               "Subscription to stream "

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

@@ -135,6 +135,8 @@ private interface Tracker {
 
     void flushIfNecessary();
 
+    long flush();
+
     StreamConsumer consumer();
 
     LongConsumer trackingCallback();
@@ -161,6 +163,10 @@ LongConsumer trackingCallback() {
     Runnable closingCallback() {
       return this.tracker.closingCallback();
     }
+
+    long flush() {
+      return this.tracker.flush();
+    }
   }
 
   private static final class AutoTrackingTracker implements Tracker {
@@ -181,6 +187,7 @@ private AutoTrackingTracker(
       this.clock = clock;
     }
 
+    @Override
     public Consumer<Context> postProcessingCallback() {
       return context -> {
         if (++count % messageCountBeforeStorage == 0) {
@@ -191,18 +198,25 @@ public Consumer<Context> postProcessingCallback() {
       };
     }
 
+    @Override
     public void flushIfNecessary() {
       if (this.count > 0) {
         if (this.clock.time() - this.lastTrackingActivity > this.flushIntervalInNs) {
-          long lastStoredOffset = consumer.lastStoredOffset();
-          if (lastStoredOffset < lastProcessedOffset) {
-            this.consumer.store(this.lastProcessedOffset);
-            this.lastTrackingActivity = clock.time();
-          }
+          this.flush();
         }
       }
     }
 
+    @Override
+    public long flush() {
+      long lastStoredOffset = consumer.lastStoredOffset();
+      if (lastStoredOffset < lastProcessedOffset) {
+        this.consumer.store(this.lastProcessedOffset);
+        this.lastTrackingActivity = clock.time();
+      }
+      return lastStoredOffset;
+    }
+
     @Override
     public StreamConsumer consumer() {
       return this.consumer;
@@ -261,6 +275,11 @@ public void flushIfNecessary() {
       }
     }
 
+    @Override
+    public long flush() {
+      throw new UnsupportedOperationException();
+    }
+
     @Override
     public StreamConsumer consumer() {
       return this.consumer;