@Test public void canFetchAndCombineDiscontinuousLogs() { Set<PaxosValue> paxosValues1 = PaxosSerializationTestUtils.writeToLogs(acceptorLog, learnerLog, 7, 54); Set<PaxosValue> paxosValues2 = PaxosSerializationTestUtils.writeToLogs(acceptorLog, learnerLog, 98, 127); Set<PaxosValue> paxosValues = ImmutableSet.<PaxosValue>builder().addAll(paxosValues1).addAll(paxosValues2).build(); int lastVerified = -1; List<HistoryQuery> historyQueries = ImmutableList.of(HistoryQuery.of( ImmutableNamespaceAndUseCase.of(CLIENT, USE_CASE), lastVerified)); List<LogsForNamespaceAndUseCase> remoteHistory = HistoryLoaderAndTransformer.getLogsForHistoryQueries(history, historyQueries); when(remote.getPaxosHistory(any(), any())).thenReturn(PaxosHistoryOnRemote.of(remoteHistory)); List<CompletePaxosHistoryForNamespaceAndUseCase> completeHistory = paxosLogHistoryProvider.getHistory(); CompletePaxosHistoryForNamespaceAndUseCase historyForNamespaceAndUseCase = Iterables.getOnlyElement(completeHistory); assertThat(historyForNamespaceAndUseCase.namespace()).isEqualTo(CLIENT); assertSanityWithValuesOfFetchedRecords(historyForNamespaceAndUseCase, CLIENT, USE_CASE, (54 - 7 + 1) + (127 - 98 + 1), paxosValues); }
public List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory() { Map<NamespaceAndUseCase, Long> lastVerifiedSequences = getNamespaceAndUseCaseToLastVerifiedSeqMap(); PaxosHistoryOnSingleNode localPaxosHistory = localHistoryLoader.getLocalPaxosHistory(lastVerifiedSequences); List<HistoryQuery> historyQueries = getHistoryQueryListForRemoteServers(lastVerifiedSequences); List<PaxosHistoryOnRemote> rawHistoryFromAllRemotes = getHistoriesFromRemoteServers(historyQueries); List<ConsolidatedPaxosHistoryOnSingleNode> historyFromAllRemotes = buildHistoryFromRemoteResponses(rawHistoryFromAllRemotes); return consolidateAndGetHistoriesAcrossAllNodes( lastVerifiedSequences, localPaxosHistory, historyFromAllRemotes); }
PaxosLogHistoryProvider { public List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory() { Map<NamespaceAndUseCase, Long> lastVerifiedSequences = getNamespaceAndUseCaseToLastVerifiedSeqMap(); PaxosHistoryOnSingleNode localPaxosHistory = localHistoryLoader.getLocalPaxosHistory(lastVerifiedSequences); List<HistoryQuery> historyQueries = getHistoryQueryListForRemoteServers(lastVerifiedSequences); List<PaxosHistoryOnRemote> rawHistoryFromAllRemotes = getHistoriesFromRemoteServers(historyQueries); List<ConsolidatedPaxosHistoryOnSingleNode> historyFromAllRemotes = buildHistoryFromRemoteResponses(rawHistoryFromAllRemotes); return consolidateAndGetHistoriesAcrossAllNodes( lastVerifiedSequences, localPaxosHistory, historyFromAllRemotes); } }
PaxosLogHistoryProvider { public List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory() { Map<NamespaceAndUseCase, Long> lastVerifiedSequences = getNamespaceAndUseCaseToLastVerifiedSeqMap(); PaxosHistoryOnSingleNode localPaxosHistory = localHistoryLoader.getLocalPaxosHistory(lastVerifiedSequences); List<HistoryQuery> historyQueries = getHistoryQueryListForRemoteServers(lastVerifiedSequences); List<PaxosHistoryOnRemote> rawHistoryFromAllRemotes = getHistoriesFromRemoteServers(historyQueries); List<ConsolidatedPaxosHistoryOnSingleNode> historyFromAllRemotes = buildHistoryFromRemoteResponses(rawHistoryFromAllRemotes); return consolidateAndGetHistoriesAcrossAllNodes( lastVerifiedSequences, localPaxosHistory, historyFromAllRemotes); } PaxosLogHistoryProvider(DataSource dataSource, List<TimeLockPaxosHistoryProvider> remoteHistoryProviders); }
PaxosLogHistoryProvider { public List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory() { Map<NamespaceAndUseCase, Long> lastVerifiedSequences = getNamespaceAndUseCaseToLastVerifiedSeqMap(); PaxosHistoryOnSingleNode localPaxosHistory = localHistoryLoader.getLocalPaxosHistory(lastVerifiedSequences); List<HistoryQuery> historyQueries = getHistoryQueryListForRemoteServers(lastVerifiedSequences); List<PaxosHistoryOnRemote> rawHistoryFromAllRemotes = getHistoriesFromRemoteServers(historyQueries); List<ConsolidatedPaxosHistoryOnSingleNode> historyFromAllRemotes = buildHistoryFromRemoteResponses(rawHistoryFromAllRemotes); return consolidateAndGetHistoriesAcrossAllNodes( lastVerifiedSequences, localPaxosHistory, historyFromAllRemotes); } PaxosLogHistoryProvider(DataSource dataSource, List<TimeLockPaxosHistoryProvider> remoteHistoryProviders); List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory(); }
PaxosLogHistoryProvider { public List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory() { Map<NamespaceAndUseCase, Long> lastVerifiedSequences = getNamespaceAndUseCaseToLastVerifiedSeqMap(); PaxosHistoryOnSingleNode localPaxosHistory = localHistoryLoader.getLocalPaxosHistory(lastVerifiedSequences); List<HistoryQuery> historyQueries = getHistoryQueryListForRemoteServers(lastVerifiedSequences); List<PaxosHistoryOnRemote> rawHistoryFromAllRemotes = getHistoriesFromRemoteServers(historyQueries); List<ConsolidatedPaxosHistoryOnSingleNode> historyFromAllRemotes = buildHistoryFromRemoteResponses(rawHistoryFromAllRemotes); return consolidateAndGetHistoriesAcrossAllNodes( lastVerifiedSequences, localPaxosHistory, historyFromAllRemotes); } PaxosLogHistoryProvider(DataSource dataSource, List<TimeLockPaxosHistoryProvider> remoteHistoryProviders); List<CompletePaxosHistoryForNamespaceAndUseCase> getHistory();  }