@Override
  public int partition(StatisticsOrRecord wrapper, int numPartitions) {
    if (wrapper.hasStatistics()) {
      this.delegatePartitioner = delegatePartitioner(wrapper.statistics());
      return (int) (roundRobinCounter(numPartitions).getAndIncrement() % numPartitions);
    } else {
      if (delegatePartitioner != null) {
        return delegatePartitioner.partition(wrapper.record(), numPartitions);
      } else {
        int partition = (int) (roundRobinCounter(numPartitions).getAndIncrement() % numPartitions);
        LOG.trace("Statistics not available. Round robin to partition {}", partition);
        return partition;
      }
    }
  }
@Test
  public void testRoundRobinStatisticsWrapper() {
    RangePartitioner partitioner = new RangePartitioner(SCHEMA, SORT_ORDER);
    Set<Integer> results = Sets.newHashSetWithExpectedSize(numPartitions);
    for (int i = 0; i < numPartitions; ++i) {
      GlobalStatistics statistics =
          GlobalStatistics.fromRangeBounds(1L, new SortKey[] {CHAR_KEYS.get("a")});
      results.add(
          partitioner.partition(StatisticsOrRecord.fromStatistics(statistics), numPartitions));
    }

    // round-robin. every partition should get an assignment
    assertThat(results).containsExactlyInAnyOrder(0, 1, 2, 3);
  }