additional check for offheap bloom filter size

patch by Vijay; reviewed by jbellis for CASSANDRA-5903

src/java/org/apache/cassandra/utils/obs/OffHeapBitSet.java

@@ -36,9 +36,12 @@ public class OffHeapBitSet implements IBitSet
     public OffHeapBitSet(long numBits)
     {
         // OpenBitSet.bits2words calculation is there for backward compatibility.
-        long byteCount = OpenBitSet.bits2words(numBits) * 8L;
+        long wordCount = OpenBitSet.bits2words(numBits);
+        if (wordCount > Integer.MAX_VALUE)
+            throw new UnsupportedOperationException("Bloom filter size is > 16GB, reduce the bloom_filter_fp_chance");
         try
         {
+            long byteCount = wordCount * 8L;
             bytes = RefCountedMemory.allocate(byteCount);
         }
         catch (OutOfMemoryError e)

test/unit/org/apache/cassandra/utils/BloomFilterTest.java

@@ -20,16 +20,25 @@ package org.apache.cassandra.utils;
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
 
+import junit.framework.Assert;
+
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.FilterFactory.Type;
 
 public class BloomFilterTest
 {
@@ -138,4 +147,31 @@ public class BloomFilterTest
     {
         testManyHashes(FilterTestHelper.randomKeys());
     }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testOffHeapException()
+    {
+        long numKeys = (Integer.MAX_VALUE * 64) + 1; // approx 128 Billion
+        FilterFactory.getFilter(numKeys, 0.01d, true);
+    }
+
+    @Test
+    @Ignore
+    public void testHugeBFSerialization() throws IOException
+    {
+        ByteBuffer test = ByteBuffer.wrap(new byte[] {0, 1});
+
+        File file = FileUtils.createTempFile("bloomFilterTest-", ".dat");
+        BloomFilter filter = (BloomFilter) FilterFactory.getFilter(((long)Integer.MAX_VALUE / 8) + 1, 0.01d, true);
+        filter.add(test);
+        DataOutputStream out = new DataOutputStream(new FileOutputStream(file));
+        FilterFactory.serialize(filter, out);
+        filter.bitset.serialize(out);
+        out.close();
+        
+        DataInputStream in = new DataInputStream(new FileInputStream(file));
+        BloomFilter filter2 = (BloomFilter) FilterFactory.deserialize(in, Type.MURMUR3, true);
+        Assert.assertTrue(filter2.isPresent(test));
+        FileUtils.closeQuietly(in);
+    }
 }