@Test public void canFetchAndCombineHistoriesAcrossNamespaceAndUseCasePairs() { Map<NamespaceAndUseCase, Set<PaxosValue>> expected = writeLogsForRangeOfNamespaceUseCasePairs(); Set<NamespaceAndUseCase> allNamespaceAndUseCases = expected.keySet(); List<HistoryQuery> historyQueries = allNamespaceAndUseCases .stream() .map(namespaceAndUseCase -> HistoryQuery.of(namespaceAndUseCase, -1)) .collect(Collectors.toList()); List<LogsForNamespaceAndUseCase> remoteHistory = HistoryLoaderAndTransformer.getLogsForHistoryQueries(history, historyQueries); when(remote.getPaxosHistory(any(), any())).thenReturn(PaxosHistoryOnRemote.of(remoteHistory)); List<CompletePaxosHistoryForNamespaceAndUseCase> completeHistory = paxosLogHistoryProvider.getHistory(); assertThat(completeHistory.size()).isEqualTo(100); Set<NamespaceAndUseCase> namespaceAndUseCasesWithHistory = completeHistory.stream().map(historyForNamespaceAndUseCase -> { Client client = historyForNamespaceAndUseCase.namespace(); String useCase = historyForNamespaceAndUseCase.useCase(); NamespaceAndUseCase namespaceAndUseCase = ImmutableNamespaceAndUseCase.of(client, useCase); assertSanityWithValuesOfFetchedRecords( historyForNamespaceAndUseCase, client, useCase, getIntegerValueOfClient(historyForNamespaceAndUseCase), expected.get(namespaceAndUseCase)); return namespaceAndUseCase; }).collect(Collectors.toSet()); assertThat(namespaceAndUseCasesWithHistory).isEqualTo(allNamespaceAndUseCases); }
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();  }