### Question:
RoundRobinPartitionRouter implements PartitionRouter { @Override public Partition getPartition(final Message message, final List<Partition> partitions) throws TubeClientException { if (partitions == null || partitions.isEmpty()) { throw new TubeClientException(new StringBuilder(512) .append("No available partition for topic: ") .append(message.getTopic()).toString()); } AtomicInteger currRouterCount = partitionRouterMap.get(message.getTopic()); if (null == currRouterCount) { AtomicInteger newCounter = new AtomicInteger(0); currRouterCount = partitionRouterMap.putIfAbsent(message.getTopic(), newCounter); if (null == currRouterCount) { currRouterCount = newCounter; } } Partition roundPartition = null; int partSize = partitions.size(); for (int i = 0; i < partSize; i++) { roundPartition = partitions.get((currRouterCount.incrementAndGet() & Integer.MAX_VALUE) % partSize); if (roundPartition != null) { if (roundPartition.getDelayTimeStamp() < System.currentTimeMillis()) { return roundPartition; } } } return partitions.get((steppedCounter.incrementAndGet() & Integer.MAX_VALUE) % partSize); }  @Override Partition getPartition(final Message message, final List<Partition> partitions);  }### Answer:
@Test public void testGetPartition() throws TubeClientException { RoundRobinPartitionRouter router = new RoundRobinPartitionRouter(); Message message = new Message("test", new byte[]{1, 2, 3}); List<Partition> partitions = new ArrayList<Partition>(); Partition partition = new Partition(new BrokerInfo("0:127.0.0.1:18080"), "test", 0); partitions.add(partition); Partition actualPartition = router.getPartition(message, partitions); assertEquals(0, actualPartition.getPartitionId()); partition.setDelayTimeStamp(System.currentTimeMillis() + 10000000); actualPartition = router.getPartition(message, partitions); assertEquals(0, actualPartition.getPartitionId()); }
@Test(expected = TubeClientException.class) public void testGetPartitionInvalidInput() throws TubeClientException { RoundRobinPartitionRouter router = new RoundRobinPartitionRouter(); Message message = new Message("test", new byte[]{1, 2, 3}); List<Partition> partitions = new ArrayList<Partition>(); router.getPartition(message, partitions); }