@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 testPartitionerBucketIdNullFail() {
    PartitionSpec partitionSpec = TableSchemaType.ONE_BUCKET.getPartitionSpec(DEFAULT_NUM_BUCKETS);
    BucketPartitioner bucketPartitioner = new BucketPartitioner(partitionSpec);

    assertThatExceptionOfType(RuntimeException.class)
        .isThrownBy(() -> bucketPartitioner.partition(null, DEFAULT_NUM_BUCKETS))
        .withMessage(BUCKET_NULL_MESSAGE);
  }