SSPAllocator { synchronized SystemStreamPartition allocate(String stream) throws NoAvailablePartitionException { Validate.notNull(availableSsps.get(stream), String.format("availableSsps is null for stream %s", stream)); if (availableSsps.get(stream).isEmpty()) { throw new NoAvailablePartitionException(String.format("More shards detected for stream %s than initially" + " registered. Could be the result of dynamic resharding.", stream)); } SystemStreamPartition ssp = availableSsps.get(stream).iterator().next(); availableSsps.get(stream).remove(ssp); LOG.info("Number of unassigned partitions for system-stream {} is {}.", ssp.getSystemStream(), availableSsps.get(ssp.getStream()).size()); return ssp; }    }
@Test (expected = NoAvailablePartitionException.class) public void testAssignMoreThanMaxPartitions() throws NoAvailablePartitionException { int numPartitions = 2; String system = "kinesis"; String stream = "stream"; List<SystemStreamPartition> ssps = new ArrayList<>(); IntStream.range(0, numPartitions) .forEach(i -> ssps.add(new SystemStreamPartition(system, stream, new Partition(i)))); SSPAllocator allocator = new SSPAllocator(); ssps.forEach(allocator::free); allocator.allocate(stream); allocator.allocate(stream); allocator.allocate(stream); }