Merge pull request #437 from rabbitmq/do-not-unsubscribe-with-closed-…

…connection

Do not unsubscribe with closed client

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

@@ -2029,8 +2029,14 @@ public void done() {
     }
   }
 
+  static Response responseOk() {
+    return Response.OK;
+  }
+
   public static class Response {
 
+    private static final Response OK = new Response(RESPONSE_CODE_OK);
+
     private final short responseCode;
 
     public Response(short responseCode) {

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

@@ -814,6 +814,10 @@ private void assignConsumersToStream(
         boolean maybeCloseClient) {
       Runnable consumersClosingCallback =
           () -> {
+            LOGGER.debug(
+                "Running consumer closing callback after recovery failure, "
+                    + "closing {} subscription(s)",
+                subscriptions.size());
             for (SubscriptionTracker affectedSubscription : subscriptions) {
               try {
                 affectedSubscription.consumer.closeAfterStreamDeletion();
@@ -1078,7 +1082,13 @@ synchronized void remove(SubscriptionTracker subscriptionTracker) {
       try {
         Client.Response unsubscribeResponse =
             Utils.callAndMaybeRetry(
-                () -> client.unsubscribe(subscriptionIdInClient),
+                () -> {
+                  if (client.isOpen()) {
+                    return client.unsubscribe(subscriptionIdInClient);
+                  } else {
+                    return Client.responseOk();
+                  }
+                },
                 RETRY_ON_TIMEOUT,
                 "Unsubscribe request for consumer %d on stream '%s'",
                 subscriptionTracker.consumer.id(),

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

@@ -230,6 +230,7 @@ static <T> T callAndMaybeRetry(
     while (keepTrying) {
       try {
         attempt++;
+        LOGGER.debug("Starting attempt #{} for operation '{}'", attempt, description);
         T result = operation.call();
         Duration operationDuration = Duration.ofNanos(System.nanoTime() - startTime);
         LOGGER.debug(

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

@@ -29,10 +29,7 @@
 import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.isNull;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.codec.WrapperMessageBuilder;
@@ -459,6 +456,42 @@ void subscribeShouldSubscribeToStreamAndDispatchMessage_UnsubscribeShouldUnsubsc
     assertThat(messageHandlerCalls.get()).isEqualTo(1);
   }
 
+  @Test
+  void shouldNotUnsubscribeIfClientIsClosed() {
+    when(locator.metadata("stream")).thenReturn(metadata(null, replicas()));
+
+    when(clientFactory.client(any())).thenReturn(client);
+    when(client.subscribe(
+            subscriptionIdCaptor.capture(),
+            anyString(),
+            any(OffsetSpecification.class),
+            anyInt(),
+            anyMap()))
+        .thenReturn(new Client.Response(Constants.RESPONSE_CODE_OK));
+
+    Runnable closingRunnable =
+        coordinator.subscribe(
+            consumer,
+            "stream",
+            OffsetSpecification.first(),
+            null,
+            NO_OP_SUBSCRIPTION_LISTENER,
+            () -> {},
+            (offset, message) -> {},
+            Collections.emptyMap(),
+            flowStrategy());
+    verify(clientFactory, times(1)).client(any());
+    verify(client, times(1))
+        .subscribe(anyByte(), anyString(), any(OffsetSpecification.class), anyInt(), anyMap());
+
+    when(client.isOpen()).thenReturn(false);
+    when(client.unsubscribe(subscriptionIdCaptor.getValue()))
+        .thenReturn(new Client.Response(Constants.RESPONSE_CODE_OK));
+
+    closingRunnable.run();
+    verify(client, never()).unsubscribe(subscriptionIdCaptor.getValue());
+  }
+
   @Test
   void subscribeShouldSubscribeToStreamAndDispatchMessageWithManySubscriptions() {
     when(locator.metadata("stream")).thenReturn(metadata(leader(), null));

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

@@ -357,7 +357,6 @@ void consumerShouldBeClosedWhenStreamGetsDeleted(TestInfo info) throws Exception
     environment.deleteStream(s);
 
     TestUtils.waitAtMost(10, () -> !consumer.isOpen());
-
     assertThat(consumer.isOpen()).isFalse();
   }
 

src/test/resources/logback-test.xml

@@ -6,10 +6,8 @@
     </appender>
 
     <logger name="com.rabbitmq.stream" level="warn" />
-    <logger name="com.rabbitmq.stream.perf" level="info" />
 
     <logger name="com.rabbitmq.stream.perf.Version" level="error" />
-    <logger name="org.eclipse.jetty" level="warn" />
 
     <root level="info">
         <appender-ref ref="STDOUT" />