bug: Revert 281f5fc fix: Turn off bitset length fatal check

#35
RuntimeException when running with very high options in 0.2.0.0 (Bitset too long to encode) #35

This is due to Support BitSet encoding lengths longer than Short.MAX_VALUE #37.
Release 0.2.0.1 was put out incorrectly removing the check, but then it suddenly occurred to me what it was really checking.

However, this situation should not have arose because of the MAX values used in the test, there should never have been that many messages uncommitted anyway.
So two issues to fix here - first the reason there were so many to encode in the first place, and second upgrade the BitSet encoder to use Integers.

#37
Support BitSet encoding lengths longer than Short.MAX_VALUE #37

Author: astubbs

CHANGELOG.adoc

@@ -1,5 +1,15 @@
 = Change Log
 
+== v0.2.0.2
+
+* Fixes
+** Turns back on the https://github.com/confluentinc/parallel-consumer/issues/35[Bitset overflow check (#35)]
+
+== v0.2.0.1 DO NOT USE
+
+* Fixes
+** Incorrectly turns off an over-flow check in https://github.com/confluentinc/parallel-consumer/issues/35[offset serialisation system (#35)]
+
 == v0.2.0.0
 
 * Features:

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/OffsetSimultaneousEncoder.java

@@ -211,6 +211,10 @@ private class BitsetEncoder extends Encoder {
         public BitsetEncoder(final int length) {
             // prep bit set buffer
             this.wrappedBitsetBytesBuffer = ByteBuffer.allocate(Short.BYTES + ((length / 8) + 1));
+            if (length > Short.MAX_VALUE) {
+                // need to upgrade to using Integer for the bitset length, but can't change serialisation format in-place
+                throw new RuntimeException("Bitset too long to encode, bitset length overflows Short.MAX_VALUE: " + length + ". (max: " + Short.MAX_VALUE + ")");
+            }
             // bitset doesn't serialise it's set capacity, so we have to as the unused capacity actually means something
             this.wrappedBitsetBytesBuffer.putShort((short) length);
             bitSet = new BitSet(length);