@Test public void shouldBeUnhealthyIfQuorumNodesAreNotUpAndOnePingedSuccessfully() { HealthCheckPinger leader1 = getMockOfPingableLeaderWherePingReturns(ALL_CLIENTS); HealthCheckPinger leader2 = getMockOfPingableLeaderWherePingThrows(); HealthCheckPinger leader3 = getMockOfPingableLeaderWherePingThrows(); List<HealthCheckPinger> leaders = ImmutableList.of(leader1, leader2, leader3); SetMultimap<TimeLockStatus, Client> expected = ImmutableSetMultimap.<TimeLockStatus, Client>builder() .putAll(TimeLockStatus.NO_QUORUM, ALL_CLIENTS) .build(); assertThat(new LeaderPingHealthCheck(TRACKER, leaders).getStatus().statusesToClient()) .isEqualTo(expected); }
public HealthCheckDigest getStatus() { Set<Client> namespacesToCheck = namespaceTracker.trackedNamespaces(); PaxosResponses<PaxosContainer<Map<Client, HealthCheckResponse>>> responses = PaxosQuorumChecker.collectAsManyResponsesAsPossible( ImmutableList.copyOf(pingers), pinger -> PaxosContainer.of(pinger.apply(namespacesToCheck)), executorService, HEALTH_CHECK_TIME_LIMIT, PaxosConstants.CANCEL_REMAINING_CALLS); Map<Client, PaxosResponses<HealthCheckResponse>> responsesByClient = responses.stream() .map(PaxosContainer::get) .map(Map::entrySet) .flatMap(Collection::stream) .collect(Collectors.groupingBy( Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.collectingAndThen(Collectors.toList(), r -> PaxosResponses.of(getQuorumSize(), r))))); SetMultimap<TimeLockStatus, Client> statusesToClient = KeyedStream.stream(responsesByClient) .map(this::convertQuorumResponsesToStatus) .mapEntries((client, status) -> Maps.immutableEntry(status, client)) .collectToSetMultimap(); return ImmutableHealthCheckDigest.of(statusesToClient); }
LeaderPingHealthCheck { public HealthCheckDigest getStatus() { Set<Client> namespacesToCheck = namespaceTracker.trackedNamespaces(); PaxosResponses<PaxosContainer<Map<Client, HealthCheckResponse>>> responses = PaxosQuorumChecker.collectAsManyResponsesAsPossible( ImmutableList.copyOf(pingers), pinger -> PaxosContainer.of(pinger.apply(namespacesToCheck)), executorService, HEALTH_CHECK_TIME_LIMIT, PaxosConstants.CANCEL_REMAINING_CALLS); Map<Client, PaxosResponses<HealthCheckResponse>> responsesByClient = responses.stream() .map(PaxosContainer::get) .map(Map::entrySet) .flatMap(Collection::stream) .collect(Collectors.groupingBy( Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.collectingAndThen(Collectors.toList(), r -> PaxosResponses.of(getQuorumSize(), r))))); SetMultimap<TimeLockStatus, Client> statusesToClient = KeyedStream.stream(responsesByClient) .map(this::convertQuorumResponsesToStatus) .mapEntries((client, status) -> Maps.immutableEntry(status, client)) .collectToSetMultimap(); return ImmutableHealthCheckDigest.of(statusesToClient); } }
LeaderPingHealthCheck { public HealthCheckDigest getStatus() { Set<Client> namespacesToCheck = namespaceTracker.trackedNamespaces(); PaxosResponses<PaxosContainer<Map<Client, HealthCheckResponse>>> responses = PaxosQuorumChecker.collectAsManyResponsesAsPossible( ImmutableList.copyOf(pingers), pinger -> PaxosContainer.of(pinger.apply(namespacesToCheck)), executorService, HEALTH_CHECK_TIME_LIMIT, PaxosConstants.CANCEL_REMAINING_CALLS); Map<Client, PaxosResponses<HealthCheckResponse>> responsesByClient = responses.stream() .map(PaxosContainer::get) .map(Map::entrySet) .flatMap(Collection::stream) .collect(Collectors.groupingBy( Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.collectingAndThen(Collectors.toList(), r -> PaxosResponses.of(getQuorumSize(), r))))); SetMultimap<TimeLockStatus, Client> statusesToClient = KeyedStream.stream(responsesByClient) .map(this::convertQuorumResponsesToStatus) .mapEntries((client, status) -> Maps.immutableEntry(status, client)) .collectToSetMultimap(); return ImmutableHealthCheckDigest.of(statusesToClient); } LeaderPingHealthCheck(
            NamespaceTracker namespaceTracker,
            List<HealthCheckPinger> pingers); }
LeaderPingHealthCheck { public HealthCheckDigest getStatus() { Set<Client> namespacesToCheck = namespaceTracker.trackedNamespaces(); PaxosResponses<PaxosContainer<Map<Client, HealthCheckResponse>>> responses = PaxosQuorumChecker.collectAsManyResponsesAsPossible( ImmutableList.copyOf(pingers), pinger -> PaxosContainer.of(pinger.apply(namespacesToCheck)), executorService, HEALTH_CHECK_TIME_LIMIT, PaxosConstants.CANCEL_REMAINING_CALLS); Map<Client, PaxosResponses<HealthCheckResponse>> responsesByClient = responses.stream() .map(PaxosContainer::get) .map(Map::entrySet) .flatMap(Collection::stream) .collect(Collectors.groupingBy( Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.collectingAndThen(Collectors.toList(), r -> PaxosResponses.of(getQuorumSize(), r))))); SetMultimap<TimeLockStatus, Client> statusesToClient = KeyedStream.stream(responsesByClient) .map(this::convertQuorumResponsesToStatus) .mapEntries((client, status) -> Maps.immutableEntry(status, client)) .collectToSetMultimap(); return ImmutableHealthCheckDigest.of(statusesToClient); } LeaderPingHealthCheck(
            NamespaceTracker namespaceTracker,
            List<HealthCheckPinger> pingers); HealthCheckDigest getStatus(); }
LeaderPingHealthCheck { public HealthCheckDigest getStatus() { Set<Client> namespacesToCheck = namespaceTracker.trackedNamespaces(); PaxosResponses<PaxosContainer<Map<Client, HealthCheckResponse>>> responses = PaxosQuorumChecker.collectAsManyResponsesAsPossible( ImmutableList.copyOf(pingers), pinger -> PaxosContainer.of(pinger.apply(namespacesToCheck)), executorService, HEALTH_CHECK_TIME_LIMIT, PaxosConstants.CANCEL_REMAINING_CALLS); Map<Client, PaxosResponses<HealthCheckResponse>> responsesByClient = responses.stream() .map(PaxosContainer::get) .map(Map::entrySet) .flatMap(Collection::stream) .collect(Collectors.groupingBy( Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.collectingAndThen(Collectors.toList(), r -> PaxosResponses.of(getQuorumSize(), r))))); SetMultimap<TimeLockStatus, Client> statusesToClient = KeyedStream.stream(responsesByClient) .map(this::convertQuorumResponsesToStatus) .mapEntries((client, status) -> Maps.immutableEntry(status, client)) .collectToSetMultimap(); return ImmutableHealthCheckDigest.of(statusesToClient); } LeaderPingHealthCheck(
            NamespaceTracker namespaceTracker,
            List<HealthCheckPinger> pingers); HealthCheckDigest getStatus();  }