Test connection is closed if consumer update takes too long

References rabbitmq/rabbitmq-server#7743

Author: acogoluegnes

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

@@ -13,6 +13,8 @@
 // [email protected].
 package com.rabbitmq.stream;
 
+import static java.lang.String.format;
+
 import com.google.common.reflect.TypeToken;
 import com.google.gson.Gson;
 import java.io.BufferedReader;
@@ -29,7 +31,7 @@ public class Host {
 
   private static final Gson GSON = new Gson();
 
-  private static String capture(InputStream is) throws IOException {
+  public static String capture(InputStream is) throws IOException {
     BufferedReader br = new BufferedReader(new InputStreamReader(is));
     String line;
     StringBuilder buff = new StringBuilder();
@@ -103,7 +105,7 @@ public static Process killConnection(String connectionName) {
       List<ConnectionInfo> cs = listConnections();
       if (cs.stream().filter(c -> connectionName.equals(c.clientProvidedName())).count() != 1) {
         throw new IllegalArgumentException(
-            String.format(
+            format(
                 "Could not find 1 connection '%s' in stream connections: %s",
                 connectionName,
                 cs.stream()
@@ -137,6 +139,10 @@ public static Process killStreamLeaderProcess(String stream) throws IOException
             + "\">>) of {ok, Pid} -> exit(Pid, kill); Pid -> exit(Pid, kill) end.'");
   }
 
+  public static void setEnv(String parameter, String value) throws IOException {
+    rabbitmqctl(format("eval 'application:set_env(rabbitmq_stream, %s, %s).'", parameter, value));
+  }
+
   public static String rabbitmqctlCommand() {
     String rabbitmqCtl = System.getProperty("rabbitmqctl.bin");
     if (rabbitmqCtl == null) {

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

@@ -13,6 +13,7 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.impl.TestUtils.BrokerVersion.RABBITMQ_3_11_14;
 import static com.rabbitmq.stream.impl.TestUtils.ResponseConditions.ko;
 import static com.rabbitmq.stream.impl.TestUtils.ResponseConditions.ok;
 import static com.rabbitmq.stream.impl.TestUtils.ResponseConditions.responseCode;
@@ -35,9 +36,11 @@
 import com.rabbitmq.stream.impl.Client.CreditNotification;
 import com.rabbitmq.stream.impl.Client.MessageListener;
 import com.rabbitmq.stream.impl.Client.Response;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast311Condition;
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfRabbitMqCtlNotSet;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -578,6 +581,7 @@ void superStreamRebalancingShouldWorkWhilePublishing(TestInfo info) throws Excep
     } finally {
       keepPublishing.set(false);
       deleteSuperStreamTopology(c, superStream, 3);
+      c.close();
     }
   }
 
@@ -593,4 +597,38 @@ void singleActiveConsumerMustHaveName() {
             Collections.singletonMap("single-active-consumer", "true"));
     assertThat(response).is(ko()).has(responseCode(Constants.RESPONSE_CODE_PRECONDITION_FAILED));
   }
+
+  @Test
+  @DisabledIfRabbitMqCtlNotSet
+  @BrokerVersionAtLeast(RABBITMQ_3_11_14)
+  void connectionShouldBeClosedIfConsumerUpdateTakesTooLong() throws Exception {
+    Duration timeout = Duration.ofSeconds(1);
+    try {
+      Host.setEnv("request_timeout", String.valueOf(timeout.getSeconds()));
+      CountDownLatch shutdownLatch = new CountDownLatch(1);
+      Client client =
+          cf.get(
+              new ClientParameters()
+                  .consumerUpdateListener(
+                      (c, subscriptionId, active) -> {
+                        try {
+                          Thread.sleep(timeout.multipliedBy(2).toMillis());
+                        } catch (InterruptedException e) {
+                          throw new RuntimeException(e);
+                        }
+                        return OffsetSpecification.first();
+                      })
+                  .shutdownListener(shutdownContext -> shutdownLatch.countDown()));
+      Map<String, String> parameters = new HashMap<>();
+      parameters.put("single-active-consumer", "true");
+      parameters.put("name", "foo");
+      Response response =
+          client.subscribe(b(0), stream, OffsetSpecification.first(), 1, parameters);
+      assertThat(response).is(ok());
+
+      assertThat(latchAssert(shutdownLatch)).completes(timeout.multipliedBy(5));
+    } finally {
+      Host.setEnv("request_timeout", "60000");
+    }
+  }
 }

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

@@ -939,7 +939,8 @@ public enum BrokerVersion {
     RABBITMQ_3_11("3.11.0"),
     RABBITMQ_3_11_7("3.11.7"),
     RABBITMQ_3_11_9("3.11.9"),
-    RABBITMQ_3_11_11("3.11.11");
+    RABBITMQ_3_11_11("3.11.11"),
+    RABBITMQ_3_11_14("3.11.14");
 
     final String value;