cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [05/10] cassandra git commit: Merge branch 'cassandra-2.2' into cassandra-3.0
Date Wed, 13 Jan 2016 19:16:57 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/0f995a2d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/0f995a2d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/0f995a2d

Branch: refs/heads/cassandra-3.3
Commit: 0f995a2dc7a116ec6def110e10af6bb9acc9f7b3
Parents: 95012da 582bdba
Author: Yuki Morishita <yukim@apache.org>
Authored: Wed Jan 13 13:13:11 2016 -0600
Committer: Yuki Morishita <yukim@apache.org>
Committed: Wed Jan 13 13:13:11 2016 -0600

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../cassandra/streaming/ConnectionHandler.java  |  2 +-
 .../cassandra/streaming/StreamReader.java       | 20 +++++++--
 .../cassandra/streaming/StreamWriter.java       | 10 ++++-
 .../compress/CompressedInputStream.java         | 41 ++++++-----------
 .../compress/CompressedStreamReader.java        | 26 +++++++----
 .../compress/CompressedStreamWriter.java        | 15 +++++++
 .../compression/CompressedInputStreamTest.java  | 46 --------------------
 8 files changed, 73 insertions(+), 88 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/0f995a2d/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 38786c1,11f2529..614d5b4
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,20 -1,5 +1,21 @@@
 -2.2.5
 +3.0.3
 + * Fix AssertionError when removing from list using UPDATE (CASSANDRA-10954)
 + * Fix UnsupportedOperationException when reading old sstable with range
 +   tombstone (CASSANDRA-10743)
 + * MV should use the maximum timestamp of the primary key (CASSANDRA-10910)
 + * Fix potential assertion error during compaction (CASSANDRA-10944)
 + * Fix counting of received sstables in streaming (CASSANDRA-10949)
 + * Implement hints compression (CASSANDRA-9428)
 + * Fix potential assertion error when reading static columns (CASSANDRA-10903)
 + * Avoid NoSuchElementException when executing empty batch (CASSANDRA-10711)
 + * Avoid building PartitionUpdate in toString (CASSANDRA-10897)
 + * Reduce heap spent when receiving many SSTables (CASSANDRA-10797)
 + * Add back support for 3rd party auth providers to bulk loader (CASSANDRA-10873)
 + * Eliminate the dependency on jgrapht for UDT resolution (CASSANDRA-10653)
 + * (Hadoop) Close Clusters and Sessions in Hadoop Input/Output classes (CASSANDRA-10837)
 + * Fix sstableloader not working with upper case keyspace name (CASSANDRA-10806)
 +Merged from 2.2:
+  * Fix error streaming section more than 2GB (CASSANDRA-10961)
   * (cqlsh) Also apply --connect-timeout to control connection
     timeout (CASSANDRA-10959)
   * Histogram buckets exposed in jmx are sorted incorrectly (CASSANDRA-10975)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0f995a2d/src/java/org/apache/cassandra/streaming/StreamReader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/StreamReader.java
index 87dcda0,8789720..268f974
--- a/src/java/org/apache/cassandra/streaming/StreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/StreamReader.java
@@@ -61,7 -60,7 +61,8 @@@ public class StreamReade
      protected final long repairedAt;
      protected final SSTableFormat.Type format;
      protected final int sstableLevel;
 +    protected final SerializationHeader.Component header;
+     protected final int fileSeqNum;
  
      protected Descriptor desc;
  
@@@ -75,7 -74,7 +76,8 @@@
          this.repairedAt = header.repairedAt;
          this.format = header.format;
          this.sstableLevel = header.sstableLevel;
 +        this.header = header.header;
+         this.fileSeqNum = header.sequenceNumber;
      }
  
      /**
@@@ -83,10 -82,9 +85,9 @@@
       * @return SSTable transferred
       * @throws IOException if reading the remote sstable fails. Will throw an RTE if local
write fails.
       */
 -    @SuppressWarnings("resource")
 -    public SSTableWriter read(ReadableByteChannel channel) throws IOException
 +    @SuppressWarnings("resource") // channel needs to remain open, streams on top of it
