KAFKA-3682; ArrayIndexOutOfBoundsException thrown by

SkimpyOffsetMap.get() when full

Limited number of attempts to number of map slots after the internal
positionOf() goes into linear search mode.
Added unit test

Co-developed with mimaison

Author: edoardo <ecomar@uk.ibm.com>

Reviewers: Jun Rao <junrao@gmail.com>

Closes #1352 from edoardocomar/KAFKA-3682

core/src/main/scala/kafka/log/OffsetMap.scala

@@ -118,7 +118,12 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm: String = "MD5") extend
     // search for the hash of this key by repeated probing until we find the hash we are looking for or we find an empty slot
     var attempt = 0
     var pos = 0
+    //we need to guard against attempt integer overflow if the map is full
+    //limit attempt to number of slots once positionOf(..) enters linear search mode
+    val maxAttempts = slots + hashSize - 4
     do {
+     if(attempt >= maxAttempts)
+        return -1L
       pos = positionOf(hash1, attempt)
       bytes.position(pos)
       if(isEmpty(pos))

core/src/test/scala/unit/kafka/log/OffsetMapTest.scala

@@ -44,7 +44,19 @@ class OffsetMapTest extends JUnitSuite {
       assertEquals(map.get(key(i)), -1L)
   }
 
-  def key(key: Int) = ByteBuffer.wrap(key.toString.getBytes)
+  @Test
+  def testGetWhenFull() {
+    val map = new SkimpyOffsetMap(4096)
+    var i = 37L  //any value would do
+    while (map.size < map.slots) {
+      map.put(key(i), i)
+      i = i + 1L
+    }
+    assertEquals(map.get(key(i)), -1L)
+    assertEquals(map.get(key(i-1L)), i-1L)
+  }
+
+  def key(key: Long) = ByteBuffer.wrap(key.toString.getBytes)
 
   def validateMap(items: Int, loadFactor: Double = 0.5): SkimpyOffsetMap = {
     val map = new SkimpyOffsetMap((items/loadFactor * 24).toInt)