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

sstables from stalled repair sessions become live after a reboot and can resurrect deleted data 

patch by jasobrown, reviewed by yukim for CASSANDRA-6503
This commit is contained in:
Jason Brown 2014-01-30 09:51:42 -08:00
parent 852e27f2e8
commit 6ad995e8fa
3 changed files with 14 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -21,6 +21,7 @@
* Fix preparing with batch and delete from collection (CASSANDRA-6607)
* Fix ABSC reverse iterator's remove() method (CASSANDRA-6629)
* Handle host ID conflicts properly (CASSANDRA-6615)
* sstables from stalled repair sessions can resurrect deleted data (CASSANDRA-6503)
1.2.13

8
src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
View File

@ -119,8 +119,8 @@ public class IncomingStreamReader
DataInput dis = new DataInputStream(underliningStream);
try
{
SSTableReader reader = streamIn(dis, localFile, remoteFile);
session.finished(remoteFile, reader);
SSTableWriter writer = streamIn(dis, localFile, remoteFile);
session.finished(remoteFile, writer);
}
catch (IOException ex)
{
@ -141,7 +141,7 @@ public class IncomingStreamReader
/**
* @throws IOException if reading the remote sstable fails. Will throw an RTE if local write fails.
*/
private SSTableReader streamIn(DataInput input, PendingFile localFile, PendingFile remoteFile) throws IOException
private SSTableWriter streamIn(DataInput input, PendingFile localFile, PendingFile remoteFile) throws IOException
{
ColumnFamilyStore cfs = Table.open(localFile.desc.ksname).getColumnFamilyStore(localFile.desc.cfname);
DecoratedKey key;
@ -197,7 +197,7 @@ public class IncomingStreamReader
}
StreamingMetrics.totalIncomingBytes.inc(totalBytesRead);
metrics.incomingBytes.inc(totalBytesRead);
return writer.closeAndOpenReader();
return writer;
}
catch (Throwable e)
{

13
src/java/org/apache/cassandra/streaming/StreamInSession.java
View File

@ -24,6 +24,7 @@ import java.net.Socket;
import java.util.*;
import java.util.concurrent.ConcurrentMap;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.cliffc.high_scale_lib.NonBlockingHashSet;
import org.slf4j.Logger;
@ -47,7 +48,7 @@ public class StreamInSession extends AbstractStreamSession
private static final ConcurrentMap<UUID, StreamInSession> sessions = new NonBlockingHashMap<UUID, StreamInSession>();
private final Set<PendingFile> files = new NonBlockingHashSet<PendingFile>();
private final List<SSTableReader> readers = new ArrayList<SSTableReader>();
private final List<SSTableWriter> writers = new ArrayList<SSTableWriter>();
private PendingFile current;
private Socket socket;
private volatile int retries;
@ -106,13 +107,13 @@ public class StreamInSession extends AbstractStreamSession
}
}
public void finished(PendingFile remoteFile, SSTableReader reader) throws IOException
public void finished(PendingFile remoteFile, SSTableWriter writer) throws IOException
{
if (logger.isDebugEnabled())
logger.debug("Finished {} (from {}). Sending ack to {}", new Object[] {remoteFile, getHost(), this});
assert reader != null;
readers.add(reader);
assert writer != null;
writers.add(writer);
files.remove(remoteFile);
if (remoteFile.equals(current))
current = null;
@ -163,6 +164,10 @@ public class StreamInSession extends AbstractStreamSession
HashMap <ColumnFamilyStore, List<SSTableReader>> cfstores = new HashMap<ColumnFamilyStore, List<SSTableReader>>();
try
{
List<SSTableReader> readers = new ArrayList<SSTableReader>();
for(SSTableWriter writer : writers)
readers.add(writer.closeAndOpenReader());
for (SSTableReader sstable : readers)
{
assert sstable.getTableName().equals(table);