cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [16/21] git commit: RangeStreamer throws when any range did not complete. Patch by brandonwilliams reviewed by yukim for CASSANDRA-5009
Date Mon, 03 Dec 2012 20:32:58 GMT
RangeStreamer throws when any range did not complete.
Patch by brandonwilliams reviewed by yukim for CASSANDRA-5009


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/d1d37203
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/d1d37203
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/d1d37203

Branch: refs/heads/cassandra-1.2
Commit: d1d372033b2f0a2ef0a63c9591b369d7789851db
Parents: 3706749
Author: Brandon Williams <brandonwilliams@apache.org>
Authored: Mon Dec 3 14:21:23 2012 -0600
Committer: Brandon Williams <brandonwilliams@apache.org>
Committed: Mon Dec 3 14:21:23 2012 -0600

----------------------------------------------------------------------
 .../org/apache/cassandra/dht/RangeStreamer.java    |   23 +++++++++++----
 1 files changed, 17 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/d1d37203/src/java/org/apache/cassandra/dht/RangeStreamer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/RangeStreamer.java b/src/java/org/apache/cassandra/dht/RangeStreamer.java
index 4e5cfb8..a0e1a93 100644
--- a/src/java/org/apache/cassandra/dht/RangeStreamer.java
+++ b/src/java/org/apache/cassandra/dht/RangeStreamer.java
@@ -1,4 +1,4 @@
-/**
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -19,12 +19,14 @@ package org.apache.cassandra.dht;
 
 import java.net.InetAddress;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
 import org.apache.cassandra.streaming.IStreamCallback;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,7 +40,6 @@ import org.apache.cassandra.locator.IEndpointSnitch;
 import org.apache.cassandra.locator.TokenMetadata;
 import org.apache.cassandra.streaming.OperationType;
 import org.apache.cassandra.streaming.StreamIn;
-import org.apache.cassandra.utils.FBUtilities;
 
 /**
  * Assists in streaming ranges to a node.
@@ -52,6 +53,9 @@ public class RangeStreamer
     private final OperationType opType;
     private final Multimap<String, Map.Entry<InetAddress, Collection<Range<Token>>>>
toFetch = HashMultimap.create();
     private final Set<ISourceFilter> sourceFilters = new HashSet<ISourceFilter>();
+    // protected for testing.
+    protected CountDownLatch latch;
+    private Set<Range<Token>> completed = Collections.newSetFromMap(new ConcurrentHashMap<Range<Token>,
Boolean>());
 
     /**
      * A filter applied to sources to stream from when constructing a fetch map.
@@ -140,7 +144,7 @@ public class RangeStreamer
     private Multimap<Range<Token>, InetAddress> getAllRangesWithSourcesFor(String
table, Collection<Range<Token>> desiredRanges)
     {
         AbstractReplicationStrategy strat = Table.open(table).getReplicationStrategy();
-        Multimap<Range<Token>, InetAddress> rangeAddresses = strat.getRangeAddresses(metadata);
+        Multimap<Range<Token>, InetAddress> rangeAddresses = strat.getRangeAddresses(metadata.cloneOnlyTokenMap());
 
         Multimap<Range<Token>, InetAddress> rangeSources = ArrayListMultimap.create();
         for (Range<Token> desiredRange : desiredRanges)
@@ -217,17 +221,19 @@ public class RangeStreamer
 
     public void fetch()
     {
-        final CountDownLatch latch = new CountDownLatch(toFetch().entries().size());
+        latch = new CountDownLatch(toFetch.entries().size());
+
         for (Map.Entry<String, Map.Entry<InetAddress, Collection<Range<Token>>>>
entry : toFetch.entries())
         {
             final String table = entry.getKey();
             final InetAddress source = entry.getValue().getKey();
-            Collection<Range<Token>> ranges = entry.getValue().getValue();
+            final Collection<Range<Token>> ranges = entry.getValue().getValue();
             /* Send messages to respective folks to stream data over to me */
             IStreamCallback callback = new IStreamCallback()
             {
                 public void onSuccess()
                 {
+                    completed.addAll(ranges);
                     latch.countDown();
                     if (logger.isDebugEnabled())
                         logger.debug(String.format("Removed %s/%s as a %s source; remaining
is %s",
@@ -236,8 +242,8 @@ public class RangeStreamer
 
                 public void onFailure()
                 {
+                    latch.countDown();
                     logger.warn("Streaming from " + source + " failed");
-                    onSuccess(); // calling onSuccess for latch countdown
                 }
             };
             if (logger.isDebugEnabled())
@@ -248,6 +254,11 @@ public class RangeStreamer
         try
         {
             latch.await();
+            for (Map.Entry<String, Map.Entry<InetAddress, Collection<Range<Token>>>>
entry : toFetch.entries())
+            {
+                if (!completed.containsAll(entry.getValue().getValue()))
+                    throw new RuntimeException(String.format("Unable to fetch range %s for
keyspace %s from any hosts", entry.getValue().getValue(), entry.getKey()));
+            }
         }
         catch (InterruptedException e)
         {


Mime
View raw message