fix size calculation

processing/src/main/java/io/druid/segment/incremental/OffheapIncrementalIndex.java

@@ -231,7 +231,7 @@ public class OffheapIncrementalIndex extends IncrementalIndex<BufferAggregator>
    */
   public boolean isFull()
   {
-    return (size() + 1) * totalAggSize > bufferHolder.get().limit() || getCurrentSize() > sizeLimit ;
+    return getCurrentSize() > sizeLimit;
   }
 
   private int getMetricPosition(int rowOffset, int metricIndex)
@@ -426,6 +426,9 @@ public class OffheapIncrementalIndex extends IncrementalIndex<BufferAggregator>
 
   private long getCurrentSize()
   {
-    return Store.forDB(db).getCurrSize() + Store.forDB(factsDb).getCurrSize() + bufferHolder.get().limit();
+    return Store.forDB(db).getCurrSize() +
+           Store.forDB(factsDb).getCurrSize()
+           // Size of aggregators
+           + (size() + 1) * totalAggSize;
   }
 }

processing/src/test/java/io/druid/segment/data/IncrementalIndexTest.java

@@ -208,4 +208,26 @@ public class IncrementalIndexTest
     }
     Assert.assertEquals(elementsPerThread, curr);
   }
+
+  @Test
+  public void testOffheapIndexIsFull()
+  {
+    OffheapIncrementalIndex index = new OffheapIncrementalIndex(
+        0L,
+        QueryGranularity.NONE,
+        new AggregatorFactory[]{new CountAggregatorFactory("count")},
+        TestQueryRunners.pool,
+        true,
+        2 * 1024 * 1024
+    );
+    int rowCount = 0;
+    for (int i = 0; i < 500; i++) {
+      rowCount = index.add(getRow(System.currentTimeMillis(), i, 100));
+      if (index.isFull()) {
+        break;
+      }
+    }
+
+    Assert.assertTrue("rowCount : " + rowCount, rowCount > 200 && rowCount < 600);
+  }
 }