Add forceReplicaForConsumer flag

To force the use of a replica for consumers. The library
tries 5 times before falling back to the stream leader when
the flag is set to true (default is false, to favor faster
recovery).

The stream leader is started first and a replica can start
a few seconds later, so with bad timing a consumer can see
only the leader when it recovers. Setting the flag to true
will force the library to retry a few times to use a replica
if one becomes available.

The performance tool has also the corresponding flag now:
--force-replica-for-consumers.

This commit also adds the following options to the performance tool:
--hearbeat, --connection-recovery-interval,
--topology-recovery-interval.

Fixes #426, #427

Author: acogoluegnes

src/docs/asciidoc/api.adoc

@@ -204,6 +204,16 @@ a new connection is open. The value must be between 1 and 255.
 |To delay the connection opening until necessary.
 |false
 
+|`requestedHeartbeat`
+|Heartbeat requested by the client.
+|60 seconds
+
+|`forceReplicaForConsumers`
+|Retry connecting until a replica is available for consumers.
+The client retries 5 times before falling back to the stream leader node.
+Set to `true` only for clustered environments, not for 1-node environments, where only the stream leader is available.
+|`false`
+
 |`id`
 |Informational ID for the environment instance.
 Used as a prefix for connection names.

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

@@ -183,7 +183,7 @@ public interface EnvironmentBuilder {
   EnvironmentBuilder virtualHost(String virtualHost);
 
   /**
-   * The hearbeat to request.
+   * The heartbeat to request.
    *
    * <p>Default is 60 seconds.
    *
@@ -322,6 +322,34 @@ EnvironmentBuilder topologyUpdateBackOffDelayPolicy(
    */
   EnvironmentBuilder lazyInitialization(boolean lazy);
 
+  /**
+   * Flag to force the connection to a stream replica for consumers.
+   *
+   * <p>The library will always prefer to connect to a stream replica to consume from, but it will
+   * fall back to the stream leader if no replica is available. This is the default behavior. Set
+   * this flag to <code>true</code> to make the library wait for a replica to become available if
+   * only the stream leader is available. This can lead to longer recovery time but help to offload
+   * a stream leader and let it deal only with write requests.
+   *
+   * <p>Note the library performs only 5 attempts to locate a replica before falling back to the
+   * leader when the flag is set to <code>true</code>.
+   *
+   * <p>The {@link #recoveryBackOffDelayPolicy(BackOffDelayPolicy)} and {@link
+   * #topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy)} policies control the time between
+   * attempts.
+   *
+   * <p><b>Do not set this flag to <code>true</code> when streams have only 1 member (the leader),
+   * e.g. for local development.</b>
+   *
+   * <p>Default is false.
+   *
+   * @param forceReplica whether to force the connection to a replica or not
+   * @return this builder instance
+   * @see #recoveryBackOffDelayPolicy(BackOffDelayPolicy)
+   * @see #topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy)
+   */
+  EnvironmentBuilder forceReplicaForConsumers(boolean forceReplica);
+
   /**
    * Create the {@link Environment} instance.
    *

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

@@ -185,6 +185,7 @@ public long applyAsLong(Object value) {
       FlushConsolidationHandler.class.getSimpleName();
   private final String NETTY_HANDLER_STREAM = StreamHandler.class.getSimpleName();
   private final String host;
+  private final String clientConnectionName;
   private final int port;
   private final Map<String, String> serverProperties;
   private final Map<String, String> connectionProperties;
@@ -313,6 +314,7 @@ public void initChannel(SocketChannel ch) {
       f = b.connect(parameters.host, parameters.port).sync();
       this.host = parameters.host;
       this.port = parameters.port;
+      this.clientConnectionName = clientConnectionName;
     } catch (Exception e) {
       String message =
           format(
@@ -2696,7 +2698,11 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
       if (evt instanceof IdleStateEvent) {
         IdleStateEvent e = (IdleStateEvent) evt;
         if (e.state() == IdleState.READER_IDLE) {
-          LOGGER.info("Closing connection because it's been idle for too long");
+          LOGGER.info(
+              "Closing connection {} on {}:{} because it's been idle for too long",
+              clientConnectionName,
+              host,
+              port);
           closing.set(true);
           closingSequence(ShutdownContext.ShutdownReason.HEARTBEAT_FAILURE);
         } else if (e.state() == IdleState.WRITER_IDLE) {

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

@@ -290,6 +290,8 @@ private Callable<List<Broker>> findBrokersForStream(String stream) {
       } else {
         mustUseReplica = false;
       }
+      LOGGER.debug(
+          "Looking for broker(s) for stream {}, forcing replica {}", stream, mustUseReplica);
       return findBrokersForStream(stream, mustUseReplica);
     };
   }

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

@@ -162,7 +162,7 @@ private void addToManager(Broker node, AgentTracker tracker) {
       try {
         pickedManager.register(tracker);
         LOGGER.debug(
-            "Assigned {} tracker {} (stream '{}') to manager {} (node {}), subscription ID {}",
+            "Assigned {} tracker {} (stream '{}') to manager {} (node {}), publisher ID {}",
             tracker.type(),
             tracker.uniqueId(),
             tracker.stream(),
@@ -605,8 +605,9 @@ private ClientProducersManager(
             }
             if (shutdownContext.isShutdownUnexpected()) {
               LOGGER.debug(
-                  "Recovering {} producer(s) after unexpected connection termination",
-                  producers.size());
+                  "Recovering {} producer(s) and {} tracking consumer(s) after unexpected connection termination",
+                  producers.size(),
+                  trackingConsumerTrackers.size());
               producers.forEach((publishingId, tracker) -> tracker.unavailable());
               trackingConsumerTrackers.forEach(AgentTracker::unavailable);
               // execute in thread pool to free the IO thread
@@ -701,7 +702,8 @@ private void assignProducersToNewManagers(
           .thenAccept(
               broker -> {
                 String key = keyForNode(broker);
-                LOGGER.debug("Assigning {} producer(s) to {}", trackers.size(), key);
+                LOGGER.debug(
+                    "Assigning {} producer(s) and consumer tracker(s) to {}", trackers.size(), key);
                 trackers.forEach(tracker -> maybeRecoverAgent(broker, tracker));
               })
           .exceptionally(
@@ -767,10 +769,10 @@ private void recoverAgent(Broker node, AgentTracker tracker) {
             | ClientClosedException
             | StreamNotAvailableException e) {
           LOGGER.debug(
-              "{} re-assignment on stream {} timed out or connection closed or stream not available, "
+              "{} re-assignment on stream {} (ID {}) timed out or connection closed or stream not available, "
                   + "refreshing candidate leader and retrying",
               tracker.type(),
-              tracker.id(),
+              tracker.identifiable() ? tracker.id() : "N/A",
               tracker.stream());
           // maybe not a good candidate, let's refresh and retry for this one
           node =

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

@@ -101,7 +101,8 @@ class StreamEnvironment implements Environment {
       boolean lazyInit,
       Function<ClientConnectionType, String> connectionNamingStrategy,
       Function<Client.ClientParameters, Client> clientFactory,
-      ObservationCollector<?> observationCollector) {
+      ObservationCollector<?> observationCollector,
+      boolean forceReplicaForConsumers) {
     this.recoveryBackOffDelayPolicy = recoveryBackOffDelayPolicy;
     this.topologyUpdateBackOffDelayPolicy = topologyBackOffDelayPolicy;
     this.byteBufAllocator = byteBufAllocator;
@@ -209,7 +210,7 @@ class StreamEnvironment implements Environment {
             maxConsumersByConnection,
             connectionNamingStrategy,
             Utils.coordinatorClientFactory(this),
-            false);
+            forceReplicaForConsumers);
     this.offsetTrackingCoordinator = new OffsetTrackingCoordinator(this);
     ClientParameters clientParametersForInit = locatorParametersCopy();
     Runnable locatorInitSequence =

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

@@ -63,6 +63,7 @@ public class StreamEnvironmentBuilder implements EnvironmentBuilder {
   private int maxConsumersByConnection = ConsumersCoordinator.MAX_SUBSCRIPTIONS_PER_CLIENT;
   private CompressionCodecFactory compressionCodecFactory;
   private boolean lazyInit = false;
+  private boolean forceReplicaForConsumers = false;
   private Function<Client.ClientParameters, Client> clientFactory = Client::new;
   private ObservationCollector<?> observationCollector = ObservationCollector.NO_OP;
 
@@ -266,6 +267,12 @@ public EnvironmentBuilder lazyInitialization(boolean lazy) {
     return this;
   }
 
+  @Override
+  public EnvironmentBuilder forceReplicaForConsumers(boolean forceReplica) {
+    this.forceReplicaForConsumers = forceReplica;
+    return this;
+  }
+
   @Override
   public TlsConfiguration tls() {
     this.tls.enable();
@@ -318,7 +325,8 @@ public Environment build() {
         lazyInit,
         connectionNamingStrategy,
         this.clientFactory,
-        this.observationCollector);
+        this.observationCollector,
+        this.forceReplicaForConsumers);
   }
 
   static final class DefaultTlsConfiguration implements TlsConfiguration {

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -22,24 +22,9 @@
 import com.google.common.util.concurrent.RateLimiter;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
-import com.rabbitmq.stream.Address;
-import com.rabbitmq.stream.AddressResolver;
-import com.rabbitmq.stream.ByteCapacity;
-import com.rabbitmq.stream.Codec;
-import com.rabbitmq.stream.ConfirmationHandler;
-import com.rabbitmq.stream.Constants;
-import com.rabbitmq.stream.Consumer;
-import com.rabbitmq.stream.ConsumerBuilder;
-import com.rabbitmq.stream.Environment;
-import com.rabbitmq.stream.EnvironmentBuilder;
+import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.EnvironmentBuilder.TlsConfiguration;
-import com.rabbitmq.stream.MessageBuilder;
-import com.rabbitmq.stream.OffsetSpecification;
-import com.rabbitmq.stream.Producer;
-import com.rabbitmq.stream.ProducerBuilder;
-import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamCreator.LeaderLocator;
-import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.codec.QpidProtonCodec;
 import com.rabbitmq.stream.codec.SimpleCodec;
 import com.rabbitmq.stream.compression.Compression;
@@ -443,6 +428,12 @@ public class StreamPerfTest implements Callable<Integer> {
       split = ",")
   private List<String> filterValues;
 
+  @CommandLine.Option(
+      names = {"--force-replica-for-consumers", "-frfc"},
+      description = "force the connection to a replica for consumers",
+      defaultValue = "false")
+  private boolean forceReplicaForConsumers;
+
   static class InstanceSyncOptions {
 
     @CommandLine.Option(
@@ -484,6 +475,33 @@ static class InstanceSyncOptions {
       converter = Utils.NotNegativeIntegerTypeConverter.class)
   private int initialCredits;
 
+  @CommandLine.Option(
+      names = {"--heartbeat", "-b"},
+      description = "requested heartbeat in seconds",
+      defaultValue = "60",
+      converter = Utils.GreaterThanOrEqualToZeroIntegerTypeConverter.class)
+  private int heartbeat;
+
+  @CommandLine.Option(
+      names = {"--connection-recovery-interval", "-cri"},
+      description =
+          "connection recovery interval in seconds. "
+              + "Examples: 5 for a fixed delay of 5 seconds, 5:10 for a first attempt after 5 seconds then "
+              + "10 seconds between attempts.",
+      defaultValue = "5",
+      converter = Utils.BackOffDelayPolicyTypeConverter.class)
+  private BackOffDelayPolicy recoveryBackOffDelayPolicy;
+
+  @CommandLine.Option(
+      names = {"--topology-recovery-interval", "-tri"},
+      description =
+          "topology recovery interval in seconds. "
+              + "Examples: 5 for a fixed delay of 5 seconds, 5:10 for a first attempt after 5 seconds then "
+              + "10 seconds between attempts.",
+      defaultValue = "5:1",
+      converter = Utils.BackOffDelayPolicyTypeConverter.class)
+  private BackOffDelayPolicy topologyBackOffDelayPolicy;
+
   private MetricsCollector metricsCollector;
   private PerformanceMetrics performanceMetrics;
   private List<Monitoring> monitorings;
@@ -747,7 +765,11 @@ public Integer call() throws Exception {
               .maxTrackingConsumersByConnection(this.trackingConsumersByConnection)
               .maxConsumersByConnection(this.consumersByConnection)
               .rpcTimeout(Duration.ofSeconds(this.rpcTimeout))
-              .requestedMaxFrameSize((int) this.requestedMaxFrameSize.toBytes());
+              .requestedMaxFrameSize((int) this.requestedMaxFrameSize.toBytes())
+              .forceReplicaForConsumers(this.forceReplicaForConsumers)
+              .requestedHeartbeat(Duration.ofSeconds(this.heartbeat))
+              .recoveryBackOffDelayPolicy(this.recoveryBackOffDelayPolicy)
+              .topologyUpdateBackOffDelayPolicy(this.topologyBackOffDelayPolicy);
 
       if (addrResolver != null) {
         environmentBuilder = environmentBuilder.addressResolver(addrResolver);
@@ -978,7 +1000,7 @@ public Integer call() throws Exception {
                           final int msgSize = this.messageSize;
 
                           try {
-                            while (true && !Thread.currentThread().isInterrupted()) {
+                            while (!Thread.currentThread().isInterrupted()) {
                               rateLimiterCallback.run();
                               // Using current time for interoperability with other tools
                               // and also across different processes.
@@ -993,9 +1015,8 @@ public Integer call() throws Exception {
                                   messageBuilder.addData(payload).build(), confirmationHandler);
                             }
                           } catch (Exception e) {
-                            if (e instanceof InterruptedException
-                                || (e.getCause() != null
-                                    && e.getCause() instanceof InterruptedException)) {
+                            if (e.getCause() != null
+                                && e.getCause() instanceof InterruptedException) {
                               LOGGER.info("Publisher #{} thread interrupted", i, e);
                             } else {
                               LOGGER.warn("Publisher #{} crashed", i, e);

src/main/java/com/rabbitmq/stream/perf/Utils.java

@@ -13,13 +13,16 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.perf;
 
+import static java.time.Duration.ofSeconds;
+
 import com.codahale.metrics.MetricRegistry;
 import com.rabbitmq.client.BuiltinExchangeType;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.SocketConfigurator;
 import com.rabbitmq.client.SocketConfigurators;
+import com.rabbitmq.stream.BackOffDelayPolicy;
 import com.rabbitmq.stream.ByteCapacity;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.StreamCreator.LeaderLocator;
@@ -676,6 +679,44 @@ public Integer convert(String input) {
     }
   }
 
+  static class BackOffDelayPolicyTypeConverter
+      implements CommandLine.ITypeConverter<BackOffDelayPolicy> {
+
+    @Override
+    public BackOffDelayPolicy convert(String input) {
+      if (input == null || input.trim().isEmpty()) {
+        typeConversionException("Value for back-off delay policy cannot be empty");
+      }
+      String[] values = input.split(":");
+      if (values.length != 1 && values.length != 2) {
+        typeConversionException("Invalid value for back-off delay policy: " + input);
+      }
+      int firstAttempt = 0, nextAttempts = 0;
+      try {
+        firstAttempt = Integer.parseInt(values[0]);
+        if (firstAttempt <= 0) {
+          throw new IllegalArgumentException();
+        }
+      } catch (Exception e) {
+        typeConversionException("Invalid value for back-off delay policy: " + input);
+      }
+      if (values.length == 2) {
+        try {
+          nextAttempts = Integer.parseInt(values[1]);
+          if (nextAttempts <= 0) {
+            throw new IllegalArgumentException();
+          }
+        } catch (Exception e) {
+          typeConversionException("Invalid value for back-off delay policy: " + input);
+        }
+      } else {
+        nextAttempts = firstAttempt;
+      }
+      return BackOffDelayPolicy.fixedWithInitialDelay(
+          ofSeconds(firstAttempt), ofSeconds(nextAttempts));
+    }
+  }
+
   private static void typeConversionException(String message) {
     throw new TypeConversionException(message);
   }
@@ -820,7 +861,7 @@ protected DistributionSummary createChunkSizeDistributionSummary(
           .tags(tags)
           .description("chunk size")
           .publishPercentiles(0.5, 0.75, 0.95, 0.99)
-          .distributionStatisticExpiry(Duration.ofSeconds(1))
+          .distributionStatisticExpiry(ofSeconds(1))
           .serviceLevelObjectives()
           .register(registry);
     }