can't be closed
 +    public SSTableMultiWriter read(ReadableByteChannel channel) throws IOException
      {
-         logger.debug("reading file from {}, repairedAt = {}, level = {}", session.peer,
repairedAt, sstableLevel);
          long totalSize = totalSize();
  
          Pair<String, String> kscf = Schema.instance.getCF(cfId);
@@@ -110,13 -117,25 +118,18 @@@
                  // TODO move this to BytesReadTracker
                  session.progress(desc, ProgressInfo.Direction.IN, in.getBytesRead(), totalSize);
              }
+             logger.debug("[Stream #{}] Finished receiving file #{} from {} readBytes = {},
totalSize = {}",
+                          session.planId(), fileSeqNum, session.peer, in.getBytesRead(),
totalSize);
              return writer;
 -        } catch (Throwable e)
 +        }
 +        catch (Throwable e)
          {
 -            if (key != null)
++            if (deserializer != null)
+                 logger.warn("[Stream {}] Error while reading partition {} from stream on
ks='{}' and table='{}'.",
 -                            session.planId(), key, cfs.keyspace.getName(), cfs.getColumnFamilyName());
++                            session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(),
cfs.getColumnFamilyName());
              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);
 -                }
 +                writer.abort(e);
              }
              drain(dis, in.getBytesRead());
              if (e instanceof IOException)
@@@ -126,16 -145,14 +139,15 @@@
          }
      }
  
 -    protected SSTableWriter createWriter(ColumnFamilyStore cfs, long totalSize, long repairedAt,
SSTableFormat.Type format) throws IOException
 +    protected SSTableMultiWriter createWriter(ColumnFamilyStore cfs, long totalSize, long
repairedAt, SSTableFormat.Type format) throws IOException
      {
 -        Directories.DataDirectory localDir = cfs.directories.getWriteableLocation(totalSize);
 +        Directories.DataDirectory localDir = cfs.getDirectories().getWriteableLocation(totalSize);
          if (localDir == null)
              throw new IOException("Insufficient disk space to store " + totalSize + " bytes");
- 
 -        desc = Descriptor.fromFilename(cfs.getTempSSTablePath(cfs.directories.getLocationForDisk(localDir),
format));
 +        desc = Descriptor.fromFilename(cfs.getSSTablePath(cfs.getDirectories().getLocationForDisk(localDir),
format));
  
 -        return SSTableWriter.create(desc, estimatedKeys, repairedAt, sstableLevel);
 +
 +        return cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, sstableLevel,
header.toHeader(cfs.metadata), session.getTransaction(cfId));
      }
  
      protected void drain(InputStream dis, long bytesRead) throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0f995a2d/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
index 56dc63a,489fed9..55ac7ac
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedInputStream.java
@@@ -30,8 -30,10 +30,11 @@@ import java.util.zip.Checksum
  import com.google.common.collect.Iterators;
  import com.google.common.primitives.Ints;
  
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+ 
  import org.apache.cassandra.io.compress.CompressionMetadata;
 +import org.apache.cassandra.utils.ChecksumType;
  import org.apache.cassandra.utils.WrappedRunnable;
  
  /**
@@@ -69,17 -71,15 +73,16 @@@ public class CompressedInputStream exte
       * @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;
  
-         readerThread = new Thread(new Reader(source, info, dataBuffer));
-         readerThread.start();
+         new Thread(new Reader(source, info, dataBuffer)).start();
      }
  
      public int read() throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0f995a2d/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
index 4d10244,c684e4f..5210d5b
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java
@@@ -24,7 -25,8 +24,8 @@@ import java.nio.channels.ReadableByteCh
  
  import com.google.common.base.Throwables;
  
+ import org.apache.cassandra.db.DecoratedKey;
 -import org.apache.cassandra.io.sstable.format.SSTableWriter;
 +import org.apache.cassandra.io.sstable.SSTableMultiWriter;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
@@@ -60,10 -64,9 +61,9 @@@ public class CompressedStreamReader ext
       * @throws java.io.IOException if reading the remote sstable fails. Will throw an RTE
if local write fails.
       */
      @Override
 -    @SuppressWarnings("resource")
 -    public SSTableWriter read(ReadableByteChannel channel) throws IOException
 +    @SuppressWarnings("resource") // channel needs to remain open, streams on top of it
