@KuduClusterTestContext(kuduClusterBasedTest = true)
@Test
public void testDefinePartitions() throws Exception
{

  // case of setting num partitions to 7 but setting one to one mapping.
  numberOfKuduInputOperatorPartitions = 7;
  initOperatorState();
  AbstractKuduInputPartitioner partitioner = unitTestStepwiseScanInputOperator.getPartitioner();
  assertEquals(KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE,partitioner.definePartitions(
      new ArrayList<Partitioner.Partition<AbstractKuduInputOperator>>(),partitioningContext).size());
  // case of setting num partition to 7 but go with many to one mapping
  partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR;
  initOperatorState();
  partitioner = unitTestStepwiseScanInputOperator.getPartitioner();
  Collection<Partitioner.Partition<AbstractKuduInputOperator>> partitionsCalculated = partitioner.definePartitions(
      new ArrayList<Partitioner.Partition<AbstractKuduInputOperator>>(),partitioningContext);
  assertEquals(7,partitionsCalculated.size());
  int maxPartitionPerOperator = -1;
  int minPartitionPerOperator = Integer.MAX_VALUE;
  Iterator<Partitioner.Partition<AbstractKuduInputOperator>> iteratorForPartitionScan =
      partitionsCalculated.iterator();
  while ( iteratorForPartitionScan.hasNext()) {
    Partitioner.Partition<AbstractKuduInputOperator> anOperatorPartition = iteratorForPartitionScan.next();
    AbstractKuduInputOperator<UnitTestTablePojo,InputOperatorControlTuple> anOperatorHandle =
        anOperatorPartition.getPartitionedInstance();
    List<KuduPartitionScanAssignmentMeta> partitionsAssigned = anOperatorHandle.getPartitionPieAssignment();
    if (  partitionsAssigned.size() > maxPartitionPerOperator ) {
      maxPartitionPerOperator = partitionsAssigned.size();
    }
    if ( partitionsAssigned.size() < minPartitionPerOperator) {
      minPartitionPerOperator = partitionsAssigned.size();
    }
  }
  assertEquals(2,maxPartitionPerOperator); // 7 kudu operator instances dealing with 12 kudu tablets
  assertEquals(1,minPartitionPerOperator); // 7 kudu operator instances dealing with 12 kudu tablets
  // revert all the changes to defaults for the next test
  numberOfKuduInputOperatorPartitions = 5;
  partitonScanStrategy = KuduPartitionScanStrategy.ONE_TABLET_PER_OPERATOR;
}