svn commit: r835891 - /incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java

[jbellis]

Author: jbellis
Date: Fri Nov 13 15:57:34 2009
New Revision: 835891

URL: http://svn.apache.org/viewvc?rev=835891&view=rev
Log:
fix for when bootstrap source has no data in the range requested
patch by jbellis; reviewed by Jaakko Laine for CASSANDRA-541

Modified:
    incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java

Modified: 
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java
URL: 
http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java?rev=835891&r1=835890&r2=835891&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java 
(original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/Streaming.java 
Fri Nov 13 15:57:34 2009
@@ -69,9 +69,6 @@
 
     private static void transferOneTable(InetAddress target, 
List<SSTableReader> sstables, String table) throws IOException
     {
-        if (sstables.isEmpty())
-            return;
-
         StreamContextManager.StreamContext[] streamContexts = new 
StreamContextManager.StreamContext[SSTable.FILES_ON_DISK * sstables.size()];
         int i = 0;
         for (SSTableReader sstable : sstables)
@@ -91,12 +88,16 @@
         if (logger.isDebugEnabled())
           logger.debug("Sending a stream initiate message to " + target + " 
...");
         MessagingService.instance().sendOneWay(message, target);
-        if (logger.isDebugEnabled())
-          logger.debug("Waiting for transfer to " + target + " to complete");
-        StreamManager.instance(target).waitForStreamCompletion();
-        // reference sstables one more time to make sure it doesn't get GC'd 
early (causing delete of its files)
-        if (logger.isDebugEnabled())
-            logger.debug("Done with transfer to " + target + " of " + 
StringUtils.join(sstables, ", "));
+
+        if (streamContexts.length > 0)
+        {
+            if (logger.isDebugEnabled())
+              logger.debug("Waiting for transfer to " + target + " to 
complete");
+            StreamManager.instance(target).waitForStreamCompletion();
+            // reference sstables one more time to make sure it doesn't get 
GC'd early (causing delete of its files)
+            if (logger.isDebugEnabled())
+                logger.debug("Done with transfer to " + target + " of " + 
StringUtils.join(sstables, ", "));
+        }
     }
 
     public static class StreamInitiateVerbHandler implements IVerbHandler
@@ -119,6 +120,14 @@
                 StreamInitiateMessage biMsg = 
StreamInitiateMessage.serializer().deserialize(bufIn);
                 StreamContextManager.StreamContext[] streamContexts = 
biMsg.getStreamContext();
 
+                if (streamContexts.length == 0 && 
StorageService.instance().isBootstrapMode())
+                {
+                    if (logger.isDebugEnabled())
+                        logger.debug("no data needed from " + 
message.getFrom());
+                    
StorageService.instance().removeBootstrapSource(message.getFrom());
+                    return;
+                }
+
                 Map<String, String> fileNames = getNewNames(streamContexts);
                 /*
                  * For each of stream context's in the incoming message
@@ -142,9 +151,9 @@
                 Message doneMessage = new 
Message(FBUtilities.getLocalAddress(), "", 
StorageService.streamInitiateDoneVerbHandler_, new byte[0] );
                 MessagingService.instance().sendOneWay(doneMessage, 
message.getFrom());
             }
-            catch ( IOException ex )
+            catch (IOException ex)
             {
-                logger.info(LogUtil.throwableToString(ex));
+                throw new IOError(ex);
             }
         }