BootstrapProcessor { protected boolean computeIsPhaseCompleted(BootstrapEventProcessResult result, Checkpoint ckpt, long numRowsReadFromDb, long maxRowsPerFetch, long windowScn) { assert(numRowsReadFromDb <= maxRowsPerFetch); boolean isPhaseCompleted = false; if (null == result) { assert(numRowsReadFromDb == 0); isPhaseCompleted = true; } else { if ( (numRowsReadFromDb < maxRowsPerFetch) && (DbusClientMode.BOOTSTRAP_SNAPSHOT == ckpt.getConsumptionMode()) ) { isPhaseCompleted = true; } else if ( (numRowsReadFromDb < maxRowsPerFetch) && (DbusClientMode.BOOTSTRAP_CATCHUP == ckpt.getConsumptionMode()) && (ckpt.getBootstrapTargetScn() == windowScn) ) { isPhaseCompleted = true; } LOG.info("Terminating batch with result: " + result); if (result.isError() || result.isClientBufferLimitExceeded()) { isPhaseCompleted = false; } } return isPhaseCompleted; } BootstrapProcessor(BootstrapServerStaticConfig config,
                            DbusEventsStatisticsCollector curStatsCollector); protected  BootstrapProcessor(); DbusKeyFilter getKeyFilter(); void setKeyFilter(DbusKeyFilter keyFilter); boolean streamCatchupRows(Checkpoint currState, BootstrapEventCallback callBack); String getCatchupSQLString(String catchupTab); String getCatchupSQLString(String catchupTab, String source); String getSnapshotSQLString(String snapShotTable); String getSnapshotSQLString(String snapShotTable, String source); boolean streamSnapShotRows(Checkpoint currState, BootstrapEventCallback callBack); void shutdown(); static final String MODULE; static final Logger LOG; final static String EVENT_COLUMNS; final static String PHASE_COMPLETED_HEADER_NAME; final static String PHASE_COMPLETED_HEADER_TRUE; final static String EMPTY_STRING; }
@Test public void testIsPhaseCompletedFlag() throws Exception { BootstrapProcessor bp = new BootstrapProcessor(); long processedRowCount = 1; 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; 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; pc = bp.computeIsPhaseCompleted(result_ok, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertTrue(pc); pc = bp.computeIsPhaseCompleted(result_err, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); pc = bp.computeIsPhaseCompleted(result_ile, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); numRowsReadFromDb = 2; pc = bp.computeIsPhaseCompleted(result_ok, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); pc = bp.computeIsPhaseCompleted(result_err, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); pc = bp.computeIsPhaseCompleted(result_ile, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); numRowsReadFromDb = 1; 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); pc = bp.computeIsPhaseCompleted(result_err, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); pc = bp.computeIsPhaseCompleted(result_ile, ckpt_ss, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); numRowsReadFromDb = 2; windowScn = 10; pc = bp.computeIsPhaseCompleted(result_ok, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); windowScn = ckpt_cu.getBootstrapTargetScn(); pc = bp.computeIsPhaseCompleted(result_ok, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); pc = bp.computeIsPhaseCompleted(result_err, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); pc = bp.computeIsPhaseCompleted(result_ile, ckpt_cu, numRowsReadFromDb, maxRowsPerFetch, windowScn); Assert.assertFalse(pc); }