### Question:
FairPartitioner implements Partitioner { @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { return getPartition(topic, cluster, getTimeHash() + getKeyHash(key)); }  @Override void configure(Map<String, ?> configs); @Override int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster); @Override void close(); static final int ROTATE_MILLIS; static final int ROTATE_OFFSET; }### Answer:
@Test public void partitionAvailable() { int partition = partitioner.partition(topic, key, keyBytes, value, valueBytes, cluster); assertThat(partition, is(lessThan(allPartitions.size()))); assertThat(partition, is(greaterThanOrEqualTo(0))); assertThat(partition % 2, is(0)); }
@Test public void partitionNotAvailable() { cluster = new Cluster("clusterId", Collections.singleton(node), notAvailablePartitions, Collections.emptySet(), Collections.emptySet()); int partition = partitioner.partition(topic, key, keyBytes, value, valueBytes, cluster); assertThat(partition, is(lessThan(allPartitions.size()))); assertThat(partition, is(greaterThanOrEqualTo(0))); assertThat(partition % 2, is(1)); }
@Test public void partitionSameTimeWindow() { int messages = allPartitions.size(); Set<Integer> partitions = new HashSet<>(); for (int i = 0; i < messages; ++i) { int partition = partitioner.partition(topic, key, keyBytes, value, valueBytes, cluster); assertThat(partition, is(greaterThanOrEqualTo(0))); assertThat(partition, is(lessThan(allPartitions.size()))); partitions.add(partition); } assertThat(partitions.size(), is(1)); }
@Test public void partitionDifferentTimeWindows() throws InterruptedException { int messages = allPartitions.size(); Set<Integer> partitions = new HashSet<>(); for (int i = 0; i < messages; ++i) { int partition = partitioner.partition(topic, key, keyBytes, value, valueBytes, cluster); assertThat(partition, is(greaterThanOrEqualTo(0))); assertThat(partition, is(lessThan(allPartitions.size()))); partitions.add(partition); Thread.sleep(FairPartitioner.ROTATE_MILLIS); } assertThat(partitions.size(), is(allPartitions.size() - notAvailablePartitions.size())); }