Merge pull request #1173 from ably/ECO-5621/fix-orphan-timer-threads

fix: orphan Timer instances on release

Author: ttypic

lib/src/main/java/io/ably/lib/realtime/ChannelBase.java

@@ -8,6 +8,7 @@
 import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.ably.lib.http.BasePaginatedQuery;
 import io.ably.lib.http.Http;
@@ -95,7 +96,7 @@ public abstract class ChannelBase extends EventEmitter<ChannelEvent, ChannelStat
     /**
      * @see #markAsReleased()
      */
-    private boolean released = false;
+    private AtomicBoolean released = new AtomicBoolean(false);
 
     @Nullable private final LiveObjectsPlugin liveObjectsPlugin;
 
@@ -330,8 +331,8 @@ public void detach() throws AblyException {
     /**
      * Mark channel as released that means we can't perform any operation on this channel anymore
      */
-    public synchronized void markAsReleased() {
-        released = true;
+    public void markAsReleased() {
+        released.set(true);
     }
 
     /**
@@ -395,7 +396,7 @@ private void sendDetachMessage(CompletionListener listener) throws AblyException
             }
 
             this.attachResume = false;
-            if (released) {
+            if (released.get()) {
                 setDetached(null);
             } else {
                 setState(ChannelState.detaching, null);
@@ -572,7 +573,7 @@ public void run() {
     }
 
     private void checkChannelIsNotReleased() {
-        if (released) throw new IllegalStateException("Unable to perform any operation on released channel");
+        if (released.get()) throw new IllegalStateException("Unable to perform any operation on released channel");
     }
 
     /**
@@ -621,17 +622,17 @@ synchronized private void detachWithTimeout(final CompletionListener listener) {
         final ChannelState originalState = state;
         Timer currentDetachTimer;
         try {
-            currentDetachTimer = new Timer();
+            currentDetachTimer = released.get() ? null : new Timer();
         } catch(Throwable t) {
             /* an exception instancing the timer can arise because the runtime is exiting */
             callCompletionListenerError(listener, ErrorInfo.fromThrowable(t));
             return;
         }
-        attachTimer = released ? null : currentDetachTimer;
+        attachTimer = currentDetachTimer;
 
         try {
             // If channel has been released, completionListener won't be invoked anyway
-            CompletionListener completionListener = released ? null : new CompletionListener() {
+            CompletionListener completionListener = released.get() ? null : new CompletionListener() {
                 @Override
                 public void onSuccess() {
                     clearAttachTimers();

lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java

@@ -36,6 +36,7 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Set;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
@@ -2573,6 +2574,29 @@ public void connect_on_closing_client_should_reinitialize_channels() throws Ably
         }
     }
 
+    @Test
+    public void release_should_not_prevent_graceful_test_end() throws Exception {
+        ClientOptions opts = createOptions(testVars.keys[0].keyStr);
+        try (AblyRealtime ably = new AblyRealtime(opts)) {
+            ably.channels.get("channel_should_be_released");
+            ably.channels.release("channel_should_be_released");
+        }
+
+        Set<Thread> threads = Thread.getAllStackTraces().keySet();
+        List<Thread> timers = threads.stream()
+            .filter(t -> t.getName().startsWith("Timer-"))
+            .toList();
+
+        long timeout = 1_000;
+        long start = System.currentTimeMillis();
+
+        while(timers.stream().anyMatch(Thread::isAlive) && System.currentTimeMillis() - start < timeout) {
+            Thread.sleep(100);
+        }
+
+        assertFalse("Found orphan Timer threads", timers.stream().anyMatch(Thread::isAlive));
+    }
+
     /**
      * This test ensures that when the connection is manually triggered, the channel can successfully
      * transition to the attached state without interference or rewriting of its immediate attach action.