DataService extends AbstractService implements IDataService { public List<DataGap> findDataGaps() { final long maxDataToSelect = parameterService .getLong(ParameterConstants.ROUTING_LARGEST_GAP_SIZE); List<DataGap> gaps = findDataGapsUnchecked(); boolean lastGapExists = false; long lastGapStartId = 0; for (DataGap dataGap : gaps) { lastGapExists |= dataGap.gapSize() >= maxDataToSelect - 1; lastGapStartId = Math.max(lastGapStartId, dataGap.getEndId()); } if (!lastGapExists) { if (lastGapStartId == 0) { long maxRoutedDataId = findMaxDataEventDataId(); long minDataId = findMinDataId() - 1; lastGapStartId = Math.max(minDataId, maxRoutedDataId); } if (lastGapStartId > -1) { lastGapStartId++; } DataGap gap = new DataGap(lastGapStartId, lastGapStartId + maxDataToSelect); log.info("Inserting missing last data gap: {}", gap); insertDataGap(gap); gaps = findDataGaps(); } return gaps; } DataService(ISymmetricEngine engine, IExtensionService extensionService); boolean insertReloadEvent(TableReloadRequest request, boolean deleteAtClient); int cancelTableReloadRequest(TableReloadRequest request); void insertTableReloadRequest(TableReloadRequest request); void insertTableReloadRequest(ISqlTransaction transaction, TableReloadRequest request); TableReloadRequest getTableReloadRequest(final TableReloadRequestKey key); @Override TableReloadRequest getTableReloadRequest(long loadId); @Override TableReloadRequest getTableReloadRequest(long loadId, String triggerId, String routerId); List<TableReloadRequest> getTableReloadRequestToProcess(final String sourceNodeId); List<TableReloadRequest> getTableReloadRequests(); List<TableReloadStatus> getTableReloadStatus(); List<TableReloadStatus> getActiveTableReloadStatus(); TableReloadStatus getTableReloadStatusByLoadId(long loadId); List<TableReloadStatus> getTableReloadStatusByTarget(String targetNodeId); List<TableReloadRequest> getTableReloadRequestByLoadId(); List<TableReloadRequest> collapseTableReloadRequestsByLoadId(List<TableReloadRequest> requests); void updateTableReloadStatusDataLoaded(ISqlTransaction transaction, long loadId, long batchId, int batchCount); void updateTableReloadStatusDataCounts(ISqlTransaction transaction, long loadId, long startBatchId, long endBatchId, 
            long dataBatchCount, long rowsCount); void updateTableReloadRequestsLoadId(ISqlTransaction transaction, long loadId, TableReloadRequest request); void updateTableReloadStatusTableCount(ISqlTransaction transaction, long loadId, int tableCount); void createTableReloadStatus(ISqlTransaction transaction, long loadId, String sourceNodeId, String targetNodeId); void updateTableReloadStatusSetupCount(ISqlTransaction transaction, long loadId, int setupBatchCount); int updateTableReloadRequestsCancelled(long loadId); long insertReloadEvent(ISqlTransaction transaction, Node targetNode,
            TriggerRouter triggerRouter, TriggerHistory triggerHistory,
            String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy,
            Status status, long estimatedBatchRowCount); long insertReloadEvent(ISqlTransaction transaction, Node targetNode,
            TriggerRouter triggerRouter, TriggerHistory triggerHistory,
            String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy,
            Status status, String channelId, long estimatedBatchRowCount); long insertReloadEventImmediate(ISqlTransaction transaction, Node targetNode,
            TriggerRouter triggerRouter, TriggerHistory triggerHistory,
            String overrideInitialLoadSelect, boolean isLoad, long loadId, String createBy,
            Status status, String channelId, long estimatedBatchRowCount); @Override Map<Integer, ExtractRequest> insertReloadEvents(Node targetNode, boolean reverse, List<TableReloadRequest> reloadRequests, ProcessInfo processInfo, 
            List<TriggerRouter> triggerRouters, Map<Integer, ExtractRequest> extractRequests,
            IReloadGenerator reloadGenerator); List<String> resolveTargetTables(String sql, TriggerRouter triggerRouter, TriggerHistory triggerHistory, Node targetNode); void insertSqlEvent(Node targetNode, String sql, boolean isLoad, long loadId,
            String createBy); void insertSqlEvent(ISqlTransaction transaction, Node targetNode, String sql,
            boolean isLoad, long loadId, String createBy); void insertSqlEvent(ISqlTransaction transaction, TriggerHistory history,
            String channelId, Node targetNode, String sql, boolean isLoad, long loadId,
            String createBy); void insertScriptEvent(String channelId, Node targetNode, String script, boolean isLoad,
            long loadId, String createBy); void insertScriptEvent(ISqlTransaction transaction, String channelId,
            Node targetNode, String script, boolean isLoad, long loadId, String createBy); int countDataInRange(long firstDataId, long secondDataId); @Override void insertCreateEvent(final Node targetNode, TriggerHistory triggerHistory,
            boolean isLoad, long loadId, String createBy); @Override void insertCreateEvent(Node targetNode, TriggerHistory triggerHistory, String createBy); @Override void insertCreateEvent(ISqlTransaction transaction, Node targetNode,
            TriggerHistory triggerHistory, String channelId, boolean isLoad, long loadId, String createBy); long insertData(Data data); long insertData(ISqlTransaction transaction, final Data data); void insertDataEvents(ISqlTransaction transaction, final List<DataEvent> events); void insertDataAndDataEventAndOutgoingBatch(Data data, String channelId,
            List<Node> nodes, boolean isLoad, long loadId, String createBy); long insertDataAndDataEventAndOutgoingBatch(Data data, String nodeId,
            boolean isLoad, long loadId, String createBy); long insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction, Data data, String nodeId, boolean isLoad,
            long loadId, String createBy, Status status, String overrideChannelId, long estimatedBatchRowCount); long insertDataAndDataEventAndOutgoingBatch(ISqlTransaction transaction, Data data,
            String nodeId, boolean isLoad, long loadId, String createBy,
            Status status, long estimatedBatchRowCount); String reloadNode(String nodeId, boolean reverseLoad, String createBy); void sendScript(String nodeId, String script, boolean isLoad); boolean sendSchema(String nodeId, String catalogName, String schemaName,
            String tableName, boolean isLoad); String sendSQL(String nodeId, String catalogName, String schemaName, String tableName,
            String sql); String sendSQL(String nodeId, String sql); @Override String reloadTable(String nodeId, String catalogName, String schemaName, String tableName); @Override String reloadTable(String nodeId, String catalogName, String schemaName,
            String tableName, String overrideInitialLoadSelect); @Override String reloadTableImmediate(String nodeId, String catalogName, String schemaName, String tableName, 
            String overrideInitialLoadSelect, String overrideChannelId); void reloadMissingForeignKeyRowsReverse(String sourceNodeId, Table table, CsvData data, String channelId, boolean sendCorrectionToPeers); void reloadMissingForeignKeyRows(String nodeId, long dataId); void insertHeartbeatEvent(Node node, boolean isReload); Data createData(String catalogName, String schemaName, String tableName); Data createData(String catalogName, String schemaName, String tableName,
            String whereClause); Data createData(ISqlTransaction transaction, String catalogName, String schemaName,
            String tableName, String whereClause); long countDataGaps(); List<DataGap> findDataGapsUnchecked(); List<DataGap> findDataGaps(); long findMaxDataEventDataId(); void insertDataGap(DataGap gap); void insertDataGap(ISqlTransaction transaction, DataGap gap); @Override void insertDataGaps(ISqlTransaction transaction, Collection<DataGap> gaps); @Override void deleteDataGap(DataGap gap); @Override void deleteDataGap(ISqlTransaction transaction, DataGap gap); @Override void deleteDataGaps(ISqlTransaction transaction, Collection<DataGap> gaps); void deleteAllDataGaps(ISqlTransaction transaction); Date findCreateTimeOfEvent(long dataId); Date findCreateTimeOfData(long dataId); Date findNextCreateTimeOfDataStartingAt(long dataId); void heartbeat(boolean force); List<Number> listDataIds(long batchId, String nodeId); List<Data> listData(long batchId, String nodeId, long startDataId, String channelId,
            final int maxRowsToRetrieve); Data findData(long dataId); ISqlRowMapper<Data> getDataMapper(); ISqlReadCursor<Data> selectDataFor(Batch batch); ISqlReadCursor<Data> selectDataFor(Long batchId, String targetNodeId, boolean isContainsBigLob); ISqlReadCursor<Data> selectDataFor(Long batchId, String channelId); @Override long findMaxDataId(); @Override long findMinDataId(); @Override void deleteCapturedConfigChannelData(); @Override Map<String, Date> getLastDataCaptureByChannel(); @Override boolean fixLastDataGap(); @Override String findNodeIdsByNodeGroupId();  }
@Test public void testFindDataGaps2() throws Exception { final List<DataGap> gaps1 = new ArrayList<DataGap>(); gaps1.add(new DataGap(30953884, 80953883)); gaps1.add(new DataGap(30953883, 80953883)); when(sqlTemplate.queryForLong(ArgumentMatchers.anyString())).thenReturn(0L); String sql = ArgumentMatchers.anyString(); @SuppressWarnings("unchecked") ISqlRowMapper<DataGap> anyMapper = (ISqlRowMapper<DataGap>) ArgumentMatchers.any(); when(sqlTemplate.query(sql, anyMapper, (Object[])ArgumentMatchers.any())).thenReturn(gaps1); dataService.findDataGaps(); verifyNoMoreInteractions(sqlTransaction); }