KuduOneToOnePartitioner extends AbstractKuduInputPartitioner { @Override public Map<Integer, List<KuduPartitionScanAssignmentMeta>> assign(List<KuduPartitionScanAssignmentMeta> totalList, PartitioningContext context) { Map<Integer,List<KuduPartitionScanAssignmentMeta>> partitionAssignments = new HashMap<>(); for ( int i = 0; i < totalList.size(); i++) { List<KuduPartitionScanAssignmentMeta> assignmentForThisPartition = new ArrayList<>(); assignmentForThisPartition.add(totalList.get(i)); partitionAssignments.put(i,assignmentForThisPartition); } return partitionAssignments; } KuduOneToOnePartitioner(AbstractKuduInputOperator prototypeOperator); @Override Map<Integer, List<KuduPartitionScanAssignmentMeta>> assign(List<KuduPartitionScanAssignmentMeta> totalList,
      PartitioningContext context);  }
@KuduClusterTestContext(kuduClusterBasedTest = false) @Test public void testAssignPartitions() throws Exception { AbstractKuduInputOperator<UnitTestTablePojo,InputOperatorControlTuple> mockedInputOperator = PowerMockito.mock(AbstractKuduInputOperator.class); when(mockedInputOperator.getNumberOfPartitions()).thenReturn(9); PowerMockito.mockStatic(KryoCloneUtils.class); when(KryoCloneUtils.cloneObject(mockedInputOperator)).thenReturn(mockedInputOperator); KuduOneToOnePartitioner kuduOneToOnePartitioner = new KuduOneToOnePartitioner(mockedInputOperator); buildMockWiring(mockedInputOperator, KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE); kuduOneToOnePartitioner.setPrototypeKuduInputOperator(mockedInputOperator); Map<Integer,List<KuduPartitionScanAssignmentMeta>> assignedPartitions = kuduOneToOnePartitioner.assign( kuduOneToOnePartitioner.getListOfPartitionAssignments( new ArrayList<Partitioner.Partition<AbstractKuduInputOperator>>(), partitioningContext),partitioningContext); assertThat(assignedPartitions.size(), is(12)); for (List<KuduPartitionScanAssignmentMeta> eachOperatorassignment: assignedPartitions.values()) { assertThat(eachOperatorassignment.size(), is(1)); } }