BootstrapProcessor { protected void writeCkptIfAppropriate(BootstrapEventProcessResult result, BootstrapEventCallback callback, long numRowsReadFromDb, Checkpoint ckpt, String resultSetStmtStr) throws SQLException { assert (null != callback); assert (numRowsReadFromDb >= 0); assert (null != ckpt); if (null != result && !result.isError()) { assert (numRowsReadFromDb >= result.getNumRowsWritten()); if (result.getNumRowsWritten() > 0) { callback.onCheckpointEvent(ckpt, _curStatsCollector); } else if ((result.getNumRowsWritten() == 0) && (numRowsReadFromDb > 0)) { if (!result.isClientBufferLimitExceeded()) { LOG.info("All the rows read from DB have been filtered out by user-level filter. numRowsReadFromDb = " + numRowsReadFromDb + " sending checkpoint = " + ckpt); callback.onCheckpointEvent(ckpt, _curStatsCollector); } else { LOG.info("There have been rowsReadFromDb that could not be written as the clientBufferLimit has been exceeded. " + "A checkpoint will not be sent, but pendingEvent header will be set. numRowsReadFromDb = " + numRowsReadFromDb + " checkpoint = " + ckpt); } } else { String errMsg = "This is an error-case that should not happen. First, there were no rows in the resultSet " + " Second, this is not a case where all the events have been filtered out. Both of these cannot happen simulateneously. " + " Debug information is logged below. " + " numRowsReadFromDb = " + numRowsReadFromDb + " result = " + result + " resultSet statement " + resultSetStmtStr; LOG.error(errMsg); throw new RuntimeException(errMsg); } } } 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 testCkptWriteLogic() throws Exception { BootstrapProcessor bp = new BootstrapProcessor(); long processedRowCount = 1; BootstrapEventProcessResult result_ok = new BootstrapEventProcessResult(processedRowCount, false, false, false); BootstrapEventProcessResult result_zero_ok = new BootstrapEventProcessResult(0, false, false, false); BootstrapEventProcessResult result_err = new BootstrapEventProcessResult(processedRowCount, false, false, true); BootstrapEventProcessResult result_zero_ile = new BootstrapEventProcessResult(0, 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(); long numRowsReadFromDb = 1; BootstrapEventCallback callback1 = EasyMock.createMock(BootstrapEventCallback.class); EasyMock.replay(callback1); BootstrapEventCallback callback2 = EasyMock.createMock(BootstrapEventCallback.class); callback2.onCheckpointEvent(ckpt_ss, null); EasyMock.replay(callback2); BootstrapEventCallback callback3 = EasyMock.createMock(BootstrapEventCallback.class); callback3.onCheckpointEvent(ckpt_ss, null); EasyMock.replay(callback3); bp.writeCkptIfAppropriate(null, callback1, numRowsReadFromDb, ckpt_ss, "test"); bp.writeCkptIfAppropriate(result_err, callback1, numRowsReadFromDb, ckpt_ss, "test"); bp.writeCkptIfAppropriate(result_ok, callback2, numRowsReadFromDb, ckpt_ss, "test"); bp.writeCkptIfAppropriate(result_zero_ok, callback3, numRowsReadFromDb, ckpt_ss, "test"); bp.writeCkptIfAppropriate(result_zero_ile, callback1, numRowsReadFromDb, ckpt_ss, "test"); try { bp.writeCkptIfAppropriate(result_zero_ok, callback2, 0, ckpt_ss, "test"); Assert.fail(); } catch (RuntimeException e) { } }