SourceInfo extends AbstractSourceInfo { public Map<String, ?> offset() { return offsetUsingPosition(this.restartRowsToSkip); } SourceInfo(MySqlConnectorConfig connectorConfig); void setQuery(final String query); String getQuery(); Map<String, String> partition(); void setBinlogStartPoint(String binlogFilename, long positionOfFirstEvent); void setEventPosition(long positionOfCurrentEvent, long eventSizeInBytes); Map<String, ?> offset(); Map<String, Object> offsetForRow(int eventRowNumber, int totalNumberOfRows); void databaseEvent(String databaseName); void tableEvent(Set<TableId> tableIds); void tableEvent(TableId tableId); boolean isSnapshotInEffect(); void startNextTransaction(); void completeEvent(); long eventsToSkipUponRestart(); void commitTransaction(); void startGtid(String gtid, String gtidSet); void setCompletedGtidSet(String gtidSet); void setBinlogServerId(long serverId); void setBinlogTimestampSeconds(long timestampInSeconds); void setBinlogThread(long threadId); void startSnapshot(); void markLastSnapshot(Configuration config); void completeSnapshot(); void setFilterDataFromConfig(Configuration config); void maybeSetFilterDataFromConfig(Configuration config); boolean hasFilterInfo(); String getDatabaseIncludeList(); String getDatabaseExcludeList(); String getTableIncludeList(); String getTableExcludeList(); void setOffset(Map<String, ?> sourceOffset); static boolean offsetsHaveFilterInfo(Map<String, ?> sourceOffset); String gtidSet(); String binlogFilename(); long binlogPosition(); int rowsToSkipUponRestart(); @Override String toString(); static Document createDocumentFromOffset(Map<String, ?> offset); static boolean isPositionAtOrBefore(Document recorded, Document desired, Predicate<String> gtidFilter); static final String SERVER_ID_KEY; static final String SERVER_PARTITION_KEY; static final String GTID_SET_KEY; static final String GTID_KEY; static final String EVENTS_TO_SKIP_OFFSET_KEY; static final String BINLOG_FILENAME_OFFSET_KEY; static final String BINLOG_POSITION_OFFSET_KEY; static final String BINLOG_ROW_IN_EVENT_OFFSET_KEY; static final String TIMESTAMP_KEY; static final String THREAD_KEY; static final String QUERY_KEY; static final String DATABASE_WHITELIST_KEY; static final String DATABASE_INCLUDE_LIST_KEY; static final String DATABASE_BLACKLIST_KEY; static final String DATABASE_EXCLUDE_LIST_KEY; static final String TABLE_WHITELIST_KEY; static final String TABLE_INCLUDE_LIST_KEY; static final String TABLE_BLACKLIST_KEY; static final String TABLE_EXCLUDE_LIST_KEY; static final String RESTART_PREFIX; }
@Test public void shouldAdvanceSourceInfoFromNonZeroPositionAndRowZeroForEventsWithOneRow() { sourceWith(offset(100, 0)); handleTransactionBegin(150, 2); handleNextEvent(200, 10, withRowCount(1)); handleTransactionCommit(210, 2); handleTransactionBegin(210, 2); handleNextEvent(220, 10, withRowCount(1)); handleTransactionCommit(230, 3); handleTransactionBegin(240, 2); handleNextEvent(250, 50, withRowCount(1)); handleTransactionCommit(300, 4); handleTransactionBegin(340, 2); handleNextEvent(350, 20, withRowCount(1)); handleNextEvent(370, 30, withRowCount(1)); handleNextEvent(400, 40, withRowCount(1)); handleTransactionCommit(440, 4); handleTransactionBegin(500, 2); handleNextEvent(510, 20, withRowCount(1)); handleNextEvent(540, 15, withRowCount(1)); handleNextEvent(560, 10, withRowCount(1)); handleTransactionCommit(580, 4); handleTransactionBegin(600, 2); handleNextEvent(610, 50, withRowCount(1)); handleTransactionCommit(660, 4); handleNextEvent(670, 10, withRowCount(1)); handleTransactionBegin(700, 2); handleNextEvent(710, 50, withRowCount(1)); handleTransactionCommit(760, 4); }
@Test public void shouldAdvanceSourceInfoFromNonZeroPositionAndRowZeroForEventsWithMultipleRow() { sourceWith(offset(100, 0)); handleTransactionBegin(150, 2); handleNextEvent(200, 10, withRowCount(3)); handleTransactionCommit(210, 2); handleTransactionBegin(210, 2); handleNextEvent(220, 10, withRowCount(4)); handleTransactionCommit(230, 3); handleTransactionBegin(240, 2); handleNextEvent(250, 50, withRowCount(5)); handleTransactionCommit(300, 4); handleTransactionBegin(340, 2); handleNextEvent(350, 20, withRowCount(6)); handleNextEvent(370, 30, withRowCount(1)); handleNextEvent(400, 40, withRowCount(3)); handleTransactionCommit(440, 4); handleTransactionBegin(500, 2); handleNextEvent(510, 20, withRowCount(8)); handleNextEvent(540, 15, withRowCount(9)); handleNextEvent(560, 10, withRowCount(1)); handleTransactionCommit(580, 4); handleTransactionBegin(600, 2); handleNextEvent(610, 50, withRowCount(1)); handleTransactionCommit(660, 4); handleNextEvent(670, 10, withRowCount(5)); handleTransactionBegin(700, 2); handleNextEvent(710, 50, withRowCount(3)); handleTransactionCommit(760, 4); }