natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix for when bootstrap source has no data in the range requested 

patch by jbellis; reviewed by Jaakko Laine for CASSANDRA-541

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@835891 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-11-13 15:57:34 +00:00
parent f3cdaa96b7
commit e734d837a0
1 changed files with 20 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/java/org/apache/cassandra/io/Streaming.java
View File

@ -69,9 +69,6 @@ public class Streaming
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 @@ public class Streaming
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 @@ public class Streaming
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 @@ public class Streaming
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);
}
}