You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yu...@apache.org on 2015/11/03 17:08:01 UTC
[02/10] cassandra git commit: Fix streaming to catch exception so
retry not fail
Fix streaming to catch exception so retry not fail
patch by yukim; reviewed by Paulo Motta for CASSANDRA-10557
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/068614cc
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/068614cc
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/068614cc
Branch: refs/heads/cassandra-2.2
Commit: 068614ccc7ba6c5b8ccb50a0840af57bb45b4b36
Parents: 986a1a7
Author: Yuki Morishita <yu...@apache.org>
Authored: Tue Oct 20 14:25:55 2015 -0500
Committer: Yuki Morishita <yu...@apache.org>
Committed: Tue Nov 3 09:33:37 2015 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/streaming/StreamReader.java | 16 ++++++++++++++--
.../streaming/compress/CompressedStreamReader.java | 17 ++++++++++++++---
3 files changed, 29 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/068614cc/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 3d22b91..4c24b35 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.1.12
+ * Fix streaming to catch exception so retry not fail (CASSANDRA-10557)
* Add validation method to PerRowSecondaryIndex (CASSANDRA-10092)
* Support encrypted and plain traffic on the same port (CASSANDRA-10559)
* Do STCS in DTCS windows (CASSANDRA-10276)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/068614cc/src/java/org/apache/cassandra/streaming/StreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamReader.java b/src/java/org/apache/cassandra/streaming/StreamReader.java
index c96a925..5389a80 100644
--- a/src/java/org/apache/cassandra/streaming/StreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/StreamReader.java
@@ -89,11 +89,12 @@ public class StreamReader
}
ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
- SSTableWriter writer = createWriter(cfs, totalSize, repairedAt);
DataInputStream dis = new DataInputStream(new LZFInputStream(Channels.newInputStream(channel)));
BytesReadTracker in = new BytesReadTracker(dis);
+ SSTableWriter writer = null;
try
{
+ writer = createWriter(cfs, totalSize, repairedAt);
while (in.getBytesRead() < totalSize)
{
writeRow(writer, in, cfs);
@@ -104,7 +105,18 @@ public class StreamReader
}
catch (Throwable e)
{
- writer.abort();
+ if (writer != null)
+ {
+ try
+ {
+ writer.abort();
+ }
+ catch (Throwable e2)
+ {
+ // add abort error to original and continue so we can drain unread stream
+ e.addSuppressed(e2);
+ }
+ }
drain(dis, in.getBytesRead());
if (e instanceof IOException)
throw (IOException) e;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/068614cc/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
index fb2599f..0529496 100644
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
@@ -72,12 +72,12 @@ public class CompressedStreamReader extends StreamReader
}
ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
- SSTableWriter writer = createWriter(cfs, totalSize, repairedAt);
-
CompressedInputStream cis = new CompressedInputStream(Channels.newInputStream(channel), compressionInfo, inputVersion.hasPostCompressionAdlerChecksums);
BytesReadTracker in = new BytesReadTracker(new DataInputStream(cis));
+ SSTableWriter writer = null;
try
{
+ writer = createWriter(cfs, totalSize, repairedAt);
for (Pair<Long, Long> section : sections)
{
long length = section.right - section.left;
@@ -95,7 +95,18 @@ public class CompressedStreamReader extends StreamReader
}
catch (Throwable e)
{
- writer.abort();
+ if (writer != null)
+ {
+ try
+ {
+ writer.abort();
+ }
+ catch (Throwable e2)
+ {
+ // add abort error to original and continue so we can drain unread stream
+ e.addSuppressed(e2);
+ }
+ }
drain(cis, in.getBytesRead());
if (e instanceof IOException)
throw (IOException) e;