cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tylerho...@apache.org
Subject [1/3] cassandra git commit: Preserve stream ID for more protocol errors
Date Fri, 27 Feb 2015 22:19:50 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk aa7c7362a -> 369ed3e1d


Preserve stream ID for more protocol errors

Patch by Chris Bannister and Tyler Hobbs for CASSANDRA-8848


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

Branch: refs/heads/trunk
Commit: 5654e7368c2d68b3701cb5cdf190975f1079b10c
Parents: 98502e0
Author: Chris Bannister <c.bannister@gmail.com>
Authored: Fri Feb 27 16:09:26 2015 -0600
Committer: Tyler Hobbs <tylerhobbs@apache.org>
Committed: Fri Feb 27 16:09:26 2015 -0600

----------------------------------------------------------------------
 CHANGES.txt                                     |  4 +-
 .../org/apache/cassandra/transport/Frame.java   | 18 +++-
 .../transport/messages/ErrorMessage.java        |  9 +-
 .../cassandra/transport/ProtocolErrorTest.java  | 97 ++++++++++++++++++++
 4 files changed, 123 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/5654e736/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index f2b4469..3a8d824 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.0.13:
+ * 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)
@@ -7,7 +8,8 @@
    table with ASC ordering and paging (CASSANDRA-8767)
  * AssertionError: "Memory was freed" when running cleanup (CASSANDRA-8716)
  * Make it possible to set max_sstable_age to fractional days (CASSANDRA-8406)
- * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate() (CASSANDRA-8748)
+ * Fix memory leak in SSTableSimple*Writer and SSTableReader.validate()
+   (CASSANDRA-8748)
  * Fix some multi-column relations with indexes on some clustering
    columns (CASSANDRA-8275)
  * Fix IllegalArgumentException in dynamic snitch (CASSANDRA-8448)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5654e736/src/java/org/apache/cassandra/transport/Frame.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Frame.java
index 89755df..7520c41 100644
--- a/src/java/org/apache/cassandra/transport/Frame.java
+++ b/src/java/org/apache/cassandra/transport/Frame.java
@@ -166,14 +166,22 @@ public class Frame
             int streamId = buffer.getByte(idx + 2);
 
             // This throws a protocol exceptions if the opcode is unknown
-            Message.Type type = Message.Type.fromOpcode(buffer.getByte(idx + 3), direction);
+            Message.Type type;
+            try
+            {
+                type = Message.Type.fromOpcode(buffer.getByte(idx + 3), direction);
+            }
+            catch (ProtocolException e)
+            {
+                throw ErrorMessage.wrap(e, streamId);
+            }
 
             long bodyLength = buffer.getUnsignedInt(idx + Header.BODY_LENGTH_OFFSET);
 
             if (bodyLength < 0)
             {
                 buffer.skipBytes(Header.LENGTH);
-                throw new ProtocolException("Invalid frame body length: " + bodyLength);
+                throw ErrorMessage.wrap(new ProtocolException("Invalid frame body length:
" + bodyLength), streamId);
             }
 
             long frameLength = bodyLength + Header.LENGTH;
@@ -207,7 +215,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);
             }
 
             return new Frame(new Header(version, flags, streamId, type), body);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5654e736/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java
index 4d60a1f..e27fb88 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 com.google.common.base.Predicate;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.slf4j.Logger;
@@ -243,7 +244,8 @@ public class ErrorMessage extends Message.Response
         return new WrappedException(t, streamId);
     }
 
-    private static class WrappedException extends RuntimeException
+    @VisibleForTesting
+    public static class WrappedException extends RuntimeException
     {
         private final int streamId;
 
@@ -252,6 +254,11 @@ public class ErrorMessage extends Message.Response
             super(cause);
             this.streamId = streamId;
         }
+
+        public int getStreamId()
+        {
+            return this.streamId;
+        }
     }
 
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5654e736/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java
----------------------------------------------------------------------
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 0000000..387d159
--- /dev/null
+++ b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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 org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.HeapChannelBufferFactory;
+import org.apache.cassandra.transport.messages.ErrorMessage;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ProtocolErrorTest {
+
+    @Test
+    public void testInvalidDirection() throws Exception
+    {
+        Frame.Decoder dec = new Frame.Decoder(null);
+
+        // 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
+        };
+        ChannelBuffer buf = new HeapChannelBufferFactory().getBuffer(frame, 0, frame.length);
+        try {
+            dec.decode(null, null, buf);
+        } 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);
+
+        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)
+        };
+        ChannelBuffer buf = new HeapChannelBufferFactory().getBuffer(frame, 0, frame.length);
+        try {
+            dec.decode(null, null, buf);
+        } 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);
+
+        byte[] frame = new byte[] {
+                (byte) 0x82,  // direction & version
+                0x00,  // flags
+                0x01,  // stream ID
+                0x09,  // opcode
+                0x7f, (byte) 0xff, (byte) 0xff, (byte) 0xff,  // body length
+        };
+        ChannelBuffer buf = new HeapChannelBufferFactory().getBuffer(frame, 0, frame.length);
+        try {
+            dec.decode(null, null, buf);
+        } catch (ErrorMessage.WrappedException e) {
+            // make sure the exception has the correct stream ID
+            Assert.assertEquals(1, e.getStreamId());
+        }
+    }
+}


Mime
View raw message