@Override
  public int partition(Integer bucketId, int numPartitions) {
    Preconditions.checkNotNull(bucketId, BUCKET_NULL_MESSAGE);
    Preconditions.checkArgument(bucketId >= 0, BUCKET_LESS_THAN_LOWER_BOUND_MESSAGE, bucketId);
    Preconditions.checkArgument(
        bucketId < maxNumBuckets, BUCKET_GREATER_THAN_UPPER_BOUND_MESSAGE, bucketId, maxNumBuckets);

    if (numPartitions <= maxNumBuckets) {
      return bucketId % numPartitions;
    } else {
      return getPartitionWithMoreWritersThanBuckets(bucketId, numPartitions);
    }
  }
@Test
  public void testPartitionerBucketIdOutOfRangeFail() {
    PartitionSpec partitionSpec = TableSchemaType.ONE_BUCKET.getPartitionSpec(DEFAULT_NUM_BUCKETS);
    BucketPartitioner bucketPartitioner = new BucketPartitioner(partitionSpec);

    int negativeBucketId = -1;
    assertThatExceptionOfType(IllegalArgumentException.class)
        .isThrownBy(() -> bucketPartitioner.partition(negativeBucketId, 1))
        .withMessage(BUCKET_LESS_THAN_LOWER_BOUND_MESSAGE, negativeBucketId);

    int tooBigBucketId = DEFAULT_NUM_BUCKETS;
    assertThatExceptionOfType(IllegalArgumentException.class)
        .isThrownBy(() -> bucketPartitioner.partition(tooBigBucketId, 1))
        .withMessage(BUCKET_GREATER_THAN_UPPER_BOUND_MESSAGE, tooBigBucketId, DEFAULT_NUM_BUCKETS);
  }