can't be closed
 +    public SSTableMultiWriter read(ReadableByteChannel channel) throws IOException
      {
-         logger.debug("reading file from {}, repairedAt = {}", session.peer, repairedAt);
          long totalSize = totalSize();
  
          Pair<String, String> kscf = Schema.instance.getCF(cfId);
@@@ -72,13 -79,15 +76,16 @@@
              // schema was dropped during streaming
              throw new IOException("CF " + cfId + " was dropped during streaming");
          }
-         ColumnFamilyStore cfs = Keyspace.open(kscf.left).getColumnFamilyStore(kscf.right);
+ 
+         logger.debug("[Stream #{}] Start receiving file #{} from {}, repairedAt = {}, size
= {}, ks = '{}', table = '{}'.",
+                      session.planId(), fileSeqNum, session.peer, repairedAt, totalSize,
cfs.keyspace.getName(),
+                      cfs.getColumnFamilyName());
  
 -        CompressedInputStream cis = new CompressedInputStream(Channels.newInputStream(channel),
compressionInfo);
 +        CompressedInputStream cis = new CompressedInputStream(Channels.newInputStream(channel),
compressionInfo,
 +                                                              inputVersion.compressedChecksumType(),
cfs::getCrcCheckChance);
          BytesReadTracker in = new BytesReadTracker(new DataInputStream(cis));
 -        SSTableWriter writer = null;
 -        DecoratedKey key = null;
 +        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata, in, inputVersion,
header.toHeader(cfs.metadata));
 +        SSTableMultiWriter writer = null;
          try
          {
              writer = createWriter(cfs, totalSize, repairedAt, format);
@@@ -102,9 -117,20 +113,12 @@@
          }
          catch (Throwable e)
          {
 -            if (key != null)
++            if (deserializer != null)
+                 logger.warn("[Stream {}] Error while reading partition {} from stream on
ks='{}' and table='{}'.",
 -                            session.planId(), key, cfs.keyspace.getName(), cfs.getColumnFamilyName());
++                            session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(),
cfs.getColumnFamilyName());
              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);
 -                }
 +                writer.abort(e);
              }
              drain(cis, in.getBytesRead());
              if (e instanceof IOException)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0f995a2d/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java
index adbd091,99e9bd6..f37af29
--- a/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java
+++ b/src/java/org/apache/cassandra/streaming/compress/CompressedStreamWriter.java
@@@ -55,7 -59,9 +60,9 @@@ public class CompressedStreamWriter ext
      public void write(DataOutputStreamPlus out) throws IOException
      {
          long totalSize = totalSize();
+         logger.debug("[Stream #{}] Start streaming file {} to {}, repairedAt = {}, totalSize
= {}", session.planId(),
+                      sstable.getFilename(), session.peer, sstable.getSSTableMetadata().repairedAt,
totalSize);
 -        try (RandomAccessReader file = sstable.openDataReader(); final ChannelProxy fc =
file.getChannel())
 +        try (ChannelProxy fc = sstable.getDataChannel().sharedCopy())
          {
              long progress = 0L;
              // calculate chunks to transfer. we want to send continuous chunks altogether.

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0f995a2d/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
index 2162e32,0000000..a3300ac
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,183 -1,0 +1,137 @@@
 +/*
 + * 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);
 +        Map<Long, Long> index = new HashMap<Long, Long>();
 +        try (CompressedSequentialWriter writer = new CompressedSequentialWriter(tmp, desc.filenameFor(Component.COMPRESSION_INFO),
param, collector))
 +        {
 +            for (long l = 0L; l < 1000; l++)
 +            {
 +                index.put(l, writer.position());
 +                writer.writeLong(l);
 +            }
 +            writer.finish();
 +        }
 +
 +        CompressionMetadata comp = CompressionMetadata.create(tmp.getAbsolutePath());
 +        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];
 +
 +        try (RandomAccessFile f = new RandomAccessFile(tmp, "r"))
 +        {
 +            int pos = 0;
 +            for (CompressionMetadata.Chunk c : chunks)
 +            {
 +                f.seek(c.offset);
 +                pos += f.read(toRead, pos, c.length + 4);
 +            }
 +        }
 +
 +        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);
 +
 +        try (DataInputStream in = new DataInputStream(input))
 +        {
 +            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