FIX: Isolate connection observer exceptions

src/main/java/net/spy/memcached/MemcachedConnection.java

@@ -931,7 +931,13 @@ private void connected(MemcachedNode qa) {
     qa.setupForAuth();
     qa.connected();
     for (ConnectionObserver observer : connObservers) {
-      observer.connectionEstablished(qa, rt);
+      try {
+        observer.connectionEstablished(qa, rt);
+      } catch (Exception e) {
+        getLogger().warn(
+            "Exception in connectionEstablished observer %s for %s",
+            observer.getClass().getName(), qa, e);
+      }
     }
     prepareVersionInfo(qa);
     prepareAuthentication(qa);
@@ -940,7 +946,13 @@ private void connected(MemcachedNode qa) {
   private void lostConnection(MemcachedNode qa, ReconnDelay type, String cause) {
     queueReconnect(qa, type, cause);
     for (ConnectionObserver observer : connObservers) {
-      observer.connectionLost(qa);
+      try {
+        observer.connectionLost(qa);
+      } catch (Exception e) {
+        getLogger().warn(
+            "Exception in connectionLost observer %s for %s",
+            observer.getClass().getName(), qa, e);
+      }
     }
   }
 

src/test/java/net/spy/memcached/InitialObserverTest.java

@@ -0,0 +1,106 @@
+package net.spy.memcached;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class InitialObserverTest {
+
+  private ArcusClient client;
+
+  @AfterEach
+  void tearDown() {
+    if (client != null) {
+      client.shutdown();
+    }
+  }
+
+  @Test
+  void shouldNotifyNextObserverWhenConnectionEstablishedThrows()
+      throws InterruptedException {
+    // given
+    CountDownLatch latch = new CountDownLatch(1);
+    ConnectionFactoryBuilder cfb = new ConnectionFactoryBuilder();
+
+    cfb.addInitialObserver(new ConnectionObserver() {
+      @Override
+      public void connectionEstablished(MemcachedNode node, int reconnectCount) {
+        throw new RuntimeException("Test exception in connectionEstablished");
+      }
+
+      @Override
+      public void connectionLost(MemcachedNode node) {
+        // do nothing
+      }
+    });
+
+    cfb.addInitialObserver(new ConnectionObserver() {
+      @Override
+      public void connectionEstablished(MemcachedNode node, int reconnectCount) {
+        latch.countDown();
+      }
+
+      @Override
+      public void connectionLost(MemcachedNode node) {
+        // do nothing
+      }
+    });
+
+    // when
+    client = ArcusClient.createArcusClient(
+        "127.0.0.1:2181", "test", cfb
+    );
+
+    // then
+    assertTrue(latch.await(1, TimeUnit.SECONDS));
+  }
+
+  @Test
+  void shouldNotifyNextObserverWhenConnectionLostThrows()
+      throws IOException, InterruptedException {
+    // given
+    CountDownLatch latch = new CountDownLatch(1);
+    ConnectionFactoryBuilder cfb = new ConnectionFactoryBuilder();
+
+    cfb.addInitialObserver(new ConnectionObserver() {
+      @Override
+      public void connectionEstablished(MemcachedNode node, int reconnectCount) {
+        // do nothing
+      }
+
+      @Override
+      public void connectionLost(MemcachedNode node) {
+        throw new RuntimeException("Test exception in connectionLost");
+      }
+    });
+
+    cfb.addInitialObserver(new ConnectionObserver() {
+      @Override
+      public void connectionEstablished(MemcachedNode node, int reconnectCount) {
+        // do nothing
+      }
+
+      @Override
+      public void connectionLost(MemcachedNode node) {
+        latch.countDown();
+      }
+    });
+
+    client = ArcusClient.createArcusClient("127.0.0.1:2181", "test", cfb);
+
+    // when
+    for (MemcachedNode node : client.getAllNodes()) {
+      assertTrue(node.isConnected());
+      node.getChannel().socket().shutdownInput();
+    }
+    client.asyncGet("observer-exception-test");
+
+    // then
+    assertTrue(latch.await(1, TimeUnit.SECONDS));
+  }
+}