@VisibleForTesting
  protected int sizeIncrement(String partitionKey, @Nullable String hashKey, byte[] record) {
    final int[] size = {0, 0}; // wrapper & record
    sizeIncrementOfKey(size, partitionKey, partitionKeys); // partition key encoding
    if (hashKey != null) {
      sizeIncrementOfKey(size, hashKey, explicitHashKeys); // optional hash key encoding
    }
    if (record != null) {
      size[1] += 1 + VarInt.getLength(record.length) + record.length; // record encoding
    }
    // wrapper with partition / hash key table + record with partition / hash key reference
    return size[0] + 1 + VarInt.getLength(size[1]) + size[1];
  }
@Test
  public void testSizeIncrement() {
    Random rnd = new Random();
    List<String> keys = Stream.generate(() -> randomAscii(1, 256)).limit(3).collect(toList());
    List<String> hashKeys = Stream.generate(() -> randomNumeric(1, 38)).limit(3).collect(toList());
    hashKeys.add(null);

    int sizeBytes = BASE_OVERHEAD;
    int increment = 0;

    String pk, ehk, data;
    do {
      int currentBytes = aggregator.toBytes().length;
      assertThat(increment).isEqualTo(currentBytes - sizeBytes);
      assertThat(aggregator.getSizeBytes()).isEqualTo(currentBytes);
      sizeBytes = currentBytes;

      pk = keys.get(rnd.nextInt(keys.size()));
      ehk = hashKeys.get(rnd.nextInt(hashKeys.size()));
      data = randomAscii(256, 512);
      increment = aggregator.sizeIncrement(pk, ehk, data.getBytes(UTF_8));
    } while (aggregator.addRecord(pk, ehk, data.getBytes(UTF_8)));
  }