cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [10/15] cassandra git commit: Merge branch 'cassandra-2.2' into cassandra-3.0
Date Mon, 23 Nov 2015 20:00:48 GMT
Merge branch 'cassandra-2.2' into cassandra-3.0


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

Branch: refs/heads/trunk
Commit: 0b3cfae4e619d1ece5ff8afc774eeb52b93166d8
Parents: 9fe790d 056055f
Author: Yuki Morishita <yukim@apache.org>
Authored: Mon Nov 23 13:58:58 2015 -0600
Committer: Yuki Morishita <yukim@apache.org>
Committed: Mon Nov 23 13:58:58 2015 -0600

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../compress/CompressedInputStream.java         | 46 ++++++++--
 .../compress/CompressedStreamReader.java        |  4 +
 .../compression/CompressedInputStreamTest.java  | 88 ++++++++++++++++----
 4 files changed, 114 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/0b3cfae4/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 4182cc1,d11be26..608d8f8
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -11,7 -3,16 +11,8 @@@ Merged from 2.2
   * Fix SimpleDateType type compatibility (CASSANDRA-10027)
   * (Hadoop) fix splits calculation (CASSANDRA-10640)
   * (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058)
 - * (cqlsh) show partial trace if incomplete after max_trace_wait (CASSANDRA-7645)
 - * Use most up-to-date version of schema for system tables (CASSANDRA-10652)
 - * Deprecate memory_allocator in cassandra.yaml (CASSANDRA-10581,10628)
 - * Expose phi values from failure detector via JMX and tweak debug
 -   and trace logging (CASSANDRA-9526)
 - * Fix RangeNamesQueryPager (CASSANDRA-10509)
 - * 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)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0b3cfae4/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
index ccd0ac5,daa339a..56dc63a
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
@@@ -42,10 -41,9 +42,10 @@@ public class CompressedInputStream exte
      private final CompressionInfo info;
      // chunk buffer
      private final BlockingQueue<byte[]> dataBuffer;
 +    private final Supplier<Double> crcCheckChanceSupplier;
  
      // uncompressed bytes
-     private byte[] buffer;
+     private final byte[] buffer;
  
      // offset from the beginning of the buffer
      protected long bufferOffset = 0;
@@@ -67,16 -67,16 +69,17 @@@
       * @param source Input source to read compressed data from
       * @param info Compression info
       */
 -    public CompressedInputStream(InputStream source, CompressionInfo info)
 +    public CompressedInputStream(InputStream source, CompressionInfo info, ChecksumType
checksumType, Supplier<Double> crcCheckChanceSupplier)
      {
          this.info = info;
 -        this.checksum =  new Adler32();
 +        this.checksum =  checksumType.newInstance();
          this.buffer = new byte[info.parameters.chunkLength()];
          // buffer is limited to store up to 1024 chunks
-         this.dataBuffer = new ArrayBlockingQueue<byte[]>(Math.min(info.chunks.length,
1024));
+         this.dataBuffer = new ArrayBlockingQueue<>(Math.min(info.chunks.length, 1024));
 +        this.crcCheckChanceSupplier = crcCheckChanceSupplier;
  
-         new Thread(new Reader(source, info, dataBuffer)).start();
+         readerThread = new Thread(new Reader(source, info, dataBuffer));
+         readerThread.start();
      }
  
      public int read() throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0b3cfae4/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0b3cfae4/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
index 5646592,0000000..2162e32
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
+++ b/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
@@@ -1,129 -1,0 +1,183 @@@
 +/*
 + * Licensed to the Apache Software Foundation (ASF) under one
 + * or more contributor license agreements.  See the NOTICE file
 + * distributed with this work for additional information
 + * regarding copyright ownership.  The ASF licenses this file
 + * to you under the Apache License, Version 2.0 (the
 + * "License"); you may not use this file except in compliance
 + * with the License.  You may obtain a copy of the License at
 + *
 + *     http://www.apache.org/licenses/LICENSE-2.0
 + *
 + * Unless required by applicable law or agreed to in writing, software
 + * distributed under the License is distributed on an "AS IS" BASIS,
 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 + * See the License for the specific language governing permissions and
 + * limitations under the License.
 + */
 +package org.apache.cassandra.streaming.compression;
 +
 +import java.io.*;
 +import java.util.*;
