fixes

Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>

Author: OlegDokuka

rsocket-test/src/main/java/io/rsocket/test/TransportTest.java

@@ -48,14 +48,12 @@
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Assumptions;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscription;
 import reactor.core.*;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.Operators;
-import reactor.core.publisher.Sinks;
+import reactor.core.publisher.*;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
@@ -85,38 +83,47 @@ static String read(String resourceName) {
     }
   }
 
+  @BeforeEach
+  default void setup() {
+    Hooks.onOperatorDebug();
+  }
+
   @AfterEach
   default void close() {
-    getTransportPair().responder.awaitAllInteractionTermination(getTimeout());
-    getTransportPair().dispose();
-    getTransportPair().awaitClosed();
-    RuntimeException throwable =
-        new RuntimeException() {
-          @Override
-          public synchronized Throwable fillInStackTrace() {
-            return this;
-          }
-
-          @Override
-          public String getMessage() {
-            return Arrays.toString(getSuppressed());
-          }
-        };
-
-    try {
-      getTransportPair().byteBufAllocator2.assertHasNoLeaks();
-    } catch (Throwable t) {
-      throwable = Exceptions.addSuppressed(throwable, t);
-    }
-
     try {
-      getTransportPair().byteBufAllocator1.assertHasNoLeaks();
-    } catch (Throwable t) {
-      throwable = Exceptions.addSuppressed(throwable, t);
-    }
-
-    if (throwable.getSuppressed().length > 0) {
-      throw throwable;
+      getTransportPair().responder.awaitAllInteractionTermination(getTimeout());
+      getTransportPair().dispose();
+      getTransportPair().awaitClosed();
+      RuntimeException throwable =
+          new RuntimeException() {
+            @Override
+            public synchronized Throwable fillInStackTrace() {
+              return this;
+            }
+
+            @Override
+            public String getMessage() {
+              return Arrays.toString(getSuppressed());
+            }
+          };
+
+      try {
+        getTransportPair().byteBufAllocator2.assertHasNoLeaks();
+      } catch (Throwable t) {
+        throwable = Exceptions.addSuppressed(throwable, t);
+      }
+
+      try {
+        getTransportPair().byteBufAllocator1.assertHasNoLeaks();
+      } catch (Throwable t) {
+        throwable = Exceptions.addSuppressed(throwable, t);
+      }
+
+      if (throwable.getSuppressed().length > 0) {
+        throw throwable;
+      }
+    } finally {
+      Hooks.resetOnOperatorDebug();
     }
   }