MINOR: Fix BaseHashTable sizing

metadata/src/main/java/org/apache/kafka/timeline/BaseHashTable.java

@@ -40,14 +40,19 @@ class BaseHashTable<T> {
     private final static double MAX_LOAD_FACTOR = 0.75f;
 
     /**
-     * The natural log of 2
+     * The minimum number of slots we can have in the hash table.
      */
-    private final static double LN_2 = Math.log(2);
+    final static int MIN_CAPACITY = 2;
+
+    /**
+     * The maximum 31-bit power of two.
+     */
+    private final static int MAX_SIGNED_POWER_OF_TWO = 0x4000_0000;
 
     /**
      * The maximum number of slots we can have in the hash table.
      */
-    private final static int MAX_CAPACITY = 0x4000000;
+    final static int MAX_CAPACITY = MAX_SIGNED_POWER_OF_TWO;
 
     private Object[] elements;
     private int size = 0;
@@ -56,12 +61,27 @@ class BaseHashTable<T> {
         this.elements = new Object[expectedSizeToCapacity(expectedSize)];
     }
 
+    /**
+     * Calculate the capacity we should provision, given the expected size.
+     *
+     * Our capacity must always be a power of 2, and never less than 2.
+     */
     static int expectedSizeToCapacity(int expectedSize) {
-        if (expectedSize <= 1) {
-            return 2;
+        if (expectedSize >= MAX_CAPACITY / 2) {
+            return MAX_CAPACITY;
+        }
+        return Math.max(MIN_CAPACITY, roundUpToPowerOfTwo(expectedSize * 2));
+    }
+
+    private static int roundUpToPowerOfTwo(int i) {
+        if (i <= 0) {
+            return 0;
+        } else if (i > MAX_SIGNED_POWER_OF_TWO) {
+            throw new ArithmeticException("There is no 31-bit power of two higher than " +
+                    "or equal to " + i);
+        } else {
+            return 1 << -Integer.numberOfLeadingZeros(i - 1);
         }
-        double sizeToFit = expectedSize / MAX_LOAD_FACTOR;
-        return (int) Math.min(MAX_CAPACITY, Math.ceil(Math.log(sizeToFit) / LN_2));
     }
 
     final int baseSize() {

metadata/src/test/java/org/apache/kafka/timeline/BaseHashTableTest.java

@@ -121,4 +121,23 @@ public void testExpansion() {
             assertEquals(Integer.valueOf(i), table.baseRemove(Integer.valueOf(i)));
         }
     }
+
+    @Test
+    public void testExpectedSizeToCapacity() {
+        assertEquals(2, BaseHashTable.expectedSizeToCapacity(Integer.MIN_VALUE));
+        assertEquals(2, BaseHashTable.expectedSizeToCapacity(-123));
+        assertEquals(2, BaseHashTable.expectedSizeToCapacity(0));
+        assertEquals(2, BaseHashTable.expectedSizeToCapacity(1));
+        assertEquals(4, BaseHashTable.expectedSizeToCapacity(2));
+        assertEquals(8, BaseHashTable.expectedSizeToCapacity(3));
+        assertEquals(8, BaseHashTable.expectedSizeToCapacity(4));
+        assertEquals(16, BaseHashTable.expectedSizeToCapacity(5));
+        assertEquals(0x4000000, BaseHashTable.expectedSizeToCapacity(0x1010400));
+        assertEquals(0x4000000, BaseHashTable.expectedSizeToCapacity(0x2000000));
+        assertEquals(0x8000000, BaseHashTable.expectedSizeToCapacity(0x2000001));
+        assertEquals(BaseHashTable.MAX_CAPACITY, BaseHashTable.expectedSizeToCapacity(BaseHashTable.MAX_CAPACITY));
+        assertEquals(BaseHashTable.MAX_CAPACITY, BaseHashTable.expectedSizeToCapacity(BaseHashTable.MAX_CAPACITY + 1));
+        assertEquals(BaseHashTable.MAX_CAPACITY, BaseHashTable.expectedSizeToCapacity(Integer.MAX_VALUE - 1));
+        assertEquals(BaseHashTable.MAX_CAPACITY, BaseHashTable.expectedSizeToCapacity(Integer.MAX_VALUE));
+    }
 }