@Override
    public Set<TopicPartition> getPartitionsForThisTask(List<TopicPartition> allPartitionsSorted, TopologyContext context) {
        int thisTaskIndex = context.getThisTaskIndex();
        int totalTaskCount = context.getComponentTasks(context.getThisComponentId()).size();
        Set<TopicPartition> myPartitions = new HashSet<>(allPartitionsSorted.size() / totalTaskCount + 1);
        for (int i = thisTaskIndex; i < allPartitionsSorted.size(); i += totalTaskCount) {
            myPartitions.add(allPartitionsSorted.get(i));
        }
        return myPartitions;
    }
@Test
    public void testRoundRobinPartitioning() {
        List<TopicPartition> allPartitions = new ArrayList<>();
        for(int i = 0; i < 11; i++) {
            allPartitions.add(createTp(i));
        }
        List<TopologyContext> contextMocks = new ArrayList<>();
        String thisComponentId = "A spout";
        List<Integer> allTasks = Arrays.asList(0, 1, 2);
        for(int i = 0; i < 3; i++) {
            TopologyContext contextMock = mock(TopologyContext.class);
            when(contextMock.getThisTaskIndex()).thenReturn(i);
            when(contextMock.getThisComponentId()).thenReturn(thisComponentId);
            when(contextMock.getComponentTasks(thisComponentId)).thenReturn(allTasks);
            contextMocks.add(contextMock);
        }
        RoundRobinManualPartitioner partitioner = new RoundRobinManualPartitioner();
        
        Set<TopicPartition> partitionsForFirstTask = partitioner.getPartitionsForThisTask(allPartitions, contextMocks.get(0));
        assertThat(partitionsForFirstTask, is(partitionsToTps(new int[]{0, 3, 6, 9})));
        
        Set<TopicPartition> partitionsForSecondTask = partitioner.getPartitionsForThisTask(allPartitions, contextMocks.get(1));
        assertThat(partitionsForSecondTask, is(partitionsToTps(new int[]{1, 4, 7, 10})));
        
        Set<TopicPartition> partitionsForThirdTask = partitioner.getPartitionsForThisTask(allPartitions, contextMocks.get(2));
        assertThat(partitionsForThirdTask, is(partitionsToTps(new int[]{2, 5, 8})));
    }