@Test
public void testIsPhaseCompletedFlag()
throws Exception
{

  BootstrapProcessor bp = new BootstrapProcessor();
  long processedRowCount = 1;
  // order of args : processedRowCount, isLimitExceeded, isDropped, isError
  BootstrapEventProcessResult result_ok = new BootstrapEventProcessResult(processedRowCount, false, false, false);
  BootstrapEventProcessResult result_err = new BootstrapEventProcessResult(processedRowCount, false, false, true);
  BootstrapEventProcessResult result_ile = new BootstrapEventProcessResult(processedRowCount, true, false, false);
  Checkpoint ckpt_ss = new Checkpoint("{\"consumption_mode\":\"BOOTSTRAP_SNAPSHOT\", \"bootstrap_since_scn\":0," +
      "\"bootstrap_start_scn\":1000,\"bootstrap_target_scn\":2000,\"bootstrap_catchup_source_index\":0," +
      "\"bootstrap_snapshot_source_index\":1}");
  ckpt_ss.assertCheckpoint();
  Checkpoint ckpt_cu = new Checkpoint("{\"consumption_mode\":\"BOOTSTRAP_CATCHUP\", \"bootstrap_since_scn\":0," +
          "\"bootstrap_start_scn\":1000,\"bootstrap_target_scn\":2000,\"bootstrap_catchup_source_index\":1," +
          "\"bootstrap_snapshot_source_index\":1}");
  ckpt_cu.assertCheckpoint();
  long numRowsReadFromDb = 1;
  long maxRowsPerFetch = 2;
  long windowScn = 1;
  // result is null, phaseCompleted == false ( irrespective of snapshot / catchup phase )
  numRowsReadFromDb = 0;
  boolean pc = bp.computeIsPhaseCompleted(null, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertTrue(pc);
  pc = bp.computeIsPhaseCompleted(null, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertTrue(pc);
  numRowsReadFromDb = 1;
  // numRowsReadFromDb < maxRowsPerFetch, in SNAPSHOT mode
  pc = bp.computeIsPhaseCompleted(result_ok, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertTrue(pc);
  // Same as above, but result.isError == true ( overrides all other conditions )
  pc = bp.computeIsPhaseCompleted(result_err, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // Same as above, but result.isLimitExceeded == true ( overrides all other conditions )
  pc = bp.computeIsPhaseCompleted(result_ile, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // numRowsReadFromDb == maxRowsPerFetch, in SNAPSHOT mode
  numRowsReadFromDb = 2;
  pc = bp.computeIsPhaseCompleted(result_ok, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // Same as above, but result.isError == true
  pc = bp.computeIsPhaseCompleted(result_err, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // Same as above, but result.isLimitExceeded == true
  pc = bp.computeIsPhaseCompleted(result_ile, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  numRowsReadFromDb = 1;
  // numRowsReadFromDb < maxRowsPerFetch, in CATCHUP mode, windowScn != bootstrap_target_scn
  pc = bp.computeIsPhaseCompleted(result_ok, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertTrue(pc);
  windowScn = ckpt_cu.getBootstrapTargetScn();
  pc = bp.computeIsPhaseCompleted(result_ok, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertTrue(pc);
  // Same as above, but result.isError == true ( overrides all other conditions )
  pc = bp.computeIsPhaseCompleted(result_err, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // Same as above, but result.isLimitExceeded == true ( overrides result being null )
  pc = bp.computeIsPhaseCompleted(result_ile, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // numRowsReadFromDb == maxRowsPerFetch, in CATCHUP mode
  numRowsReadFromDb = 2;
  windowScn = 10; // not equal to bootstrap_target_scn
  pc = bp.computeIsPhaseCompleted(result_ok, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  windowScn = ckpt_cu.getBootstrapTargetScn(); // equal to bootstrap_target_scn, but does not matter still
  pc = bp.computeIsPhaseCompleted(result_ok, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // Same as above, but result.isError == true
  pc = bp.computeIsPhaseCompleted(result_err, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
  // Same as above, but result.isLimitExceeded == true
  pc = bp.computeIsPhaseCompleted(result_ile, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn);
  Assert.assertFalse(pc);
}