Author: gdusbabek
Date: Fri Jun 11 17:00:40 2010
New Revision: 953770
URL: http://svn.apache.org/viewvc?rev=953770&view=rev
Log:
force AES to stream files in the stream stage. patch by gdusbabek, reviewed by
jbellis. CASSANDRA-1169
Modified:
cassandra/branches/cassandra-0.6/CHANGES.txt
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/AntiEntropyService.java
Modified: cassandra/branches/cassandra-0.6/CHANGES.txt
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/CHANGES.txt?rev=953770&r1=953769&r2=953770&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.6/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.6/CHANGES.txt Fri Jun 11 17:00:40 2010
@@ -20,6 +20,8 @@
* restructure the startup ordering of Gossiper and MessageService to avoid
timing anomalies (CASSANDRA-1160)
* detect incomplete commit log hearders (CASSANDRA-1119)
+ * force anti-entropy service to stream files on the stream stage to avoid
+ sending streams out of order (CASSANDRA-1169)
0.6.2
Modified:
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/AntiEntropyService.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/AntiEntropyService.java?rev=953770&r1=953769&r2=953770&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/AntiEntropyService.java
(original)
+++
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/service/AntiEntropyService.java
Fri Jun 11 17:00:40 2010
@@ -623,8 +623,15 @@ public class AntiEntropyService
try
{
List<Range> ranges = new ArrayList<Range>(differences);
- List<SSTableReader> sstables =
CompactionManager.instance.submitAnticompaction(cfstore, ranges, remote).get();
- StreamOut.transferSSTables(remote, sstables, cf.left);
+ final List<SSTableReader> sstables =
CompactionManager.instance.submitAnticompaction(cfstore, ranges, remote).get();
+ Future f =
StageManager.getStage(StageManager.STREAM_STAGE).submit(new WrappedRunnable()
+ {
+ protected void runMayThrow() throws Exception
+ {
+ StreamOut.transferSSTables(remote, sstables, cf.left);
+ }
+ });
+ f.get();
}
catch(Exception e)
{
