PartitionMap { public synchronized boolean addPartition(Partition partition, int maxPartitions) { if (maxPartitions <= 0) { return true; } Set<Partition> partitions = partitionMap.get(partition.getStream()); if (null == partitions) { partitions = new HashSet<Partition>(); partitions.add(partition); partitionMap.put(partition.getStream(), partitions); return true; } if (partitions.contains(partition) || partitions.size() < maxPartitions) { partitions.add(partition); return true; } return false; } PartitionMap(); synchronized boolean addPartition(Partition partition, int maxPartitions); synchronized boolean removePartition(Partition partition);  }
@Test(timeout = 20000) public void testAddPartitionNegativeMax() { PartitionMap map = new PartitionMap(); for (int i = 0; i < 20; i++) { assertTrue(map.addPartition(new Partition("test", i), -1)); } }
@Test(timeout = 20000) public void testAddPartitionMultipleTimes() { PartitionMap map = new PartitionMap(); for (int i = 0; i < 20; i++) { assertTrue(map.addPartition(new Partition("test", 0), 3)); } }
@Test(timeout = 20000) public void testAddPartition() { PartitionMap map = new PartitionMap(); for (int i = 0; i < 3; i++) { assertTrue(map.addPartition(new Partition("test", i), 3)); } for (int i = 3; i < 20; i++) { assertFalse(map.addPartition(new Partition("test", i), 3)); } }