++import java.util.concurrent.SynchronousQueue;
++import java.util.concurrent.TimeUnit;
 +
 +import org.junit.Test;
 +import org.apache.cassandra.db.ClusteringComparator;
 +import org.apache.cassandra.db.marshal.BytesType;
 +import org.apache.cassandra.io.compress.CompressedSequentialWriter;
 +import org.apache.cassandra.io.compress.CompressionMetadata;
 +import org.apache.cassandra.schema.CompressionParams;
 +import org.apache.cassandra.io.sstable.Component;
 +import org.apache.cassandra.io.sstable.Descriptor;
 +import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 +import org.apache.cassandra.streaming.compress.CompressedInputStream;
 +import org.apache.cassandra.streaming.compress.CompressionInfo;
 +import org.apache.cassandra.utils.ChecksumType;
 +import org.apache.cassandra.utils.Pair;
 +
 +import static org.junit.Assert.assertEquals;
 +
 +/**
 + */
 +public class CompressedInputStreamTest
 +{
 +    @Test
 +    public void testCompressedRead() throws Exception
 +    {
 +        testCompressedReadWith(new long[]{0L}, false);
 +        testCompressedReadWith(new long[]{1L}, false);
 +        testCompressedReadWith(new long[]{100L}, false);
 +
 +        testCompressedReadWith(new long[]{1L, 122L, 123L, 124L, 456L}, false);
 +    }
 +
 +    @Test(expected = EOFException.class)
 +    public void testTruncatedRead() throws Exception
 +    {
 +        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
++    {
++        CompressionParams param = CompressionParams.snappy(32);
++        CompressionMetadata.Chunk[] chunks = {new CompressionMetadata.Chunk(0, 100)};
++        final SynchronousQueue<Integer> 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,
ChecksumType.CRC32, () -> 1.0))
++        {
++            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
 +     */
 +    private void testCompressedReadWith(long[] valuesToCheck, boolean testTruncate) throws
Exception
 +    {
 +        assert valuesToCheck != null && valuesToCheck.length > 0;
 +
 +        // write compressed data file of longs
 +        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 ClusteringComparator(BytesType.instance));
 +        CompressionParams param = CompressionParams.snappy(32);
-         CompressedSequentialWriter writer = new CompressedSequentialWriter(tmp, desc.filenameFor(Component.COMPRESSION_INFO),
param, collector);
 +        Map<Long, Long> index = new HashMap<Long, Long>();
-         for (long l = 0L; l < 1000; l++)
++        try (CompressedSequentialWriter writer = new CompressedSequentialWriter(tmp, desc.filenameFor(Component.COMPRESSION_INFO),
param, collector))
 +        {
-             index.put(l, writer.position());
-             writer.writeLong(l);
++            for (long l = 0L; l < 1000; l++)
++            {
++                index.put(l, writer.position());
++                writer.writeLong(l);
++            }
++            writer.finish();
 +        }
-         writer.finish();
 +
 +        CompressionMetadata comp = CompressionMetadata.create(tmp.getAbsolutePath());
-         List<Pair<Long, Long>> sections = new ArrayList<Pair<Long, Long>>();
++        List<Pair<Long, Long>> sections = new ArrayList<>();
 +        for (long l : valuesToCheck)
 +        {
 +            long position = index.get(l);
 +            sections.add(Pair.create(position, position + 8));
 +        }
 +        CompressionMetadata.Chunk[] chunks = comp.getChunksForSections(sections);
 +        long totalSize = comp.getTotalSizeForSections(sections);
 +        long expectedSize = 0;
 +        for (CompressionMetadata.Chunk c : chunks)
 +            expectedSize += c.length + 4;
 +        assertEquals(expectedSize, totalSize);
 +
 +        // buffer up only relevant parts of file
 +        int size = 0;
 +        for (CompressionMetadata.Chunk c : chunks)
 +            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)
 +        {
 +            byte [] actuallyRead = new byte[50];
 +            System.arraycopy(toRead, 0, actuallyRead, 0, 50);
 +            toRead = actuallyRead;
 +        }
 +
 +        // read buffer using CompressedInputStream
 +        CompressionInfo info = new CompressionInfo(chunks, param);
 +        CompressedInputStream input = new CompressedInputStream(new ByteArrayInputStream(toRead),
info,
 +                                                                ChecksumType.CRC32, () ->
1.0);
-         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);
++            }
 +        }
 +    }
 +}


Mime
View raw message