cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jasobr...@apache.org
Subject cassandra git commit: Properly close StreamCompressionInputStream to release any ByteBuffer
Date Tue, 03 Oct 2017 19:23:38 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk d97d95ff1 -> 982ab93a2


Properly close StreamCompressionInputStream to release any ByteBuffer

patch by jasobrown; reviewed by Ariel Weisberg for CASSANDRA-13906


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/982ab93a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/982ab93a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/982ab93a

Branch: refs/heads/trunk
Commit: 982ab93a2f8a0f5c56af9378f65d3e9e430000b9
Parents: d97d95f
Author: Jason Brown <jasedbrown@gmail.com>
Authored: Tue Sep 26 15:52:54 2017 -0700
Committer: Jason Brown <jasedbrown@gmail.com>
Committed: Wed Oct 4 04:21:20 2017 +0900

----------------------------------------------------------------------
 CHANGES.txt                                        |  1 +
 .../apache/cassandra/streaming/StreamReader.java   | 11 ++++++-----
 .../streaming/compress/CompressedInputStream.java  |  4 +++-
 .../streaming/compress/CompressedStreamReader.java | 17 ++++++-----------
 .../compress/StreamCompressionInputStream.java     |  7 ++++++-
 5 files changed, 22 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/982ab93a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 2498270..5a8ab47 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.0
+ * Properly close StreamCompressionInputStream to release any ByteBuf (CASSANDRA-13906)
  * Add SERIAL and LOCAL_SERIAL support for cassandra-stress (CASSANDRA-13925)
  * LCS needlessly checks for L0 STCS candidates multiple times (CASSANDRA-12961)
  * Correctly close netty channels when a stream session ends (CASSANDRA-13905)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/982ab93a/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 590ba5f..f4eb9c4 100644
--- a/src/java/org/apache/cassandra/streaming/StreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/StreamReader.java
@@ -106,12 +106,12 @@ public class StreamReader
                      session.planId(), fileSeqNum, session.peer, repairedAt, totalSize, cfs.keyspace.getName(),
                      cfs.getTableName(), pendingRepair);
 
-
-        TrackedDataInputPlus in = new TrackedDataInputPlus(new StreamCompressionInputStream(inputPlus,
StreamMessage.CURRENT_VERSION));
-        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion,
getHeader(cfs.metadata()));
+        StreamDeserializer deserializer = null;
         SSTableMultiWriter writer = null;
-        try
+        try (StreamCompressionInputStream streamCompressionInputStream = new StreamCompressionInputStream(inputPlus,
StreamMessage.CURRENT_VERSION))
         {
+            TrackedDataInputPlus in = new TrackedDataInputPlus(streamCompressionInputStream);
+            deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion, getHeader(cfs.metadata()));
             writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, format);
             while (in.getBytesRead() < totalSize)
             {
@@ -125,8 +125,9 @@ public class StreamReader
         }
         catch (Throwable e)
         {
+            Object partitionKey = deserializer != null ? deserializer.partitionKey() : "";
             logger.warn("[Stream {}] Error while reading partition {} from stream on ks='{}'
and table='{}'.",
-                        session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(),
cfs.getTableName(), e);
+                        session.planId(), partitionKey, cfs.keyspace.getName(), cfs.getTableName(),
e);
             if (writer != null)
             {
                 writer.abort(e);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/982ab93a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
index 76f76ea..4b9fc61 100644
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
@@ -44,7 +44,7 @@ import org.apache.cassandra.utils.WrappedRunnable;
  * InputStream which reads data from underlining source with given {@link CompressionInfo}.
Uses {@link #buffer} as a buffer
  * for uncompressed data (which is read by stream consumers - {@link StreamDeserializer}
in this case).
  */
-public class CompressedInputStream extends RebufferingInputStream
+public class CompressedInputStream extends RebufferingInputStream implements AutoCloseable
 {
 
     private static final Logger logger = LoggerFactory.getLogger(CompressedInputStream.class);
@@ -200,6 +200,8 @@ public class CompressedInputStream extends RebufferingInputStream
     }
 
     /**
+     * {@inheritDoc}
+     *
      * Releases the resources specific to this instance, but not the {@link DataInputPlus}
that is used by the {@link Reader}.
      */
     @Override

http://git-wip-us.apache.org/repos/asf/cassandra/blob/982ab93a/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 e40788b..bd44209 100644
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
@@ -75,14 +75,12 @@ public class CompressedStreamReader extends StreamReader
                      session.planId(), fileSeqNum, session.peer, repairedAt, totalSize, cfs.keyspace.getName(),
pendingRepair,
                      cfs.getTableName());
 
-        CompressedInputStream cis = new CompressedInputStream(inputPlus, compressionInfo,
-                                                              ChecksumType.CRC32, cfs::getCrcCheckChance);
-        TrackedDataInputPlus in = new TrackedDataInputPlus(cis);
-
-        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion,
getHeader(cfs.metadata()));
+        StreamDeserializer deserializer = null;
         SSTableMultiWriter writer = null;
-        try
+        try (CompressedInputStream cis = new CompressedInputStream(inputPlus, compressionInfo,
ChecksumType.CRC32, cfs::getCrcCheckChance))
         {
+            TrackedDataInputPlus in = new TrackedDataInputPlus(cis);
+            deserializer = new StreamDeserializer(cfs.metadata(), in, inputVersion, getHeader(cfs.metadata()));
             writer = createWriter(cfs, totalSize, repairedAt, pendingRepair, format);
             String filename = writer.getFilename();
             int sectionIdx = 0;
@@ -109,8 +107,9 @@ public class CompressedStreamReader extends StreamReader
         }
         catch (Throwable e)
         {
+            Object partitionKey = deserializer != null ? deserializer.partitionKey() : "";
             logger.warn("[Stream {}] Error while reading partition {} from stream on ks='{}'
and table='{}'.",
-                        session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(),
cfs.getTableName());
+                        session.planId(), partitionKey, cfs.keyspace.getName(), cfs.getTableName());
             if (writer != null)
             {
                 writer.abort(e);
@@ -119,10 +118,6 @@ public class CompressedStreamReader extends StreamReader
                 throw e;
             throw Throwables.propagate(e);
         }
-        finally
-        {
-            cis.close();
-        }
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/cassandra/blob/982ab93a/src/java/org/apache/cassandra/streaming/compress/StreamCompressionInputStream.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/compress/StreamCompressionInputStream.java
b/src/java/org/apache/cassandra/streaming/compress/StreamCompressionInputStream.java
index 4b1459d..daf6d28 100644
--- a/src/java/org/apache/cassandra/streaming/compress/StreamCompressionInputStream.java
+++ b/src/java/org/apache/cassandra/streaming/compress/StreamCompressionInputStream.java
@@ -31,7 +31,7 @@ import org.apache.cassandra.io.util.RebufferingInputStream;
 import org.apache.cassandra.net.async.RebufferingByteBufDataInputPlus;
 import org.apache.cassandra.streaming.async.StreamCompressionSerializer;
 
-public class StreamCompressionInputStream extends RebufferingInputStream
+public class StreamCompressionInputStream extends RebufferingInputStream implements AutoCloseable
 {
     /**
      * The stream which contains buffers of compressed data that came from the peer.
@@ -70,6 +70,11 @@ public class StreamCompressionInputStream extends RebufferingInputStream
         buffer = currentBuf.nioBuffer(0, currentBuf.readableBytes());
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * Close resources except {@link #dataInputPlus} as that needs to remain open for other
streaming activity.
+     */
     @Override
     public void close()
     {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message