diff --git a/CHANGES.txt b/CHANGES.txt index 146a0cefe1..d11be2630f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -12,6 +12,7 @@ * Deprecate Pig support (CASSANDRA-10542) * Reduce contention getting instances of CompositeType (CASSANDRA-10433) Merged from 2.1: + * Fix CompressedInputStream for proper cleanup (CASSANDRA-10012) * (cqlsh) Support counters in COPY commands (CASSANDRA-9043) * Try next replica if not possible to connect to primary replica on ColumnFamilyRecordReader (CASSANDRA-2388) diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java index 872afcd0a7..daa339a312 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java @@ -43,7 +43,7 @@ public class CompressedInputStream extends InputStream private final BlockingQueue dataBuffer; // uncompressed bytes - private byte[] buffer; + private final byte[] buffer; // offset from the beginning of the buffer protected long bufferOffset = 0; @@ -61,6 +61,8 @@ public class CompressedInputStream extends InputStream private long totalCompressedBytesRead; + private Thread readerThread; + /** * @param source Input source to read compressed data from * @param info Compression info @@ -71,9 +73,10 @@ public class CompressedInputStream extends InputStream this.checksum = new Adler32(); this.buffer = new byte[info.parameters.chunkLength()]; // buffer is limited to store up to 1024 chunks - this.dataBuffer = new ArrayBlockingQueue(Math.min(info.chunks.length, 1024)); + this.dataBuffer = new ArrayBlockingQueue<>(Math.min(info.chunks.length, 1024)); - new Thread(new Reader(source, info, dataBuffer)).start(); + readerThread = new Thread(new Reader(source, info, dataBuffer)); + readerThread.start(); } public int read() throws IOException @@ -132,7 +135,7 @@ public class CompressedInputStream extends InputStream return totalCompressedBytesRead; } - static class Reader extends WrappedRunnable + class Reader extends WrappedRunnable { private final InputStream source; private final Iterator chunks; @@ -148,7 +151,7 @@ public class CompressedInputStream extends InputStream protected void runMayThrow() throws Exception { byte[] compressedWithCRC; - while (chunks.hasNext()) + while (!Thread.currentThread().isInterrupted() && chunks.hasNext()) { CompressionMetadata.Chunk chunk = chunks.next(); @@ -158,16 +161,43 @@ public class CompressedInputStream extends InputStream int bufferRead = 0; while (bufferRead < readLength) { - int r = source.read(compressedWithCRC, bufferRead, readLength - bufferRead); - if (r < 0) + int r; + try + { + r = source.read(compressedWithCRC, bufferRead, readLength - bufferRead); + if (r < 0) + { + dataBuffer.put(POISON_PILL); + return; // throw exception where we consume dataBuffer + } + } + catch (IOException e) { dataBuffer.put(POISON_PILL); - return; // throw exception where we consume dataBuffer + throw e; } bufferRead += r; } dataBuffer.put(compressedWithCRC); } + synchronized(CompressedInputStream.this) + { + readerThread = null; + } } } + + @Override + public void close() throws IOException + { + synchronized(this) + { + if (readerThread != null) + { + readerThread.interrupt(); + readerThread = null; + } + } + } + } diff --git a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java index facb906c63..22779432f4 100644 --- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java +++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java @@ -120,6 +120,10 @@ public class CompressedStreamReader extends StreamReader else throw Throwables.propagate(e); } + finally + { + cis.close(); + } } @Override diff --git a/test/unit/org/apache/cassandra/streaming/compress/CompressedInputStreamTest.java b/test/unit/org/apache/cassandra/streaming/compress/CompressedInputStreamTest.java index 0214c76250..e692441570 100644 --- a/test/unit/org/apache/cassandra/streaming/compress/CompressedInputStreamTest.java +++ b/test/unit/org/apache/cassandra/streaming/compress/CompressedInputStreamTest.java @@ -17,12 +17,10 @@ */ package org.apache.cassandra.streaming.compress; -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; -import java.io.EOFException; -import java.io.File; -import java.io.RandomAccessFile; +import java.io.*; import java.util.*; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.TimeUnit; import org.junit.Test; @@ -58,6 +56,53 @@ public class CompressedInputStreamTest { testCompressedReadWith(new long[]{1L, 122L, 123L, 124L, 456L}, true); } + + /** + * Test CompressedInputStream not hang when closed while reading + * @throws IOException + */ + @Test(expected = EOFException.class) + public void testClose() throws IOException + { + CompressionParameters param = new CompressionParameters(SnappyCompressor.instance, 32, Collections.emptyMap()); + CompressionMetadata.Chunk[] chunks = {new CompressionMetadata.Chunk(0, 100)}; + final SynchronousQueue blocker = new SynchronousQueue<>(); + InputStream blockingInput = new InputStream() + { + @Override + public int read() throws IOException + { + try + { + // 10 second cut off not to stop other test in case + return Objects.requireNonNull(blocker.poll(10, TimeUnit.SECONDS)); + } + catch (InterruptedException e) + { + throw new IOException("Interrupted as expected", e); + } + } + }; + CompressionInfo info = new CompressionInfo(chunks, param); + try (CompressedInputStream cis = new CompressedInputStream(blockingInput, info)) + { + new Thread(new Runnable() + { + @Override + public void run() + { + try + { + cis.close(); + } + catch (Exception ignore) {} + } + }).start(); + // block here + cis.read(); + } + } + /** * @param valuesToCheck array of longs of range(0-999) * @throws Exception @@ -70,18 +115,20 @@ public class CompressedInputStreamTest File tmp = new File(File.createTempFile("cassandra", "unittest").getParent(), "ks-cf-ib-1-Data.db"); Descriptor desc = Descriptor.fromFilename(tmp.getAbsolutePath()); MetadataCollector collector = new MetadataCollector(new SimpleDenseCellNameType(BytesType.instance)); - CompressionParameters param = new CompressionParameters(SnappyCompressor.instance, 32, Collections.EMPTY_MAP); - CompressedSequentialWriter writer = new CompressedSequentialWriter(tmp, desc.filenameFor(Component.COMPRESSION_INFO), param, collector); - Map index = new HashMap(); - for (long l = 0L; l < 1000; l++) + CompressionParameters param = new CompressionParameters(SnappyCompressor.instance, 32, Collections.emptyMap()); + Map index = new HashMap<>(); + try (CompressedSequentialWriter writer = new CompressedSequentialWriter(tmp, desc.filenameFor(Component.COMPRESSION_INFO), param, collector)) { - index.put(l, writer.getFilePointer()); - writer.stream.writeLong(l); + for (long l = 0L; l < 1000; l++) + { + index.put(l, writer.getFilePointer()); + writer.stream.writeLong(l); + } + writer.finish(); } - writer.finish(); CompressionMetadata comp = CompressionMetadata.create(tmp.getAbsolutePath()); - List> sections = new ArrayList>(); + List> sections = new ArrayList<>(); for (long l : valuesToCheck) { long position = index.get(l); @@ -100,14 +147,15 @@ public class CompressedInputStreamTest size += (c.length + 4); // 4bytes CRC byte[] toRead = new byte[size]; - RandomAccessFile f = new RandomAccessFile(tmp, "r"); - int pos = 0; - for (CompressionMetadata.Chunk c : chunks) + try (RandomAccessFile f = new RandomAccessFile(tmp, "r")) { - f.seek(c.offset); - pos += f.read(toRead, pos, c.length + 4); + int pos = 0; + for (CompressionMetadata.Chunk c : chunks) + { + f.seek(c.offset); + pos += f.read(toRead, pos, c.length + 4); + } } - f.close(); if (testTruncate) { @@ -119,13 +167,15 @@ public class CompressedInputStreamTest // read buffer using CompressedInputStream CompressionInfo info = new CompressionInfo(chunks, param); CompressedInputStream input = new CompressedInputStream(new ByteArrayInputStream(toRead), info); - DataInputStream in = new DataInputStream(input); - for (int i = 0; i < sections.size(); i++) + try (DataInputStream in = new DataInputStream(input)) { - input.position(sections.get(i).left); - long readValue = in.readLong(); - assert readValue == valuesToCheck[i] : "expected " + valuesToCheck[i] + " but was " + readValue; + for (int i = 0; i < sections.size(); i++) + { + input.position(sections.get(i).left); + long readValue = in.readLong(); + assertEquals("expected " + valuesToCheck[i] + " but was " + readValue, valuesToCheck[i], readValue); + } } } }