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

merge from 1.0.0 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1170264 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Sylvain Lebresne 2011-09-13 17:34:08 +00:00
parent 8165671565
commit acf93f044b
2 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -58,7 +58,7 @@
* generate hints for replicas that timeout, not just replicas that are known
to be down before starting (CASSANDRA-2034)
* Add throttling for internode streaming (CASSANDRA-3080)
* make the repair of a range repair all replica (CASSANDRA-2610)
* make the repair of a range repair all replica (CASSANDRA-2610, 3194)
* expose the ability to repair the first range (as returned by the
partitioner) of a node (CASSANDRA-2606)
* Streams Compression (CASSANDRA-3015)

12
src/java/org/apache/cassandra/streaming/StreamingRepairTask.java
View File

@ -78,7 +78,7 @@ public class StreamingRepairTask implements Runnable
public static StreamingRepairTask create(InetAddress ep1, InetAddress ep2, String tableName, String cfName, Collection<Range> ranges, Runnable callback)
{
InetAddress local = FBUtilities.getLocalAddress();
InetAddress local = FBUtilities.getBroadcastAddress();
UUID id = UUIDGen.makeType1UUIDFromHost(local);
// We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
InetAddress src = ep2.equals(local) ? ep2 : ep1;
@ -99,7 +99,7 @@ public class StreamingRepairTask implements Runnable
public void run()
{
if (src.equals(FBUtilities.getLocalAddress()))
if (src.equals(FBUtilities.getBroadcastAddress()))
{
initiateStreaming();
}
@ -205,7 +205,7 @@ public class StreamingRepairTask implements Runnable
throw new IOError(e);
}
assert task.src.equals(FBUtilities.getLocalAddress());
assert task.src.equals(FBUtilities.getBroadcastAddress());
assert task.owner.equals(message.getFrom());
logger.info(String.format("[streaming task #%s] Received task from %s to stream %d ranges to %s", task.id, message.getFrom(), task.ranges.size(), task.dst));
@ -219,7 +219,7 @@ public class StreamingRepairTask implements Runnable
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
StreamingRepairTask.serializer.serialize(task, dos, version);
Message msg = new Message(FBUtilities.getLocalAddress(), StorageService.Verb.STREAMING_REPAIR_REQUEST, bos.toByteArray(), version);
Message msg = new Message(FBUtilities.getBroadcastAddress(), StorageService.Verb.STREAMING_REPAIR_REQUEST, bos.toByteArray(), version);
MessagingService.instance().sendOneWay(msg, task.src);
}
}
@ -248,7 +248,7 @@ public class StreamingRepairTask implements Runnable
return;
}
assert task.owner.equals(FBUtilities.getLocalAddress());
assert task.owner.equals(FBUtilities.getBroadcastAddress());
logger.info(String.format("[streaming task #%s] task succeeded", task.id));
if (task.callback != null)
@ -262,7 +262,7 @@ public class StreamingRepairTask implements Runnable
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
UUIDGen.write(taskid, dos);
Message msg = new Message(FBUtilities.getLocalAddress(), StorageService.Verb.STREAMING_REPAIR_RESPONSE, bos.toByteArray(), version);
Message msg = new Message(FBUtilities.getBroadcastAddress(), StorageService.Verb.STREAMING_REPAIR_RESPONSE, bos.toByteArray(), version);
MessagingService.instance().sendOneWay(msg, remote);
}
}