svn commit: r1170306 - in /cassandra/branches/cassandra-0.8: CHANGES.txt src/java/org/apache/cassandra/service/StorageService.java

[jb...@apache.org]

Author: jbellis
Date: Tue Sep 13 19:22:05 2011
New Revision: 1170306

URL: http://svn.apache.org/viewvc?rev=1170306&view=rev
Log:
Avoid streaming data to nodes that already have it, on move as well as decommission
patch by Nick Bailey and jbellis for CASSANDRA-3041

Modified:
    cassandra/branches/cassandra-0.8/CHANGES.txt
    cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/StorageService.java

Modified: cassandra/branches/cassandra-0.8/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/CHANGES.txt?rev=1170306&r1=1170305&r2=1170306&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.8/CHANGES.txt Tue Sep 13 19:22:05 2011
@@ -13,6 +13,8 @@
    (CASSANDRA-3158)
  * Randomize choice of first replica for counter increment (CASSANDRA-2890)
  * Fix using read_repair_chance instead of merge_shard_change (CASSANDRA-3202)
+ * Avoid streaming data to nodes that already have it, on move as well as
+   decommission (CASSANDRA-3041)
 
 
 0.8.5

Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/StorageService.java?rev=1170306&r1=1170305&r2=1170306&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/StorageService.java Tue Sep 13 19:22:05 2011
@@ -30,9 +30,7 @@ import java.util.concurrent.*;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.Multimap;
+import com.google.common.collect.*;
 import org.apache.log4j.Level;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -46,6 +44,7 @@ import org.apache.cassandra.config.Confi
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.KSMetaData;
 import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.Table;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.dht.*;
 import org.apache.cassandra.gms.*;
@@ -1049,10 +1048,9 @@ public class StorageService implements I
         // all leaving nodes are gone.
         for (Range range : affectedRanges)
         {
-            Collection<InetAddress> currentEndpoints = strategy.calculateNaturalEndpoints(range.right, tm);
-            Collection<InetAddress> newEndpoints = strategy.calculateNaturalEndpoints(range.right, allLeftMetadata);
-            newEndpoints.removeAll(currentEndpoints);
-            pendingRanges.putAll(range, newEndpoints);
+            Set<InetAddress> currentEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(range.right, tm));
+            Set<InetAddress> newEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(range.right, allLeftMetadata));
+            pendingRanges.putAll(range, Sets.difference(newEndpoints, currentEndpoints));
         }
 
         // At this stage pendingRanges has been updated according to leave operations. We can
@@ -2025,8 +2023,9 @@ public class StorageService implements I
 
             for (Range toStream : rangesPerTable.left)
             {
-                List<InetAddress> endpoints = strategy.calculateNaturalEndpoints(toStream.right, tokenMetaClone);
-                rangeWithEndpoints.putAll(toStream, endpoints);
+                Set<InetAddress> currentEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(toStream.right, tokenMetadata_));
+                Set<InetAddress> newEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(toStream.right, tokenMetaClone));
+                rangeWithEndpoints.putAll(toStream, Sets.difference(newEndpoints, currentEndpoints));
             }
 
             // associating table with range-to-endpoints map