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

  partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR;
  numberOfKuduInputOperatorPartitions = 1;
  initOperatorState();
  AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> currentScanner =
      unitTestStepwiseScanInputOperator.getScanner();
  // truncate and add some data to the unit test table
  truncateTable();
  addTestDataRows(10); // This is per partition and there are 12 partitions
  assertEquals((KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE * 10 ),countNumRowsInTable());
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      "select * from unittests",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  List<KuduPartitionScanAssignmentMeta> scansForThisQuery = currentScanner.preparePlanForScanners(translator);
  // Now scan for exact match of counts
  long totalRowsRead = 0;
  unitTestStepwiseScanInputOperator.getBuffer().clear();
  for (KuduPartitionScanAssignmentMeta aSegmentToScan :  scansForThisQuery) {
    KuduPartitionScannerCallable<UnitTestTablePojo,InputOperatorControlTuple> threadToScan = new
        KuduPartitionScannerCallable<>(unitTestStepwiseScanInputOperator,aSegmentToScan,
        currentScanner.getConnectionPoolForThreads().get(0),
        unitTestStepwiseScanInputOperator.extractSettersForResultObject(translator),translator);
    totalRowsRead += threadToScan.call();
  }
  // 144 = 120 records + 12 * 2 markers
  int expectedCount = ( 10 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE) +
      ( 2 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE);
  assertEquals(expectedCount,unitTestStepwiseScanInputOperator.getBuffer().size());
  // revert all configs to default
  partitonScanStrategy = KuduPartitionScanStrategy.ONE_TABLET_PER_OPERATOR;
  numberOfKuduInputOperatorPartitions = 5;
}