svn commit: r1021806 - in /cassandra/trunk: CHANGES.txt src/java/org/apache/cassandra/service/StorageService.java

[jb...@apache.org]

Author: jbellis
Date: Tue Oct 12 15:51:42 2010
New Revision: 1021806

URL: http://svn.apache.org/viewvc?rev=1021806&view=rev
Log:
avoid blocking for dead nodes during removetoken.  patch by Nick Bailey; reviewed by jbellis for CASSANDRA-1605

Modified:
    cassandra/trunk/CHANGES.txt
    cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java

Modified: cassandra/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=1021806&r1=1021805&r2=1021806&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Tue Oct 12 15:51:42 2010
@@ -27,6 +27,7 @@ dev
  * add cli support for column metadata (CASSANDRA-1578)
  * add cli support for schema modification (CASSANDRA-1584)
  * delete temp files on failed compactions (CASSANDRA-1596)
+ * avoid blocking for dead nodes during removetoken (CASSANDRA-1605)
 
 
 0.7-beta2

Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=1021806&r1=1021805&r2=1021806&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Tue Oct 12 15:51:42 2010
@@ -884,27 +884,6 @@ public class StorageService implements I
     }
 
     /**
-     * Determines the endpoints that are going to become responsible for data due to
-     * a node leaving the cluster.
-     *
-     * @param endpoint the node that is leaving the cluster
-     * @return A set of endpoints
-     */
-    private Set<InetAddress> getNewEndpoints(InetAddress endpoint)
-    {
-        Set<InetAddress> newEndpoints = new HashSet<InetAddress>();
-
-        for (String table : DatabaseDescriptor.getNonSystemTables())
-        {
-            // get all ranges that change ownership (that is, a node needs
-            // to take responsibility for new range)
-            Multimap<Range, InetAddress> changedRanges = getChangedRangesForLeaving(table, endpoint);
-            newEndpoints.addAll(changedRanges.values());
-        }
-        return newEndpoints;
-    }
-
-    /**
      * Finds living endpoints responsible for the given ranges
      *
      * @param table the table ranges belong to
@@ -1759,12 +1738,26 @@ public class StorageService implements I
         if (tokenMetadata_.isLeaving(endpoint)) 
             throw new UnsupportedOperationException("Node " + endpoint + " is already being removed.");
 
-        if (replicatingNodes != null || replicateLatch != null)
+        if (this.replicatingNodes != null || replicateLatch != null)
             throw new UnsupportedOperationException("This node is already processing a removal. Wait for it to complete.");
 
         // Find the endpoints that are going to become responsible for data
-        replicatingNodes = Collections.synchronizedSet(getNewEndpoints(endpoint));
-        replicateLatch = new CountDownLatch(replicatingNodes.size());
+        replicatingNodes = Collections.synchronizedSet(new HashSet<InetAddress>());
+        for (String table : DatabaseDescriptor.getNonSystemTables())
+        {
+            // get all ranges that change ownership (that is, a node needs
+            // to take responsibility for new range)
+            Multimap<Range, InetAddress> changedRanges = getChangedRangesForLeaving(table, endpoint);
+            IFailureDetector failureDetector = FailureDetector.instance;
+            for (InetAddress ep : changedRanges.values())
+            {
+                if (failureDetector.isAlive(ep))
+                    replicatingNodes.add(ep);
+                else
+                    logger_.warn("Endpoint " + ep + " is down and will not receive data for re-replication of " + endpoint);
+            }
+        }
+        replicateLatch = new CountDownLatch(this.replicatingNodes.size());
         removingNode = endpoint;
 
         tokenMetadata_.addLeavingEndpoint(endpoint);
@@ -1793,10 +1786,10 @@ public class StorageService implements I
         calculatePendingRanges();
         Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.removedNonlocal(localToken, token));
 
-        if(!replicatingNodes.isEmpty())
-            logger_.error("Failed to recieve removal confirmation for " + StringUtils.join(replicatingNodes, ","));
+        if(!this.replicatingNodes.isEmpty())
+            logger_.error("Failed to recieve removal confirmation for " + StringUtils.join(this.replicatingNodes, ","));
 
-        replicatingNodes = null;
+        this.replicatingNodes = null;
         removingNode = null;
         replicateLatch = null;
     }