ObjectCleaner should continue cleaning despite exceptions

Motivation:
ObjectCleaner inovkes a Runnable which may execute user code (FastThreadLocal#onRemoval) and therefore exceptions maybe thrown. If an exception is thrown the cleanup thread will exit prematurely and we may never finish cleaning up which will result in leaks.

Modifications:
- ObjectCleaner should suppress exceptions and continue cleaning

Result:
ObjectCleaner will reliably clean despite exceptions being thrown.

common/src/main/java/io/netty/util/internal/ObjectCleaner.java

@@ -16,6 +16,8 @@
 package io.netty.util.internal;
 
 import io.netty.util.concurrent.FastThreadLocalThread;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
 
 import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
@@ -45,19 +47,22 @@ public void run() {
                 // Keep on processing as long as the LIVE_SET is not empty and once it becomes empty
                 // See if we can let this thread complete.
                 while (!LIVE_SET.isEmpty()) {
+                    final AutomaticCleanerReference reference;
                     try {
-                        AutomaticCleanerReference reference =
-                                (AutomaticCleanerReference) REFERENCE_QUEUE.remove(REFERENCE_QUEUE_POLL_TIMEOUT_MS);
-                        if (reference != null) {
-                            try {
-                                reference.cleanup();
-                            } finally {
-                                LIVE_SET.remove(reference);
-                            }
-                        }
+                        reference = (AutomaticCleanerReference) REFERENCE_QUEUE.remove(REFERENCE_QUEUE_POLL_TIMEOUT_MS);
                     } catch (InterruptedException ex) {
                         // Just consume and move on
                         interrupted = true;
+                        continue;
+                    }
+                    if (reference != null) {
+                        try {
+                            reference.cleanup();
+                        } catch (Throwable ignored) {
+                            // ignore exceptions, and don't log in case the logger throws an exception, blocks, or has
+                            // other unexpected side effects.
+                        }
+                        LIVE_SET.remove(reference);
                     }
                 }
                 CLEANER_RUNNING.set(false);

common/src/test/java/io/netty/util/internal/ObjectCleanerTest.java

@@ -20,10 +20,14 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
 
 public class ObjectCleanerTest {
 
     private Thread temporaryThread;
+    private Object temporaryObject;
 
     @Test(timeout = 5000)
     public void testCleanup() throws Exception {
@@ -60,4 +64,50 @@ public void run() {
             Thread.sleep(100);
         }
     }
+
+    @Test(timeout = 5000)
+    public void testCleanupContinuesDespiteThrowing() throws InterruptedException {
+        final AtomicInteger freeCalledCount = new AtomicInteger();
+        final CountDownLatch latch = new CountDownLatch(1);
+        temporaryThread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    latch.await();
+                } catch (InterruptedException ignore) {
+                    // just ignore
+                }
+            }
+        });
+        temporaryThread.start();
+        temporaryObject = new Object();
+        ObjectCleaner.register(temporaryThread, new Runnable() {
+            @Override
+            public void run() {
+                freeCalledCount.incrementAndGet();
+                throw new RuntimeException("expected");
+            }
+        });
+        ObjectCleaner.register(temporaryObject, new Runnable() {
+            @Override
+            public void run() {
+                freeCalledCount.incrementAndGet();
+                throw new RuntimeException("expected");
+            }
+        });
+
+        latch.countDown();
+        temporaryThread.join();
+        assertEquals(0, freeCalledCount.get());
+
+        // Null out the temporary object to ensure it is enqueued for GC.
+        temporaryThread = null;
+        temporaryObject = null;
+
+        while (freeCalledCount.get() != 2) {
+            System.gc();
+            System.runFinalization();
+            Thread.sleep(100);
+        }
+    }
 }