From e32423159d9b31131cc95bccfcfa6b5dabbc070a Mon Sep 17 00:00:00 2001
From: Tyler Hobbs <tyler@datastax.com>
Date: Fri, 27 Jun 2014 09:28:18 -0500
Subject: [PATCH 1/2] Properly handle unrecognized opcodes and flags

Patch by Tyler Hobbs; reviewed by Sylvain Lebresne for CASSANDRA-7440
---
 CHANGES.txt                                          | 1 +
 src/java/org/apache/cassandra/transport/Frame.java   | 7 ++++---
 src/java/org/apache/cassandra/transport/Message.java | 2 ++
 3 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index 365722f766..d587a94b5c 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -2,6 +2,7 @@
  * Fix race in FileCacheService RemovalListener (CASSANDRA-7278)
  * Fix inconsistent use of consistencyForCommit that allowed LOCAL_QUORUM
    operations to incorrect become full QUORUM (CASSANDRA-7345)
+ * Properly handle unrecognized opcodes and flags (CASSANDRA-7440)
 
 
 2.0.9
diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Frame.java
index 6472b39ae3..89755df95b 100644
--- a/src/java/org/apache/cassandra/transport/Frame.java
+++ b/src/java/org/apache/cassandra/transport/Frame.java
@@ -92,14 +92,15 @@ public class Frame
             COMPRESSED,
             TRACING;
 
+            private static final Flag[] ALL_VALUES = values();
+
             public static EnumSet<Flag> deserialize(int flags)
             {
                 EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
-                Flag[] values = Flag.values();
-                for (int n = 0; n < 8; n++)
+                for (int n = 0; n < ALL_VALUES.length; n++)
                 {
                     if ((flags & (1 << n)) != 0)
-                        set.add(values[n]);
+                        set.add(ALL_VALUES[n]);
                 }
                 return set;
             }
diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java
index 07310812c9..eea3b48a90 100644
--- a/src/java/org/apache/cassandra/transport/Message.java
+++ b/src/java/org/apache/cassandra/transport/Message.java
@@ -103,6 +103,8 @@ public abstract class Message
 
         public static Type fromOpcode(int opcode, Direction direction)
         {
+            if (opcode >= opcodeIdx.length)
+                throw new ProtocolException(String.format("Unknown opcode %d", opcode));
             Type t = opcodeIdx[opcode];
             if (t == null)
                 throw new ProtocolException(String.format("Unknown opcode %d", opcode));

From 536ae7ae98a5901cf1aa6ae6ac60a17e2518fcb7 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Fri, 27 Jun 2014 10:26:29 -0500
Subject: [PATCH 2/2] close CqlRecordWriter clients when finished patch by
 Peter Williams; reviewed by jbellis for CASSANDRA-7459

---
 CHANGES.txt                                                    | 1 +
 src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java | 3 +++
 2 files changed, 4 insertions(+)

diff --git a/CHANGES.txt b/CHANGES.txt
index d587a94b5c..e722392acc 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -3,6 +3,7 @@
  * Fix inconsistent use of consistencyForCommit that allowed LOCAL_QUORUM
    operations to incorrect become full QUORUM (CASSANDRA-7345)
  * Properly handle unrecognized opcodes and flags (CASSANDRA-7440)
+ * (Hadoop) close CqlRecordWriter clients when finished (CASSANDRA-7459)
 
 
 2.0.9
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
index ee7aabf717..b967494a86 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordWriter.java
@@ -279,6 +279,9 @@ final class CqlRecordWriter extends AbstractColumnFamilyRecordWriter<Map<String,
                     }
                 }
             }
+
+            // close all our connections once we are done.
+            closeInternal();
         }
 
         /** get prepared statement id from cache, otherwise prepare it from Cassandra server*/