getOutboundHeaderFlags();
+
+}
diff --git a/src/java/org/apache/cassandra/transport/frame/checksum/ChecksummingTransformer.java b/src/java/org/apache/cassandra/transport/frame/checksum/ChecksummingTransformer.java
new file mode 100644
index 0000000000..3b15cee928
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/frame/checksum/ChecksummingTransformer.java
@@ -0,0 +1,361 @@
+/*
+ * 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.transport.frame.checksum;
+
+import java.io.IOException;
+import java.util.EnumSet;
+
+import com.google.common.collect.ImmutableTable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.transport.Frame;
+import org.apache.cassandra.transport.ProtocolException;
+import org.apache.cassandra.transport.frame.FrameBodyTransformer;
+import org.apache.cassandra.transport.frame.compress.Compressor;
+import org.apache.cassandra.transport.frame.compress.LZ4Compressor;
+import org.apache.cassandra.transport.frame.compress.SnappyCompressor;
+import org.apache.cassandra.utils.ChecksumType;
+
+import static org.apache.cassandra.transport.CBUtil.readUnsignedShort;
+
+/**
+ * Provides a format that implements chunking and checksumming logic
+ * that maybe used in conjunction with a frame Compressor if required
+ *
+ * 1.1. Checksummed/Compression Serialized Format
+ *
+ *
+ * {@code
+ * 1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
+ * 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Number of Compressed Chunks | Compressed Length (e1) /
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * / Compressed Length cont. (e1) | Uncompressed Length (e1) /
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Uncompressed Length cont. (e1)| Checksum of Lengths (e1) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Checksum of Lengths cont. (e1)| Compressed Bytes (e1) +//
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Checksum (e1) ||
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Compressed Length (e2) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Uncompressed Length (e2) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Checksum of Lengths (e2) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Compressed Bytes (e2) +//
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Checksum (e2) ||
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Compressed Length (en) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Uncompressed Length (en) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Checksum of Lengths (en) |
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Compressed Bytes (en) +//
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * | Checksum (en) ||
+ * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+ * }
+ *
+ *
+ *
+ * 1.2. Checksum Compression Description
+ *
+ * The entire payload is broken into n chunks each with a pair of checksums:
+ *
+ * - [int]: compressed length of serialized bytes for this chunk (e.g. the length post compression)
+ *
- [int]: expected length of the decompressed bytes (e.g. the length after decompression)
+ *
- [int]: digest of decompressed and compressed length components above
+ *
- [k bytes]: compressed payload for this chunk
+ *
- [int]: digest of the decompressed result of the payload above for this chunk
+ *
+ *
+ */
+public class ChecksummingTransformer implements FrameBodyTransformer
+{
+ private static final Logger logger = LoggerFactory.getLogger(ChecksummingTransformer.class);
+
+ private static final EnumSet CHECKSUMS_ONLY = EnumSet.of(Frame.Header.Flag.CHECKSUMMED);
+ private static final EnumSet CHECKSUMS_AND_COMPRESSION = EnumSet.of(Frame.Header.Flag.CHECKSUMMED, Frame.Header.Flag.COMPRESSED);
+
+ private static final int CHUNK_HEADER_OVERHEAD = Integer.BYTES + Integer.BYTES + Integer.BYTES + Integer.BYTES;
+
+ private static final ChecksummingTransformer CRC32_NO_COMPRESSION = new ChecksummingTransformer(ChecksumType.CRC32, null);
+ private static final ChecksummingTransformer ADLER32_NO_COMPRESSION = new ChecksummingTransformer(ChecksumType.Adler32, null);
+ private static final ImmutableTable transformers;
+ static
+ {
+ ImmutableTable.Builder builder = ImmutableTable.builder();
+ builder.put(ChecksumType.CRC32, LZ4Compressor.INSTANCE, new ChecksummingTransformer(ChecksumType.CRC32, LZ4Compressor.INSTANCE));
+ builder.put(ChecksumType.CRC32, SnappyCompressor.INSTANCE, new ChecksummingTransformer(ChecksumType.CRC32, SnappyCompressor.INSTANCE));
+ builder.put(ChecksumType.Adler32, LZ4Compressor.INSTANCE, new ChecksummingTransformer(ChecksumType.Adler32, LZ4Compressor.INSTANCE));
+ builder.put(ChecksumType.Adler32, SnappyCompressor.INSTANCE, new ChecksummingTransformer(ChecksumType.Adler32, SnappyCompressor.INSTANCE));
+ transformers = builder.build();
+ }
+
+ private final int blockSize;
+ private final Compressor compressor;
+ private final ChecksumType checksum;
+
+ public static ChecksummingTransformer getTransformer(ChecksumType checksumType, Compressor compressor)
+ {
+ ChecksummingTransformer transformer = compressor == null
+ ? checksumType == ChecksumType.CRC32 ? CRC32_NO_COMPRESSION : ADLER32_NO_COMPRESSION
+ : transformers.get(checksumType, compressor);
+
+ if (transformer == null)
+ {
+ logger.warn("Invalid compression/checksum options supplied. %s / %s", checksumType, compressor.getClass().getName());
+ throw new RuntimeException("Invalid compression / checksum options supplied");
+ }
+
+ return transformer;
+ }
+
+ ChecksummingTransformer(ChecksumType checksumType, Compressor compressor)
+ {
+ this(checksumType, DatabaseDescriptor.getNativeTransportFrameBlockSize(), compressor);
+ }
+
+ ChecksummingTransformer(ChecksumType checksumType, int blockSize, Compressor compressor)
+ {
+ this.checksum = checksumType;
+ this.blockSize = blockSize;
+ this.compressor = compressor;
+ }
+
+ public EnumSet getOutboundHeaderFlags()
+ {
+ return null == compressor ? CHECKSUMS_ONLY : CHECKSUMS_AND_COMPRESSION;
+ }
+
+ public ByteBuf transformOutbound(ByteBuf inputBuf)
+ {
+ // be pessimistic about life and assume the compressed output will be the same size as the input bytes
+ int maxTotalCompressedLength = maxCompressedLength(inputBuf.readableBytes());
+ int expectedChunks = (int) Math.ceil((double) maxTotalCompressedLength / blockSize);
+ int expectedMaxSerializedLength = Short.BYTES + (expectedChunks * CHUNK_HEADER_OVERHEAD) + maxTotalCompressedLength;
+ byte[] retBuf = new byte[expectedMaxSerializedLength];
+ ByteBuf ret = Unpooled.wrappedBuffer(retBuf);
+ ret.writerIndex(0);
+ ret.readerIndex(0);
+
+ // write out bogus short to start with as we'll encode one at the end
+ // when we finalize the number of compressed chunks to expect and this
+ // sets the writer index correctly for starting the first chunk
+ ret.writeShort((short) 0);
+
+ byte[] inBuf = new byte[blockSize];
+ byte[] outBuf = new byte[maxCompressedLength(blockSize)];
+ byte[] chunkLengths = new byte[8];
+
+ int numCompressedChunks = 0;
+ int readableBytes;
+ int lengthsChecksum;
+ while ((readableBytes = inputBuf.readableBytes()) > 0)
+ {
+ int lengthToRead = Math.min(blockSize, readableBytes);
+ inputBuf.readBytes(inBuf, 0, lengthToRead);
+ int uncompressedChunkChecksum = (int) checksum.of(inBuf, 0, lengthToRead);
+ int compressedSize = maybeCompress(inBuf, lengthToRead, outBuf);
+
+ if (compressedSize < lengthToRead)
+ {
+ // there was some benefit to compression so write out the compressed
+ // and uncompressed sizes of the chunk
+ ret.writeInt(compressedSize);
+ ret.writeInt(lengthToRead);
+ putInt(compressedSize, chunkLengths, 0);
+ }
+ else
+ {
+ // if no compression was possible, there's no need to write two lengths, so
+ // just write the size of the original content (or block size), with its
+ // sign flipped to signal no compression.
+ ret.writeInt(-lengthToRead);
+ putInt(-lengthToRead, chunkLengths, 0);
+ }
+
+ putInt(lengthToRead, chunkLengths, 4);
+
+ // calculate the checksum of the compressed and decompressed lengths
+ // protect us against a bogus length causing potential havoc on deserialization
+ lengthsChecksum = (int) checksum.of(chunkLengths, 0, chunkLengths.length);
+ ret.writeInt(lengthsChecksum);
+
+ // figure out how many actual bytes we're going to write and make sure we have capacity
+ int toWrite = Math.min(compressedSize, lengthToRead);
+ if (ret.writableBytes() < (CHUNK_HEADER_OVERHEAD + toWrite))
+ {
+ // this really shouldn't ever happen -- it means we either mis-calculated the number of chunks we
+ // expected to create, we gave some input to the compressor that caused the output to be much
+ // larger than the input.. or some other edge condition. Regardless -- resize if necessary.
+ byte[] resizedRetBuf = new byte[(retBuf.length + (CHUNK_HEADER_OVERHEAD + toWrite)) * 3 / 2];
+ System.arraycopy(retBuf, 0, resizedRetBuf, 0, retBuf.length);
+ retBuf = resizedRetBuf;
+ ByteBuf resizedRetByteBuf = Unpooled.wrappedBuffer(retBuf);
+ resizedRetByteBuf.writerIndex(ret.writerIndex());
+ ret = resizedRetByteBuf;
+ }
+
+ // write the bytes, either compressed or uncompressed
+ if (compressedSize < lengthToRead)
+ ret.writeBytes(outBuf, 0, toWrite); // compressed
+ else
+ ret.writeBytes(inBuf, 0, toWrite); // uncompressed
+
+ // checksum of the uncompressed chunk
+ ret.writeInt(uncompressedChunkChecksum);
+
+ numCompressedChunks++;
+ }
+
+ // now update the number of chunks
+ ret.setShort(0, (short) numCompressedChunks);
+ return ret;
+ }
+
+ public ByteBuf transformInbound(ByteBuf inputBuf, EnumSet flags)
+ {
+ int numChunks = readUnsignedShort(inputBuf);
+
+ int currentPosition = 0;
+ int decompressedLength;
+ int lengthsChecksum;
+
+ byte[] buf = null;
+ byte[] retBuf = new byte[inputBuf.readableBytes()];
+ byte[] chunkLengths = new byte[8];
+ for (int i = 0; i < numChunks; i++)
+ {
+ int compressedLength = inputBuf.readInt();
+ // if the input was actually compressed, then the writer should have written a decompressed
+ // length. If not, then we can infer that the compressed length has had its sign bit flipped
+ // and can derive the decompressed length from that
+ decompressedLength = compressedLength >= 0 ? inputBuf.readInt() : Math.abs(compressedLength);
+
+ putInt(compressedLength, chunkLengths, 0);
+ putInt(decompressedLength, chunkLengths, 4);
+ lengthsChecksum = inputBuf.readInt();
+ // calculate checksum on lengths (decompressed and compressed) and make sure it matches
+ int calculatedLengthsChecksum = (int) checksum.of(chunkLengths, 0, chunkLengths.length);
+ if (lengthsChecksum != calculatedLengthsChecksum)
+ {
+ throw new ProtocolException(String.format("Checksum invalid on chunk bytes lengths. Deserialized compressed " +
+ "length: %d decompressed length: %d. %d != %d", compressedLength,
+ decompressedLength, lengthsChecksum, calculatedLengthsChecksum));
+ }
+
+ // do we have enough space in the decompression buffer?
+ if (currentPosition + decompressedLength > retBuf.length)
+ {
+ byte[] resizedBuf = new byte[retBuf.length + decompressedLength * 3 / 2];
+ System.arraycopy(retBuf, 0, resizedBuf, 0, retBuf.length);
+ retBuf = resizedBuf;
+ }
+
+ // now we've validated the lengths checksum, we can abs the compressed length
+ // to figure out the actual number of bytes we're going to read
+ int toRead = Math.abs(compressedLength);
+ if (buf == null || buf.length < toRead)
+ buf = new byte[toRead];
+
+ // get the (possibly) compressed bytes for this chunk
+ inputBuf.readBytes(buf, 0, toRead);
+
+ // decompress using the original compressed length, so it's a no-op if that's < 0
+ byte[] decompressedChunk = maybeDecompress(buf, compressedLength, decompressedLength, flags);
+
+ // add the decompressed bytes into the ret buf
+ System.arraycopy(decompressedChunk, 0, retBuf, currentPosition, decompressedLength);
+ currentPosition += decompressedLength;
+
+ // get the checksum of the original source bytes and compare against what we read
+ int expectedDecompressedChecksum = inputBuf.readInt();
+ int calculatedDecompressedChecksum = (int) checksum.of(decompressedChunk, 0, decompressedLength);
+ if (expectedDecompressedChecksum != calculatedDecompressedChecksum)
+ {
+ throw new ProtocolException("Decompressed checksum for chunk does not match expected checksum");
+ }
+ }
+
+ ByteBuf ret = Unpooled.wrappedBuffer(retBuf, 0, currentPosition);
+ ret.writerIndex(currentPosition);
+ return ret;
+ }
+
+ private int maxCompressedLength(int uncompressedLength)
+ {
+ return null == compressor ? uncompressedLength : compressor.maxCompressedLength(uncompressedLength);
+
+ }
+
+ private int maybeCompress(byte[] input, int length, byte[] output)
+ {
+ if (null == compressor)
+ {
+ System.arraycopy(input, 0, output, 0, length);
+ return length;
+ }
+
+ try
+ {
+ return compressor.compress(input, 0, length, output, 0);
+ }
+ catch (IOException e)
+ {
+ logger.info("IO error during compression of frame body chunk", e);
+ throw new ProtocolException("Error compressing frame body chunk");
+ }
+ }
+
+ private byte[] maybeDecompress(byte[] input, int length, int expectedLength, EnumSet flags)
+ {
+ if (null == compressor || !flags.contains(Frame.Header.Flag.COMPRESSED) || length < 0)
+ return input;
+
+ try
+ {
+ return compressor.decompress(input, 0, length, expectedLength);
+ }
+ catch (IOException e)
+ {
+ logger.info("IO error during decompression of frame body chunk", e);
+ throw new ProtocolException("Error decompressing frame body chunk");
+ }
+ }
+
+ private void putInt(int val, byte[] dest, int offset)
+ {
+ dest[offset] = (byte) (val >>> 24);
+ dest[offset + 1] = (byte) (val >>> 16);
+ dest[offset + 2] = (byte) (val >>> 8);
+ dest[offset + 3] = (byte) (val);
+ }
+
+}
diff --git a/src/java/org/apache/cassandra/transport/frame/compress/CompressingTransformer.java b/src/java/org/apache/cassandra/transport/frame/compress/CompressingTransformer.java
new file mode 100644
index 0000000000..db99edfe9c
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/frame/compress/CompressingTransformer.java
@@ -0,0 +1,164 @@
+/*
+ * 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.transport.frame.compress;
+
+import java.io.IOException;
+import java.util.EnumSet;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.cassandra.transport.CBUtil;
+import org.apache.cassandra.transport.Frame;
+import org.apache.cassandra.transport.ProtocolException;
+import org.apache.cassandra.transport.frame.FrameBodyTransformer;
+
+public abstract class CompressingTransformer implements FrameBodyTransformer
+{
+ private static final CompressingTransformer LZ4 = new LZ4();
+ private static final CompressingTransformer SNAPPY = new Snappy();
+
+ private static final EnumSet headerFlags = EnumSet.of(Frame.Header.Flag.COMPRESSED);
+
+ public static final CompressingTransformer getTransformer(Compressor compressor)
+ {
+ if (compressor instanceof LZ4Compressor)
+ return LZ4;
+
+ if (compressor instanceof SnappyCompressor)
+ {
+ if (SnappyCompressor.INSTANCE == null)
+ throw new ProtocolException("This instance does not support Snappy compression");
+
+ return SNAPPY;
+ }
+
+ throw new ProtocolException("Unsupported compression implementation: " + compressor.getClass().getCanonicalName());
+ }
+
+ CompressingTransformer() {}
+
+ public EnumSet getOutboundHeaderFlags()
+ {
+ return headerFlags;
+ }
+
+ public ByteBuf transformInbound(ByteBuf inputBuf, EnumSet flags) throws IOException
+ {
+ return transformInbound(inputBuf);
+ }
+
+ abstract ByteBuf transformInbound(ByteBuf inputBuf) throws IOException;
+
+ // Simple LZ4 encoding prefixes the compressed bytes with the
+ // length of the uncompressed bytes. This length is explicitly big-endian
+ // as the native protocol is entirely big-endian, so it feels like putting
+ // little-endian here would be a annoying trap for client writer
+ private static class LZ4 extends CompressingTransformer
+ {
+ public ByteBuf transformOutbound(ByteBuf inputBuf) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(inputBuf);
+ int maxCompressedLength = LZ4Compressor.INSTANCE.maxCompressedLength(input.length);
+ ByteBuf outputBuf = CBUtil.allocator.heapBuffer(Integer.BYTES + maxCompressedLength);
+ byte[] output = outputBuf.array();
+ int outputOffset = outputBuf.arrayOffset();
+ output[outputOffset] = (byte) (input.length >>> 24);
+ output[outputOffset + 1] = (byte) (input.length >>> 16);
+ output[outputOffset + 2] = (byte) (input.length >>> 8);
+ output[outputOffset + 3] = (byte) (input.length);
+ try
+ {
+ int written = LZ4Compressor.INSTANCE.compress(input, 0, input.length, output, Integer.BYTES + outputOffset);
+ outputBuf.writerIndex(Integer.BYTES + written);
+ return outputBuf;
+ }
+ catch (IOException e)
+ {
+ outputBuf.release();
+ throw e;
+ }
+ }
+
+ ByteBuf transformInbound(ByteBuf inputBuf) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(inputBuf);
+ int uncompressedLength = ((input[0] & 0xFF) << 24)
+ | ((input[1] & 0xFF) << 16)
+ | ((input[2] & 0xFF) << 8)
+ | ((input[3] & 0xFF));
+ ByteBuf outputBuf = CBUtil.allocator.heapBuffer(uncompressedLength);
+ try
+ {
+ outputBuf.writeBytes(LZ4Compressor.INSTANCE.decompress(input,
+ Integer.BYTES,
+ input.length - Integer.BYTES,
+ uncompressedLength));
+ return outputBuf;
+ }
+ catch (IOException e)
+ {
+ outputBuf.release();
+ throw e;
+ }
+ }
+ }
+
+ // Simple Snappy encoding simply writes the compressed bytes, without the preceding length
+ private static class Snappy extends CompressingTransformer
+ {
+ public ByteBuf transformOutbound(ByteBuf inputBuf) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(inputBuf);
+ int uncompressedLength = input.length;
+ int maxCompressedLength = SnappyCompressor.INSTANCE.maxCompressedLength(uncompressedLength);
+ ByteBuf outputBuf = CBUtil.allocator.heapBuffer(maxCompressedLength);
+ try
+ {
+ int written = SnappyCompressor.INSTANCE.compress(input,
+ 0,
+ uncompressedLength,
+ outputBuf.array(),
+ outputBuf.arrayOffset());
+ outputBuf.writerIndex(written);
+ return outputBuf;
+ }
+ catch (IOException e)
+ {
+ outputBuf.release();
+ throw e;
+ }
+ }
+
+ ByteBuf transformInbound(ByteBuf inputBuf) throws IOException
+ {
+ byte[] input = CBUtil.readRawBytes(inputBuf);
+ int uncompressedLength = org.xerial.snappy.Snappy.uncompressedLength(input);
+ ByteBuf outputBuf = CBUtil.allocator.heapBuffer(uncompressedLength);
+ try
+ {
+ outputBuf.writeBytes(SnappyCompressor.INSTANCE.decompress(input, 0, input.length, uncompressedLength));
+ return outputBuf;
+ }
+ catch (IOException e)
+ {
+ outputBuf.release();
+ throw e;
+ }
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/frame/compress/Compressor.java b/src/java/org/apache/cassandra/transport/frame/compress/Compressor.java
new file mode 100644
index 0000000000..e458bdb448
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/frame/compress/Compressor.java
@@ -0,0 +1,62 @@
+/*
+ * 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.transport.frame.compress;
+
+import java.io.IOException;
+
+/**
+ * Analogous to {@link org.apache.cassandra.io.compress.ICompressor}, but different enough that
+ * it's worth specializing:
+ *
+ * - disk IO is mostly oriented around ByteBuffers, whereas with Frames raw byte arrays are
+ * primarily used
+ * - our LZ4 compression format is opionated about the endianness of the preceding length
+ * bytes, big for protocol, little for disk
+ * - ICompressor doesn't make it easy to pre-allocate the output buffer/array
+ *
+ *
+ * In future it may be worth revisiting to unify the interfaces.
+ */
+public interface Compressor
+{
+ /**
+ * @param length the decompressed length being compressed
+ * @return the maximum length output possible for an input of the provided length
+ */
+ int maxCompressedLength(int length);
+
+ /**
+ * @param src the input bytes to be compressed
+ * @param srcOffset the offset to start compressing src from
+ * @param length the total number of bytes from srcOffset to pass to the compressor implementation
+ * @param dest the output buffer to write the compressed bytes to
+ * @param destOffset the offset into the dest buffer to start writing the compressed bytes
+ * @return the length of resulting compressed bytes written into the dest buffer
+ * @throws IOException if the compression implementation failed while compressing the input bytes
+ */
+ int compress(byte[] src, int srcOffset, int length, byte[] dest, int destOffset) throws IOException;
+
+ /**
+ * @param src the compressed bytes to be decompressed
+ * @param expectedDecompressedLength the expected length the input bytes will decompress to
+ * @return a byte[] containing the resuling decompressed bytes
+ * @throws IOException thrown if the compression implementation failed to decompress the provided input bytes
+ */
+ byte[] decompress(byte[] src, int srcOffset, int length, int expectedDecompressedLength) throws IOException;
+}
diff --git a/src/java/org/apache/cassandra/transport/frame/compress/LZ4Compressor.java b/src/java/org/apache/cassandra/transport/frame/compress/LZ4Compressor.java
new file mode 100644
index 0000000000..8ac42e2698
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/frame/compress/LZ4Compressor.java
@@ -0,0 +1,68 @@
+/*
+ * 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.transport.frame.compress;
+
+import java.io.IOException;
+
+import net.jpountz.lz4.LZ4Factory;
+import net.jpountz.lz4.LZ4FastDecompressor;
+
+public class LZ4Compressor implements Compressor
+{
+ public static final LZ4Compressor INSTANCE = new LZ4Compressor();
+
+ private final net.jpountz.lz4.LZ4Compressor compressor;
+ private final LZ4FastDecompressor decompressor;
+
+ private LZ4Compressor()
+ {
+ final LZ4Factory lz4Factory = LZ4Factory.fastestInstance();
+ compressor = lz4Factory.fastCompressor();
+ decompressor = lz4Factory.fastDecompressor();
+ }
+
+ public int maxCompressedLength(int length)
+ {
+ return compressor.maxCompressedLength(length);
+ }
+
+ public int compress(byte[] src, int srcOffset, int length, byte[] dest, int destOffset) throws IOException
+ {
+ try
+ {
+ return compressor.compress(src, srcOffset, length, dest, destOffset);
+ }
+ catch (Throwable t)
+ {
+ throw new IOException("Error caught during LZ4 compression", t);
+ }
+ }
+
+ public byte[] decompress(byte[] src, int offset, int length, int expectedDecompressedLength) throws IOException
+ {
+ try
+ {
+ return decompressor.decompress(src, offset, expectedDecompressedLength);
+ }
+ catch (Throwable t)
+ {
+ throw new IOException("Error caught during LZ4 decompression", t);
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/frame/compress/SnappyCompressor.java b/src/java/org/apache/cassandra/transport/frame/compress/SnappyCompressor.java
new file mode 100644
index 0000000000..27ea4c3224
--- /dev/null
+++ b/src/java/org/apache/cassandra/transport/frame/compress/SnappyCompressor.java
@@ -0,0 +1,79 @@
+/*
+ * 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.transport.frame.compress;
+
+import java.io.IOException;
+
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.xerial.snappy.Snappy;
+import org.xerial.snappy.SnappyError;
+
+public class SnappyCompressor implements Compressor
+{
+ public static final SnappyCompressor INSTANCE;
+ static
+ {
+ SnappyCompressor i;
+ try
+ {
+ i = new SnappyCompressor();
+ }
+ catch (Exception e)
+ {
+ JVMStabilityInspector.inspectThrowable(e);
+ i = null;
+ }
+ catch (NoClassDefFoundError | SnappyError | UnsatisfiedLinkError e)
+ {
+ i = null;
+ }
+ INSTANCE = i;
+ }
+
+ private SnappyCompressor()
+ {
+ // this would throw java.lang.NoClassDefFoundError if Snappy class
+ // wasn't found at runtime which should be processed by the calling method
+ Snappy.getNativeLibraryVersion();
+ }
+
+ @Override
+ public int maxCompressedLength(int length)
+ {
+ return Snappy.maxCompressedLength(length);
+ }
+
+ @Override
+ public int compress(byte[] src, int srcOffset, int length, byte[] dest, int destOffset) throws IOException
+ {
+ return Snappy.compress(src, 0, src.length, dest, destOffset);
+ }
+
+ @Override
+ public byte[] decompress(byte[] src, int offset, int length, int expectedDecompressedLength) throws IOException
+ {
+ if (!Snappy.isValidCompressedBuffer(src, 0, length))
+ throw new IOException("Provided frame does not appear to be Snappy compressed");
+
+ int uncompressedLength = Snappy.uncompressedLength(src);
+ byte[] output = new byte[uncompressedLength];
+ Snappy.uncompress(src, offset, length, output, 0);
+ return output;
+ }
+}
diff --git a/src/java/org/apache/cassandra/transport/messages/OptionsMessage.java b/src/java/org/apache/cassandra/transport/messages/OptionsMessage.java
index 2b8e695f78..a29145a383 100644
--- a/src/java/org/apache/cassandra/transport/messages/OptionsMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/OptionsMessage.java
@@ -26,9 +26,10 @@ import io.netty.buffer.ByteBuf;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.service.QueryState;
-import org.apache.cassandra.transport.FrameCompressor;
import org.apache.cassandra.transport.Message;
import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.transport.frame.compress.SnappyCompressor;
+import org.apache.cassandra.utils.ChecksumType;
/**
* Message to indicate that the server is ready to receive requests.
@@ -60,20 +61,29 @@ public class OptionsMessage extends Message.Request
@Override
protected Message.Response execute(QueryState state, long queryStartNanoTime, boolean traceRequest)
{
- List cqlVersions = new ArrayList();
+ List cqlVersions = new ArrayList<>();
cqlVersions.add(QueryProcessor.CQL_VERSION.toString());
- List compressions = new ArrayList();
- if (FrameCompressor.SnappyCompressor.instance != null)
+ List compressions = new ArrayList<>();
+ if (SnappyCompressor.INSTANCE != null)
compressions.add("snappy");
// LZ4 is always available since worst case scenario it default to a pure JAVA implem.
compressions.add("lz4");
- Map> supported = new HashMap>();
+ Map> supported = new HashMap<>();
supported.put(StartupMessage.CQL_VERSION, cqlVersions);
supported.put(StartupMessage.COMPRESSION, compressions);
supported.put(StartupMessage.PROTOCOL_VERSIONS, ProtocolVersion.supportedVersions());
+ if (connection.getVersion().supportsChecksums())
+ {
+ ChecksumType[] types = ChecksumType.values();
+ List checksumImpls = new ArrayList<>(types.length);
+ for (ChecksumType type : types)
+ checksumImpls.add(type.toString());
+ supported.put(StartupMessage.CHECKSUM, checksumImpls);
+ }
+
return new SupportedMessage(supported);
}
diff --git a/src/java/org/apache/cassandra/transport/messages/StartupMessage.java b/src/java/org/apache/cassandra/transport/messages/StartupMessage.java
index 01b9331cb3..92c9764861 100644
--- a/src/java/org/apache/cassandra/transport/messages/StartupMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/StartupMessage.java
@@ -26,7 +26,13 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.*;
+import org.apache.cassandra.transport.frame.checksum.ChecksummingTransformer;
+import org.apache.cassandra.transport.frame.compress.CompressingTransformer;
+import org.apache.cassandra.transport.frame.compress.Compressor;
+import org.apache.cassandra.transport.frame.compress.LZ4Compressor;
+import org.apache.cassandra.transport.frame.compress.SnappyCompressor;
import org.apache.cassandra.utils.CassandraVersion;
+import org.apache.cassandra.utils.ChecksumType;
/**
* The initial message of the protocol.
@@ -39,6 +45,7 @@ public class StartupMessage extends Message.Request
public static final String PROTOCOL_VERSIONS = "PROTOCOL_VERSIONS";
public static final String DRIVER_NAME = "DRIVER_NAME";
public static final String DRIVER_VERSION = "DRIVER_VERSION";
+ public static final String CHECKSUM = "CONTENT_CHECKSUM";
public static final Message.Codec codec = new Message.Codec()
{
@@ -83,23 +90,18 @@ public class StartupMessage extends Message.Request
throw new ProtocolException(e.getMessage());
}
- if (options.containsKey(COMPRESSION))
+ ChecksumType checksumType = getChecksumType();
+ Compressor compressor = getCompressor();
+
+ if (null != checksumType)
{
- String compression = options.get(COMPRESSION).toLowerCase();
- if (compression.equals("snappy"))
- {
- if (FrameCompressor.SnappyCompressor.instance == null)
- throw new ProtocolException("This instance does not support Snappy compression");
- connection.setCompressor(FrameCompressor.SnappyCompressor.instance);
- }
- else if (compression.equals("lz4"))
- {
- connection.setCompressor(FrameCompressor.LZ4Compressor.instance);
- }
- else
- {
- throw new ProtocolException(String.format("Unknown compression algorithm: %s", compression));
- }
+ if (!connection.getVersion().supportsChecksums())
+ throw new ProtocolException(String.format("Invalid message flag. Protocol version %s does not support frame body checksums", connection.getVersion().toString()));
+ connection.setTransformer(ChecksummingTransformer.getTransformer(checksumType, compressor));
+ }
+ else if (null != compressor)
+ {
+ connection.setTransformer(CompressingTransformer.getTransformer(compressor));
}
ClientState clientState = state.getClientState();
@@ -124,6 +126,42 @@ public class StartupMessage extends Message.Request
return newMap;
}
+ private ChecksumType getChecksumType() throws ProtocolException
+ {
+ String name = options.get(CHECKSUM);
+ try
+ {
+ return name != null ? ChecksumType.valueOf(name) : null;
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new ProtocolException(String.format("Requested checksum type %s is not known or supported by " +
+ "this version of Cassandra", name));
+ }
+ }
+
+ private Compressor getCompressor() throws ProtocolException
+ {
+ String name = options.get(COMPRESSION);
+ if (null == name)
+ return null;
+
+ switch (name.toLowerCase())
+ {
+ case "snappy":
+ {
+ if (SnappyCompressor.INSTANCE == null)
+ throw new ProtocolException("This instance does not support Snappy compression");
+
+ return SnappyCompressor.INSTANCE;
+ }
+ case "lz4":
+ return LZ4Compressor.INSTANCE;
+ default:
+ throw new ProtocolException(String.format("Unknown compression algorithm: %s", name));
+ }
+ }
+
@Override
public String toString()
{
diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java
index 2fbbc28190..adadb9c27d 100644
--- a/test/unit/org/apache/cassandra/cql3/CQLTester.java
+++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java
@@ -871,9 +871,9 @@ public abstract class CQLTester
return sessions.get(protocolVersion);
}
- protected SimpleClient newSimpleClient(ProtocolVersion version, boolean compression) throws IOException
+ protected SimpleClient newSimpleClient(ProtocolVersion version, boolean compression, boolean checksums) throws IOException
{
- return new SimpleClient(nativeAddr.getHostAddress(), nativePort, version, version.isBeta(), new EncryptionOptions()).connect(compression);
+ return new SimpleClient(nativeAddr.getHostAddress(), nativePort, version, version.isBeta(), new EncryptionOptions()).connect(compression, checksums);
}
protected String formatQuery(String query)
diff --git a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
index 0a314da6dc..11df055380 100644
--- a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
+++ b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
@@ -292,7 +292,7 @@ public class PreparedStatementsTest extends CQLTester
createTable("CREATE TABLE %s (pk int, v1 int, v2 int, PRIMARY KEY (pk))");
execute("INSERT INTO %s (pk, v1, v2) VALUES (1,1,1)");
- try (SimpleClient simpleClient = newSimpleClient(ProtocolVersion.BETA.orElse(ProtocolVersion.CURRENT), false))
+ try (SimpleClient simpleClient = newSimpleClient(ProtocolVersion.BETA.orElse(ProtocolVersion.CURRENT), false, false))
{
ResultMessage.Prepared prepUpdate = simpleClient.prepare(String.format("UPDATE %s.%s SET v1 = ?, v2 = ? WHERE pk = 1 IF v1 = ?",
keyspace(), currentTable()));
diff --git a/test/unit/org/apache/cassandra/service/ClientWarningsTest.java b/test/unit/org/apache/cassandra/service/ClientWarningsTest.java
index e939df0f23..3ae49edf96 100644
--- a/test/unit/org/apache/cassandra/service/ClientWarningsTest.java
+++ b/test/unit/org/apache/cassandra/service/ClientWarningsTest.java
@@ -51,7 +51,7 @@ public class ClientWarningsTest extends CQLTester
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V4))
{
- client.connect(false);
+ client.connect(false, false);
QueryMessage query = new QueryMessage(createBatchStatement2(1), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
@@ -70,7 +70,7 @@ public class ClientWarningsTest extends CQLTester
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V4))
{
- client.connect(false);
+ client.connect(false, false);
QueryMessage query = new QueryMessage(createBatchStatement2(DatabaseDescriptor.getBatchSizeWarnThreshold() / 2 + 1), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
@@ -90,7 +90,7 @@ public class ClientWarningsTest extends CQLTester
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V4))
{
- client.connect(false);
+ client.connect(false, false);
for (int i = 0; i < iterations; i++)
{
@@ -130,7 +130,7 @@ public class ClientWarningsTest extends CQLTester
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V3))
{
- client.connect(false);
+ client.connect(false, false);
QueryMessage query = new QueryMessage(createBatchStatement(DatabaseDescriptor.getBatchSizeWarnThreshold()), QueryOptions.DEFAULT);
Message.Response resp = client.execute(query);
diff --git a/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java b/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
index 4ade4adcfb..7f08cf2bf9 100644
--- a/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
+++ b/test/unit/org/apache/cassandra/service/ProtocolBetaVersionTest.java
@@ -70,7 +70,7 @@ public class ProtocolBetaVersionTest extends CQLTester
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, betaVersion, true, new EncryptionOptions()))
{
- client.connect(false);
+ client.connect(false, false);
for (int i = 0; i < 10; i++)
{
QueryMessage query = new QueryMessage(String.format("INSERT INTO %s.%s (pk, v) VALUES (%s, %s)",
@@ -105,7 +105,7 @@ public class ProtocolBetaVersionTest extends CQLTester
assertTrue(betaVersion.isBeta()); // change to another beta version or remove test if no beta version
try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, betaVersion, false, new EncryptionOptions()))
{
- client.connect(false);
+ client.connect(false, false);
fail("Exception should have been thrown");
}
catch (Exception e)
diff --git a/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java b/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java
index 0c15bca5db..a2ee6fb3a0 100644
--- a/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java
+++ b/test/unit/org/apache/cassandra/transport/MessagePayloadTest.java
@@ -129,7 +129,7 @@ public class MessagePayloadTest extends CQLTester
new EncryptionOptions());
try
{
- client.connect(false);
+ client.connect(false, false);
Map reqMap;
Map respMap;
@@ -205,7 +205,7 @@ public class MessagePayloadTest extends CQLTester
SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort);
try
{
- client.connect(false);
+ client.connect(false, false);
Map reqMap;
Map respMap;
@@ -274,7 +274,7 @@ public class MessagePayloadTest extends CQLTester
SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, ProtocolVersion.V3);
try
{
- client.connect(false);
+ client.connect(false, false);
Map reqMap;
diff --git a/test/unit/org/apache/cassandra/transport/frame/checksum/ChecksummingTransformerTest.java b/test/unit/org/apache/cassandra/transport/frame/checksum/ChecksummingTransformerTest.java
new file mode 100644
index 0000000000..82401b054e
--- /dev/null
+++ b/test/unit/org/apache/cassandra/transport/frame/checksum/ChecksummingTransformerTest.java
@@ -0,0 +1,224 @@
+/*
+ * 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.transport.frame.checksum;
+
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.Random;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.transport.Frame;
+import org.apache.cassandra.transport.ProtocolException;
+import org.apache.cassandra.transport.frame.compress.Compressor;
+import org.apache.cassandra.transport.frame.compress.LZ4Compressor;
+import org.apache.cassandra.transport.frame.compress.SnappyCompressor;
+import org.apache.cassandra.utils.ChecksumType;
+import org.apache.cassandra.utils.Pair;
+import org.quicktheories.core.Gen;
+
+import static org.quicktheories.QuickTheory.qt;
+import static org.quicktheories.generators.SourceDSL.*;
+
+public class ChecksummingTransformerTest
+{
+ private static final int DEFAULT_BLOCK_SIZE = 1 << 15;
+ private static final int MAX_INPUT_SIZE = 1 << 18;
+ private static final EnumSet FLAGS = EnumSet.of(Frame.Header.Flag.COMPRESSED, Frame.Header.Flag.CHECKSUMMED);
+
+ @BeforeClass
+ public static void init()
+ {
+ // required as static ChecksummingTransformer instances read default block size from config
+ DatabaseDescriptor.clientInitialization();
+ }
+
+ @Test
+ public void roundTripSafetyProperty()
+ {
+ qt().withExamples(500)
+ .forAll(inputs(),
+ compressors(),
+ checksumTypes(),
+ blocksizes())
+ .checkAssert(this::roundTrip);
+ }
+
+ @Test
+ public void roundTripZeroLengthInput()
+ {
+ qt().withExamples(20)
+ .forAll(zeroLengthInputs(),
+ compressors(),
+ checksumTypes(),
+ blocksizes())
+ .checkAssert(this::roundTrip);
+ }
+
+ @Test
+ public void corruptionCausesFailure()
+ {
+ qt().withExamples(500)
+ .forAll(inputWithCorruptablePosition(),
+ integers().between(0, Byte.MAX_VALUE).map(Integer::byteValue),
+ compressors(),
+ checksumTypes())
+ .checkAssert(this::roundTripWithCorruption);
+ }
+
+ private void roundTripWithCorruption(Pair inputAndCorruptablePosition,
+ byte corruptionValue,
+ Compressor compressor,
+ ChecksumType checksum)
+ {
+ String input = inputAndCorruptablePosition.left;
+ ByteBuf expectedBuf = Unpooled.wrappedBuffer(input.getBytes());
+ int byteToCorrupt = inputAndCorruptablePosition.right;
+ ChecksummingTransformer transformer = new ChecksummingTransformer(checksum, DEFAULT_BLOCK_SIZE, compressor);
+ ByteBuf outbound = transformer.transformOutbound(expectedBuf);
+
+ // make sure we're actually expecting to produce some corruption
+ if (outbound.getByte(byteToCorrupt) == corruptionValue)
+ return;
+
+ if (byteToCorrupt >= outbound.writerIndex())
+ return;
+
+ try
+ {
+ int oldIndex = outbound.writerIndex();
+ outbound.writerIndex(byteToCorrupt);
+ outbound.writeByte(corruptionValue);
+ outbound.writerIndex(oldIndex);
+ ByteBuf inbound = transformer.transformInbound(outbound, FLAGS);
+
+ // verify that the content was actually corrupted
+ expectedBuf.readerIndex(0);
+ Assert.assertEquals(expectedBuf, inbound);
+ } catch(ProtocolException e)
+ {
+ return;
+ }
+
+ }
+
+ @Test
+ public void roundTripWithSingleUncompressableChunk()
+ {
+ byte[] bytes = new byte[]{1};
+ ChecksummingTransformer transformer = new ChecksummingTransformer(ChecksumType.CRC32, DEFAULT_BLOCK_SIZE, LZ4Compressor.INSTANCE);
+ ByteBuf expectedBuf = Unpooled.wrappedBuffer(bytes);
+
+ ByteBuf outbound = transformer.transformOutbound(expectedBuf);
+ ByteBuf inbound = transformer.transformInbound(outbound, FLAGS);
+
+ // reset reader index on expectedBuf back to 0 as it will have been entirely consumed by the transformOutbound() call
+ expectedBuf.readerIndex(0);
+ Assert.assertEquals(expectedBuf, inbound);
+ }
+
+ @Test
+ public void roundTripWithCompressableAndUncompressableChunks() throws IOException
+ {
+ Compressor compressor = LZ4Compressor.INSTANCE;
+ Random random = new Random();
+ int inputLen = 127;
+
+ byte[] uncompressable = new byte[inputLen];
+ for (int i = 0; i < uncompressable.length; i++)
+ uncompressable[i] = (byte) random.nextInt(127);
+
+ byte[] compressed = new byte[compressor.maxCompressedLength(uncompressable.length)];
+ Assert.assertTrue(compressor.compress(uncompressable, 0, uncompressable.length, compressed, 0) > uncompressable.length);
+
+ byte[] compressable = new byte[inputLen];
+ for (int i = 0; i < compressable.length; i++)
+ compressable[i] = (byte)1;
+ Assert.assertTrue(compressor.compress(compressable, 0, compressable.length, compressable, 0) < compressable.length);
+
+ ChecksummingTransformer transformer = new ChecksummingTransformer(ChecksumType.CRC32, uncompressable.length, LZ4Compressor.INSTANCE);
+ byte[] expectedBytes = new byte[inputLen * 3];
+ ByteBuf expectedBuf = Unpooled.wrappedBuffer(expectedBytes);
+ expectedBuf.writerIndex(0);
+ expectedBuf.writeBytes(uncompressable);
+ expectedBuf.writeBytes(uncompressable);
+ expectedBuf.writeBytes(compressable);
+
+ ByteBuf outbound = transformer.transformOutbound(expectedBuf);
+ ByteBuf inbound = transformer.transformInbound(outbound, FLAGS);
+
+ // reset reader index on expectedBuf back to 0 as it will have been entirely consumed by the transformOutbound() call
+ expectedBuf.readerIndex(0);
+ Assert.assertEquals(expectedBuf, inbound);
+ }
+
+ private void roundTrip(String input, Compressor compressor, ChecksumType checksum, int blockSize)
+ {
+ ChecksummingTransformer transformer = new ChecksummingTransformer(checksum, blockSize, compressor);
+ byte[] expectedBytes = input.getBytes();
+ ByteBuf expectedBuf = Unpooled.wrappedBuffer(expectedBytes);
+
+ ByteBuf outbound = transformer.transformOutbound(expectedBuf);
+ ByteBuf inbound = transformer.transformInbound(outbound, FLAGS);
+
+ // reset reader index on expectedBuf back to 0 as it will have been entirely consumed by the transformOutbound() call
+ expectedBuf.readerIndex(0);
+ Assert.assertEquals(expectedBuf, inbound);
+ }
+
+ private Gen> inputWithCorruptablePosition()
+ {
+ // we only generate corruption for byte 2 onward. This is to skip introducing corruption in the number
+ // of chunks (which isn't checksummed
+ return inputs().flatMap(s -> integers().between(2, s.length() + 2).map(i -> Pair.create(s, i)));
+ }
+
+ private Gen inputs()
+ {
+ Gen randomStrings = strings().basicMultilingualPlaneAlphabet().ofLengthBetween(0, MAX_INPUT_SIZE);
+ Gen highlyCompressable = strings().betweenCodePoints('c', 'e').ofLengthBetween(1, MAX_INPUT_SIZE);
+ return randomStrings.mix(highlyCompressable, 50);
+ }
+
+ private Gen zeroLengthInputs()
+ {
+ return strings().ascii().ofLength(0);
+ }
+
+ private Gen compressors()
+ {
+ return arbitrary().pick(null, LZ4Compressor.INSTANCE, SnappyCompressor.INSTANCE);
+ }
+
+ private Gen checksumTypes()
+ {
+ return arbitrary().enumValuesWithNoOrder(ChecksumType.class);
+ }
+
+ private Gen blocksizes()
+ {
+ return arbitrary().constant(DEFAULT_BLOCK_SIZE);
+ }
+
+}
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java
index af354908b3..4bb5dda861 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java
@@ -88,7 +88,7 @@ public class StressSettings implements Serializable
{
String currentNode = node.randomNode();
SimpleClient client = new SimpleClient(currentNode, port.nativePort);
- client.connect(false);
+ client.connect(false, false);
client.execute("USE \"" + schema.keyspace + "\";", org.apache.cassandra.db.ConsistencyLevel.ONE);
return client;
}