MINOR: Fix BaseHashTable sizing

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

@@ -40,14 +40,14 @@ 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 number of slots we can have in the hash table.
      */
-    private final static int MAX_CAPACITY = 0x4000000;
+    final static int MAX_CAPACITY = 1 << 30;
 
     private Object[] elements;
     private int size = 0;
@@ -56,12 +56,28 @@ 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 or more
+     * than MAX_CAPACITY.  We use 64-bit numbers here to avoid overflow
+     * concerns.
+     */
     static int expectedSizeToCapacity(int expectedSize) {
-        if (expectedSize <= 1) {
-            return 2;
+        long minCapacity = (long) Math.ceil((float) expectedSize / MAX_LOAD_FACTOR);
+        return Math.max(MIN_CAPACITY,
+                (int) Math.min(MAX_CAPACITY, roundUpToPowerOfTwo(minCapacity)));
+    }
+
+    private static long roundUpToPowerOfTwo(long i) {
+        if (i <= 0) {
+            return 0;
+        } else if (i > (1L << 62)) {
+            throw new ArithmeticException("There are no 63-bit powers of 2 higher than " +
+                    "or equal to " + i);
+        } else {
+            return 1L << -Long.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,24 @@ 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(4, BaseHashTable.expectedSizeToCapacity(3));
+        assertEquals(8, BaseHashTable.expectedSizeToCapacity(4));
+        assertEquals(16, BaseHashTable.expectedSizeToCapacity(12));
+        assertEquals(32, BaseHashTable.expectedSizeToCapacity(13));
+        assertEquals(0x2000000, BaseHashTable.expectedSizeToCapacity(0x1010400));
+        assertEquals(0x4000000, BaseHashTable.expectedSizeToCapacity(0x2000000));
+        assertEquals(0x4000000, 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));
+    }
 }