From 3dff61a08271275d3c1121ed2dba1ea462336baf Mon Sep 17 00:00:00 2001
From: David Arthur <mumrah@gmail.com>
Date: Thu, 3 Jan 2013 12:49:24 -0500
Subject: [PATCH 1/2] KAFKA-680 Use length of encoded byte array

Instead of the String length in ApiUtils#writeShortString
---
 core/src/main/scala/kafka/api/ApiUtils.scala |   13 ++++++++-----
 1 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/kafka/api/ApiUtils.scala b/core/src/main/scala/kafka/api/ApiUtils.scala
index ba1d199..a6b0328 100644
--- a/core/src/main/scala/kafka/api/ApiUtils.scala
+++ b/core/src/main/scala/kafka/api/ApiUtils.scala
@@ -31,11 +31,14 @@ object ApiUtils {
   def writeShortString(buffer: ByteBuffer, string: String) {
     if(string == null) {
       buffer.putShort(-1)
-    } else if(string.length > Short.MaxValue) {
-      throw new KafkaException("String exceeds the maximum size of " + Short.MaxValue + ".")
     } else {
-      buffer.putShort(string.length.asInstanceOf[Short])
-      buffer.put(string.getBytes(ProtocolEncoding))
+      val encodedString = string.getBytes(ProtocolEncoding)
+      if(encodedString.length > Short.MaxValue) {
+        throw new KafkaException("String exceeds the maximum size of " + Short.MaxValue + ".")
+      } else {
+        buffer.putShort(encodedString.length.asInstanceOf[Short])
+        buffer.put(encodedString)
+      }
     }
   }
   
@@ -89,4 +92,4 @@ object ApiUtils {
     else value
   }
   
-}
\ No newline at end of file
+}
-- 
1.7.5.4


From 0beac5f3714e9d244522c00b166d6f43865980e8 Mon Sep 17 00:00:00 2001
From: David Arthur <mumrah@gmail.com>
Date: Fri, 4 Jan 2013 15:48:29 -0500
Subject: [PATCH 2/2] KAFKA-680 Adding some unit tests for ApiUtils

---
 .../test/scala/unit/kafka/api/ApiUtilsTest.scala   |   84 ++++++++++++++++++++
 1 files changed, 84 insertions(+), 0 deletions(-)
 create mode 100644 core/src/test/scala/unit/kafka/api/ApiUtilsTest.scala

diff --git a/core/src/test/scala/unit/kafka/api/ApiUtilsTest.scala b/core/src/test/scala/unit/kafka/api/ApiUtilsTest.scala
new file mode 100644
index 0000000..2554425
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/api/ApiUtilsTest.scala
@@ -0,0 +1,84 @@
+/**
+ * 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 kafka.api
+
+import org.junit._
+import org.scalatest.junit.JUnitSuite
+import junit.framework.Assert._
+import scala.util.Random
+import java.nio.ByteBuffer
+import kafka.common.KafkaException
+import kafka.utils.TestUtils
+
+object ApiUtilsTest {
+  val rnd: Random = new Random()
+}
+
+class ApiUtilsTest extends JUnitSuite {
+
+  @Test
+  def testShortStringNonASCII() {
+    // Random-length strings
+    for(i <- 0 to 100) {
+      // Since we're using UTF-8 encoding, each encoded byte will be one to four bytes long 
+      val s: String = ApiUtilsTest.rnd.nextString(math.abs(ApiUtilsTest.rnd.nextInt()) % (Short.MaxValue / 4))  
+      val bb: ByteBuffer = ByteBuffer.allocate(ApiUtils.shortStringLength(s))
+      ApiUtils.writeShortString(bb, s)
+      bb.rewind()
+      assertEquals(s, ApiUtils.readShortString(bb))
+    }
+  }
+
+  @Test
+  def testShortStringASCII() {
+    // Random-length strings
+    for(i <- 0 to 100) {
+      val s: String = TestUtils.randomString(math.abs(ApiUtilsTest.rnd.nextInt()) % Short.MaxValue)  
+      val bb: ByteBuffer = ByteBuffer.allocate(ApiUtils.shortStringLength(s))
+      ApiUtils.writeShortString(bb, s)
+      bb.rewind()
+      assertEquals(s, ApiUtils.readShortString(bb))
+    }
+
+    // Max size string
+    val s1: String = TestUtils.randomString(Short.MaxValue)
+    val bb: ByteBuffer = ByteBuffer.allocate(ApiUtils.shortStringLength(s1))
+    ApiUtils.writeShortString(bb, s1)
+    bb.rewind()
+    assertEquals(s1, ApiUtils.readShortString(bb))
+
+    // One byte too big
+    val s2: String = TestUtils.randomString(Short.MaxValue + 1) 
+    try {
+      ApiUtils.shortStringLength(s2)
+      fail
+    } catch {
+      case e: KafkaException => {
+        // ok
+      }
+    }
+    try {
+      ApiUtils.writeShortString(bb, s2)
+      fail
+    } catch {
+      case e: KafkaException => {
+        // ok
+      }
+    }
+  }
+}
-- 
1.7.5.4