AbstractKuduInputPartitioner implements Partitioner<AbstractKuduInputOperator> { public int getNumberOfPartitions(PartitioningContext context) { int proposedPartitionCount = context.getParallelPartitionCount(); if ( prototypeKuduInputOperator.getNumberOfPartitions() != -1 ) { proposedPartitionCount = prototypeKuduInputOperator.getNumberOfPartitions(); LOG.info(" Set the partition count based on the code as opposed to configuration "); } if ( proposedPartitionCount <= 0) { LOG.info(" Defaulting to one partition as parallel partitioning is not enabled"); proposedPartitionCount = 1; } LOG.info(" Planning to use " + proposedPartitionCount + " partitions"); return proposedPartitionCount; } AbstractKuduInputPartitioner(AbstractKuduInputOperator prototypeOperator); List<KuduScanToken> getKuduScanTokensForSelectAllColumns(); int getNumberOfPartitions(PartitioningContext context); List<KuduPartitionScanAssignmentMeta> getListOfPartitionAssignments(
      Collection<Partition<AbstractKuduInputOperator>> partitions, PartitioningContext context); Partitioner.Partition<AbstractKuduInputOperator> clonePartitionAndAssignScanMeta(
      List<KuduPartitionScanAssignmentMeta> scanAssignmentsForThisPartition); @Override Collection<Partition<AbstractKuduInputOperator>> definePartitions(
      Collection<Partition<AbstractKuduInputOperator>> partitions, PartitioningContext context); @Override void partitioned(Map<Integer, Partition<AbstractKuduInputOperator>> partitions); abstract Map<Integer, List<KuduPartitionScanAssignmentMeta>> assign(
      List<KuduPartitionScanAssignmentMeta> totalList, PartitioningContext context); AbstractKuduInputOperator getPrototypeKuduInputOperator(); void setPrototypeKuduInputOperator(AbstractKuduInputOperator prototypeKuduInputOperator);  }
@KuduClusterTestContext(kuduClusterBasedTest = true) @Test public void testNumberOfPartitions() throws Exception { numberOfKuduInputOperatorPartitions = -1; initOperatorState(); AbstractKuduInputPartitioner partitioner = unitTestStepwiseScanInputOperator.getPartitioner(); assertEquals(1,partitioner.getNumberOfPartitions(partitioningContext)); numberOfKuduInputOperatorPartitions = 7; initOperatorState(); partitioner = unitTestStepwiseScanInputOperator.getPartitioner(); assertEquals(7,partitioner.getNumberOfPartitions(partitioningContext)); numberOfKuduInputOperatorPartitions = -1; }