diff --git a/CHANGES.txt b/CHANGES.txt index ec5e23313b..a3e52f78dc 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -19,6 +19,7 @@ * cqlsh: Fix keys() and full() collection indexes in DESCRIBE output (CASSANDRA-8154) Merged from 2.0: + * Preserve stream ID for more protocol errors (CASSANDRA-8848) * Fix combining token() function with multi-column relations on clustering columns (CASSANDRA-8797) * Make CFS.markReferenced() resistant to bad refcounting (CASSANDRA-8829) @@ -29,7 +30,8 @@ Merged from 2.0: * Make it possible to set max_sstable_age to fractional days (CASSANDRA-8406) * Fix some multi-column relations with indexes on some clustering columns (CASSANDRA-8275) - * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate() (CASSANDRA-8748) + * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate() + (CASSANDRA-8748) * Throw OOM if allocating memory fails to return a valid pointer (CASSANDRA-8726) * Fix SSTableSimpleUnsortedWriter ConcurrentModificationException (CASSANDRA-8619) * 'nodetool info' prints exception against older node (CASSANDRA-8796) diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Frame.java index 01bee10989..2868ed44fd 100644 --- a/src/java/org/apache/cassandra/transport/Frame.java +++ b/src/java/org/apache/cassandra/transport/Frame.java @@ -204,7 +204,15 @@ public class Frame } // This throws a protocol exceptions if the opcode is unknown - Message.Type type = Message.Type.fromOpcode(buffer.getByte(idx++), direction); + Message.Type type; + try + { + type = Message.Type.fromOpcode(buffer.getByte(idx++), direction); + } + catch (ProtocolException e) + { + throw ErrorMessage.wrap(e, streamId); + } long bodyLength = buffer.getUnsignedInt(idx); idx += Header.BODY_LENGTH_SIZE; @@ -212,7 +220,7 @@ public class Frame if (bodyLength < 0) { buffer.skipBytes(headerLength); - throw new ProtocolException("Invalid frame body length: " + bodyLength); + throw ErrorMessage.wrap(new ProtocolException("Invalid frame body length: " + bodyLength), streamId); } long frameLength = bodyLength + headerLength; @@ -247,7 +255,11 @@ public class Frame } else if (connection.getVersion() != version) { - throw new ProtocolException(String.format("Invalid message version. Got %d but previous messages on this connection had version %d", version, connection.getVersion())); + throw ErrorMessage.wrap( + new ProtocolException(String.format( + "Invalid message version. Got %d but previous messages on this connection had version %d", + version, connection.getVersion())), + streamId); } results.add(new Frame(new Header(version, flags, streamId, type), body)); diff --git a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java index 7e4a3a97d2..a049a57937 100644 --- a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.transport.messages; +import com.google.common.annotations.VisibleForTesting; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.CodecException; import com.google.common.base.Predicate; @@ -256,7 +257,7 @@ public class ErrorMessage extends Message.Response return new WrappedException(t, streamId); } - private static class WrappedException extends RuntimeException + public static class WrappedException extends RuntimeException { private final int streamId; @@ -265,6 +266,12 @@ public class ErrorMessage extends Message.Response super(cause); this.streamId = streamId; } + + @VisibleForTesting + public int getStreamId() + { + return this.streamId; + } } } diff --git a/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java new file mode 100644 index 0000000000..91f7355fa9 --- /dev/null +++ b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java @@ -0,0 +1,103 @@ +/* + * 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; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import org.apache.cassandra.transport.messages.ErrorMessage; +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +public class ProtocolErrorTest { + + @Test + public void testInvalidDirection() throws Exception + { + Frame.Decoder dec = new Frame.Decoder(null); + + List results = new ArrayList<>(); + // should generate a protocol exception for using a response frame with + // a prepare op, ensure that it comes back with stream ID 1 + byte[] frame = new byte[] { + (byte) 0x82, // direction & version + 0x00, // flags + 0x01, // stream ID + 0x09, // opcode + 0x00, 0x00, 0x00, 0x21, // body length + 0x00, 0x00, 0x00, 0x1b, 0x00, 0x1b, 0x53, 0x45, + 0x4c, 0x45, 0x43, 0x54, 0x20, 0x2a, 0x20, 0x46, + 0x52, 0x4f, 0x4d, 0x20, 0x73, 0x79, 0x73, 0x74, + 0x65, 0x6d, 0x2e, 0x6c, 0x6f, 0x63, 0x61, 0x6c, + 0x3b + }; + ByteBuf buf = Unpooled.wrappedBuffer(frame); + try { + dec.decode(null, buf, results); + } catch (ErrorMessage.WrappedException e) { + // make sure the exception has the correct stream ID + Assert.assertEquals(1, e.getStreamId()); + } + } + + @Test + public void testNegativeBodyLength() throws Exception + { + Frame.Decoder dec = new Frame.Decoder(null); + + List results = new ArrayList<>(); + byte[] frame = new byte[] { + (byte) 0x82, // direction & version + 0x00, // flags + 0x01, // stream ID + 0x09, // opcode + (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff, // body length (-1) + }; + ByteBuf buf = Unpooled.wrappedBuffer(frame); + try { + dec.decode(null, buf, results); + } catch (ErrorMessage.WrappedException e) { + // make sure the exception has the correct stream ID + Assert.assertEquals(1, e.getStreamId()); + } + } + + @Test + public void testBodyLengthOverLimit() throws Exception + { + Frame.Decoder dec = new Frame.Decoder(null); + + List results = new ArrayList<>(); + byte[] frame = new byte[] { + (byte) 0x82, // direction & version + 0x00, // flags + 0x01, // stream ID + 0x09, // opcode + 0x7f, (byte) 0xff, (byte) 0xff, (byte) 0xff, // body length + }; + ByteBuf buf = Unpooled.wrappedBuffer(frame); + try { + dec.decode(null, buf, results); + } catch (ErrorMessage.WrappedException e) { + // make sure the exception has the correct stream ID + Assert.assertEquals(1, e.getStreamId()); + } + } +}