Notify SAC when it is no longer active

A single active consumer may not always receive a notification from the
broker when it gets inactive. An obvious reason is the consumer
connection goes down. It is still possible to call the consumer update
listener from the library, which can help applications take an
appropriate action when a consumer goes from active to inactive.

This commit implements the call to the listener under such circumstances
(connection closed, stream unavailable because restarted, normal
consumer closing).

Author: acogoluegnes

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

@@ -303,12 +303,7 @@ public void initChannel(SocketChannel ch) {
         });
 
     ChannelFuture f;
-    String clientConnectionName =
-        parameters.clientProperties == null
-            ? ""
-            : (parameters.clientProperties.containsKey("connection_name")
-                ? parameters.clientProperties.get("connection_name")
-                : "");
+    String clientConnectionName = parameters.clientProperties.getOrDefault("connection_name", "");
     try {
       LOGGER.debug(
           "Trying to create stream connection to {}:{}, with client connection name '{}'",
@@ -1505,6 +1500,10 @@ String connectionName() {
     return builder.append(serverAddress()).toString();
   }
 
+  String clientConnectionName() {
+    return this.clientConnectionName;
+  }
+
   private String serverAddress() {
     SocketAddress remoteAddress = remoteAddress();
     if (remoteAddress instanceof InetSocketAddress) {

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

@@ -481,6 +481,7 @@ public void close() {
   }
 
   void closeFromEnvironment() {
+    this.maybeNotifyActiveToInactiveSac();
     LOGGER.debug("Calling consumer {} closing callback (stream {})", this.id, this.stream);
     this.closingCallback.run();
     closed.set(true);
@@ -490,6 +491,7 @@ void closeFromEnvironment() {
 
   void closeAfterStreamDeletion() {
     if (closed.compareAndSet(false, true)) {
+      this.maybeNotifyActiveToInactiveSac();
       this.environment.removeConsumer(this);
       this.status = Status.CLOSED;
     }
@@ -506,11 +508,23 @@ void setTrackingClient(Client client) {
   void setSubscriptionClient(Client client) {
     this.subscriptionClient = client;
     if (client == null && this.isSac()) {
+      maybeNotifyActiveToInactiveSac();
       // we lost the connection
       this.sacActive = false;
     }
   }
 
+  private void maybeNotifyActiveToInactiveSac() {
+    if (this.isSac() && this.sacActive) {
+      LOGGER.debug(
+          "Single active consumer {} from stream {} with name {} is unavailable, calling consumer update listener",
+          this.id,
+          this.stream,
+          this.name);
+      this.consumerUpdate(false);
+    }
+  }
+
   synchronized void unavailable() {
     this.status = Status.NOT_AVAILABLE;
     this.trackingClient = null;
@@ -623,4 +637,13 @@ private void checkNotClosed() {
   long id() {
     return this.id;
   }
+
+  String subscriptionConnectionName() {
+    Client client = this.subscriptionClient;
+    if (client == null) {
+      return "<no-connection>";
+    } else {
+      return client.clientConnectionName();
+    }
+  }
 }

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

@@ -187,6 +187,10 @@ public void store(long offset) {
         "Consumer#store(long) does not work for super streams, use MessageHandler.Context#storeOffset() instead");
   }
 
+  Consumer consumer(String partition) {
+    return this.consumers.get(partition);
+  }
+
   @Override
   public long storedOffset() {
     throw new UnsupportedOperationException(

src/test/java/com/rabbitmq/stream/Host.java

@@ -47,30 +47,34 @@ public static String capture(InputStream is) throws IOException {
     return buff.toString();
   }
 
-  private static Process executeCommand(String command) throws IOException {
+  private static Process executeCommand(String command) {
     return executeCommand(command, false);
   }
 
-  private static Process executeCommand(String command, boolean ignoreError) throws IOException {
-    Process pr = executeCommandProcess(command);
-
-    int ev = waitForExitValue(pr);
-    if (ev != 0 && !ignoreError) {
-      String stdout = capture(pr.getInputStream());
-      String stderr = capture(pr.getErrorStream());
-      throw new IOException(
-          "unexpected command exit value: "
-              + ev
-              + "\ncommand: "
-              + command
-              + "\n"
-              + "\nstdout:\n"
-              + stdout
-              + "\nstderr:\n"
-              + stderr
-              + "\n");
+  private static Process executeCommand(String command, boolean ignoreError) {
+    try {
+      Process pr = executeCommandProcess(command);
+
+      int ev = waitForExitValue(pr);
+      if (ev != 0 && !ignoreError) {
+        String stdout = capture(pr.getInputStream());
+        String stderr = capture(pr.getErrorStream());
+        throw new IOException(
+            "unexpected command exit value: "
+                + ev
+                + "\ncommand: "
+                + command
+                + "\n"
+                + "\nstdout:\n"
+                + stdout
+                + "\nstderr:\n"
+                + stderr
+                + "\n");
+      }
+      return pr;
+    } catch (IOException e) {
+      throw new RuntimeException(e);
     }
-    return pr;
   }
 
   public static String hostname() throws IOException {
@@ -110,6 +114,10 @@ public static Process rabbitmqctl(String command) throws IOException {
     return executeCommand(rabbitmqctlCommand() + " " + command);
   }
 
+  static Process rabbitmqStreams(String command) {
+    return executeCommand(rabbitmqStreamsCommand() + " " + command);
+  }
+
   public static Process rabbitmqctlIgnoreError(String command) throws IOException {
     return executeCommand(rabbitmqctlCommand() + " " + command, true);
   }
@@ -189,11 +197,19 @@ static List<ConnectionInfo> toConnectionInfoList(String json) {
     return GSON.fromJson(json, new TypeToken<List<ConnectionInfo>>() {}.getType());
   }
 
-  public static Process killStreamLeaderProcess(String stream) throws IOException {
-    return rabbitmqctl(
-        "eval 'case rabbit_stream_manager:lookup_leader(<<\"/\">>, <<\""
-            + stream
-            + "\">>) of {ok, Pid} -> exit(Pid, kill); Pid -> exit(Pid, kill) end.'");
+  public static void restartStream(String stream) {
+    rabbitmqStreams(" restart_stream " + stream);
+  }
+
+  public static Process killStreamLeaderProcess(String stream) {
+    try {
+      return rabbitmqctl(
+          "eval 'case rabbit_stream_manager:lookup_leader(<<\"/\">>, <<\""
+              + stream
+              + "\">>) of {ok, Pid} -> exit(Pid, kill); Pid -> exit(Pid, kill) end.'");
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
   }
 
   public static void addUser(String username, String password) throws IOException {
@@ -243,7 +259,7 @@ public static void setEnv(String parameter, String value) throws IOException {
   public static String rabbitmqctlCommand() {
     String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
     if (rabbitmqCtl == null) {
-      throw new IllegalStateException("Please define the rabbitmqctl.bin system property");
+      rabbitmqCtl = DOCKER_PREFIX + "rabbitmq";
     }
     if (rabbitmqCtl.startsWith(DOCKER_PREFIX)) {
       String containerId = rabbitmqCtl.split(":")[1];
@@ -253,6 +269,15 @@ public static String rabbitmqctlCommand() {
     }
   }
 
+  private static String rabbitmqStreamsCommand() {
+    String rabbitmqctl = rabbitmqctlCommand();
+    int lastIndex = rabbitmqctl.lastIndexOf("rabbitmqctl");
+    if (lastIndex == -1) {
+      throw new IllegalArgumentException("Not a valid rabbitqmctl command: " + rabbitmqctl);
+    }
+    return rabbitmqctl.substring(0, lastIndex) + "rabbitmq-streams";
+  }
+
   public static AutoCloseable diskAlarm() throws Exception {
     return new CallableAutoCloseable(
         () -> {

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

@@ -0,0 +1,62 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// 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
+// info@rabbitmq.com.
+package com.rabbitmq.stream.impl;
+
+import static org.assertj.core.api.Assertions.fail;
+
+import java.time.Duration;
+import org.assertj.core.api.AbstractObjectAssert;
+
+final class Assertions {
+
+  private Assertions() {}
+
+  static SyncAssert assertThat(TestUtils.Sync sync) {
+    return new SyncAssert(sync);
+  }
+
+  static class SyncAssert extends AbstractObjectAssert<SyncAssert, TestUtils.Sync> {
+
+    private SyncAssert(TestUtils.Sync sync) {
+      super(sync, SyncAssert.class);
+    }
+
+    SyncAssert completes() {
+      return this.completes(TestUtils.DEFAULT_CONDITION_TIMEOUT);
+    }
+
+    SyncAssert completes(Duration timeout) {
+      boolean completed = actual.await(timeout);
+      if (!completed) {
+        fail("Sync timed out after %d ms", timeout.toMillis());
+      }
+      return this;
+    }
+
+    SyncAssert hasCompleted() {
+      if (!this.actual.hasCompleted()) {
+        fail("Sync should have completed but has not");
+      }
+      return this;
+    }
+
+    SyncAssert hasNotCompleted() {
+      if (this.actual.hasCompleted()) {
+        fail("Sync should have not completed");
+      }
+      return this;
+    }
+  }
+}

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

@@ -14,24 +14,24 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream.impl;
 
-import static com.rabbitmq.stream.impl.TestUtils.publishAndWaitForConfirms;
-import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
+import static com.rabbitmq.stream.impl.Assertions.assertThat;
+import static com.rabbitmq.stream.impl.TestUtils.*;
 import static org.assertj.core.api.Assertions.assertThat;
 
-import com.rabbitmq.stream.Consumer;
-import com.rabbitmq.stream.Environment;
-import com.rabbitmq.stream.EnvironmentBuilder;
-import com.rabbitmq.stream.OffsetSpecification;
+import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast311Condition;
 import io.netty.channel.EventLoopGroup;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Stream;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 @ExtendWith({
   TestUtils.StreamTestInfrastructureExtension.class,
@@ -237,4 +237,72 @@ void externalTrackingSecondConsumerShouldTakeOverWhereTheFirstOneLeftOff() throw
     // nothing stored on the server side
     assertThat(cf.get().queryOffset(consumerName, stream).getOffset()).isZero();
   }
+
+  public static Stream<java.util.function.Consumer<Consumer>>
+      activeConsumerShouldGetUpdateNotificationAfterDisruption() {
+    return Stream.of(
+        namedConsumer(consumer -> Host.killConnection(connectionName(consumer)), "kill connection"),
+        namedConsumer(consumer -> Host.restartStream(stream(consumer)), "restart stream"),
+        namedConsumer(Consumer::close, "close consumer"));
+  }
+
+  @ParameterizedTest
+  @MethodSource
+  @TestUtils.DisabledIfRabbitMqCtlNotSet
+  void activeConsumerShouldGetUpdateNotificationAfterDisruption(
+      java.util.function.Consumer<Consumer> disruption) {
+    String consumerName = "foo";
+    Sync consumer1Active = sync();
+    Sync consumer1Inactive = sync();
+    Consumer consumer1 =
+        environment.consumerBuilder().stream(stream)
+            .name(consumerName)
+            .noTrackingStrategy()
+            .singleActiveConsumer()
+            .consumerUpdateListener(
+                context -> {
+                  if (context.isActive()) {
+                    consumer1Active.down();
+                  } else {
+                    consumer1Inactive.down();
+                  }
+                  return OffsetSpecification.next();
+                })
+            .messageHandler((context, message) -> {})
+            .build();
+
+    Sync consumer2Active = sync();
+    Sync consumer2Inactive = sync();
+    environment.consumerBuilder().stream(stream)
+        .name(consumerName)
+        .noTrackingStrategy()
+        .singleActiveConsumer()
+        .consumerUpdateListener(
+            context -> {
+              if (!context.isActive()) {
+                consumer2Inactive.down();
+              }
+              return OffsetSpecification.next();
+            })
+        .messageHandler((context, message) -> {})
+        .build();
+
+    assertThat(consumer1Active).completes();
+    assertThat(consumer2Inactive).hasNotCompleted();
+    assertThat(consumer1Inactive).hasNotCompleted();
+    assertThat(consumer2Active).hasNotCompleted();
+
+    disruption.accept(consumer1);
+
+    assertThat(consumer2Inactive).hasNotCompleted();
+    assertThat(consumer1Inactive).completes();
+  }
+
+  private static String connectionName(Consumer consumer) {
+    return ((StreamConsumer) consumer).subscriptionConnectionName();
+  }
+
+  private static String stream(Consumer consumer) {
+    return ((StreamConsumer) consumer).stream();
+  }
 }