ISPN-1654 Switch to Iterable for detecting stopped or crashed members

core/src/test/java/org/infinispan/util/ClusterIdGeneratorTest.java

@@ -32,7 +32,7 @@
 import static org.testng.AssertJUnit.assertEquals;
 
 @Test(groups = "functional", testName = "util.ClusterIdGeneratorTest")
-   public class ClusterIdGeneratorTest {
+public class ClusterIdGeneratorTest {
 
    public void testGenerateVersion() {
       ClusterIdGenerator vg = new ClusterIdGenerator(null, null);

server/hotrod/src/main/scala/org/infinispan/server/hotrod/CrashedMemberDetectorListener.scala

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2012 Red Hat, Inc. and/or its affiliates.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301 USA
+ */
+
+package org.infinispan.server.hotrod
+
+import logging.Log
+import org.infinispan.notifications.Listener
+import org.infinispan.notifications.cachemanagerlistener.annotation.ViewChanged
+import org.infinispan.notifications.cachemanagerlistener.event.ViewChangedEvent
+import scala.collection.JavaConversions._
+import org.infinispan.Cache
+import org.infinispan.remoting.transport.Address
+
+/**
+ * Listener that detects crashed or stopped members and removes them from
+ * the address cache.
+ *
+ * @author Galder Zamarreño
+ * @since 5.1
+ */
+@Listener(sync = false) // Use a separate thread to avoid blocking the view handler thread
+class CrashedMemberDetectorListener(addressCache: Cache[Address, ServerAddress]) extends Log {
+
+   @ViewChanged
+   def handleViewChange(e: ViewChangedEvent) {
+      val cacheManager = e.getCacheManager
+      // Only the coordinator can potentially make modifications related to
+      // crashed members. This is to avoid all nodes trying to make the same
+      // modification which would be wasteful and lead to deadlocks.
+      if (cacheManager.isCoordinator)
+         detectCrashedMember(e)
+   }
+
+   private[hotrod] def detectCrashedMember(e: ViewChangedEvent) {
+      trace("View change received on coordinator: %s", e)
+      try {
+         val newMembers = collectionAsScalaIterable(e.getNewMembers)
+         val oldMembers = collectionAsScalaIterable(e.getOldMembers)
+         val goneMembers = oldMembers.filterNot(newMembers contains _)
+         goneMembers.foreach { addr =>
+            trace("Remove %s from address cache", addr)
+            addressCache.remove(addr)
+         }
+      } catch {
+         case t: Throwable => logErrorDetectingCrashedMember(t)
+      }
+   }
+
+}

server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodServer.scala

@@ -110,7 +110,7 @@ class HotRodServer extends AbstractProtocolServer("HotRod") with Log {
       addressCache = cacheManager.getCache(ADDRESS_CACHE_NAME)
       clusterAddress = cacheManager.getAddress
       address = new ServerAddress(host, port)
-      cacheManager.addListener(new CrashedMemberDetectorListener)
+      cacheManager.addListener(new CrashedMemberDetectorListener(addressCache))
       // Map cluster address to server endpoint address
       debug("Map %s cluster address with %s server endpoint in address cache", clusterAddress, address)
       addressCache.put(clusterAddress, address)
@@ -153,35 +153,6 @@ class HotRodServer extends AbstractProtocolServer("HotRod") with Log {
 
    private[hotrod] def getAddressCache = addressCache
 
-   @Listener(sync = false) // Use a separate thread to avoid blocking the view handler thread
-   class CrashedMemberDetectorListener {
-
-      @ViewChanged
-      def handleViewChange(e: ViewChangedEvent) {
-         val cacheManager = e.getCacheManager
-         // Only the coordinator can potentially make modifications related to crashed members.
-         // This is to avoid all nodes trying to make the same modification which would be wasteful and lead to deadlocks.
-         if (cacheManager.isCoordinator) {
-            trace("View change received on coordinator: %s", e)
-            try {
-               val newMembers = asScalaIterator(e.getNewMembers.iterator())
-               val oldMembers = asScalaIterator(e.getOldMembers.iterator())
-               val goneMembers = oldMembers.filterNot(newMembers contains)
-               if (goneMembers.hasNext) {
-                  trace("Somone left the cluster, oldMembers=%s newMembers=%s", oldMembers, newMembers)
-                  goneMembers.foreach { addr =>
-                     trace("Remove %s from address cache", addr)
-                     addressCache.remove(addr)
-                  }
-               }
-            } catch {
-               case t: Throwable => logErrorDetectingCrashedMember(t)
-            }
-         }
-      }
-
-   }
-
 }
 
 object HotRodServer {

server/hotrod/src/test/scala/org/infinispan/server/hotrod/CrashedMemberDetectorTest.scala

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2012 Red Hat, Inc. and/or its affiliates.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+ * 02110-1301 USA
+ */
+
+package org.infinispan.server.hotrod
+
+import org.testng.annotations.Test
+import org.infinispan.test.fwk.TestCacheManagerFactory
+import org.infinispan.remoting.transport.Address
+import org.infinispan.notifications.cachemanagerlistener.event.EventImpl
+import org.infinispan.notifications.cachemanagerlistener.event.Event.Type
+import org.infinispan.distribution.TestAddress
+import java.util.ArrayList
+import org.testng.AssertJUnit._
+import org.infinispan.test.SingleCacheManagerTest
+
+/**
+ * Tests crashed or stopped member logic.
+ *
+ * @author Galder Zamarreño
+ * @since 5.1
+ */
+@Test(groups = Array("functional"), testName = "server.hotrod.CrashedMemberDetectorTest")
+class CrashedMemberDetectorTest extends SingleCacheManagerTest {
+
+   protected def createCacheManager() =
+      TestCacheManagerFactory.createLocalCacheManager(false)
+
+   def testDetectCrashedMembers() {
+      val cache = cacheManager.getCache[Address, ServerAddress]()
+      cache.put(new TestAddress(1), new ServerAddress("a", 123))
+      cache.put(new TestAddress(2), new ServerAddress("b", 456))
+      cache.put(new TestAddress(3), new ServerAddress("c", 789))
+
+      val detector = new CrashedMemberDetectorListener(cache)
+
+      val oldMembers = new ArrayList[Address]()
+      oldMembers.add(new TestAddress(1))
+      oldMembers.add(new TestAddress(3))
+      oldMembers.add(new TestAddress(2))
+
+      val newMembers = new ArrayList[Address]()
+      newMembers.add(new TestAddress(1))
+      newMembers.add(new TestAddress(2))
+
+      val e = new EventImpl("", cacheManager, Type.VIEW_CHANGED, newMembers,
+                            oldMembers, new TestAddress(1), 99)
+
+      detector.detectCrashedMember(e)
+
+      assertTrue(cache.containsKey(new TestAddress(1)))
+      assertTrue(cache.containsKey(new TestAddress(2)))
+   }
+
+}