PartitionLeaderAwarenessHealthIndicator implements HealthIndicator { @Override public Health health() { final var optClusterState = clusterStateSupplier.get(); if (optClusterState.isEmpty()) { return Health.down().build(); } else { final var clusterState = optClusterState.get(); if (clusterState.getPartitions().stream() .filter( index -> { return clusterState.getLeaderForPartition(index) != BrokerClusterState.NODE_ID_NULL; }) .findAny() .isPresent()) { return Health.up().build(); } else { return Health.down().build(); } } } PartitionLeaderAwarenessHealthIndicator(
      final Supplier<Optional<BrokerClusterState>> clusterStateSupplier); @Override Health health();  }
@Test public void shouldReportDownIfSupplierReturnsEmpty() { final Supplier<Optional<BrokerClusterState>> stateSupplier = () -> Optional.empty(); final var sutHealthIndicator = new PartitionLeaderAwarenessHealthIndicator(stateSupplier); final var actualHealth = sutHealthIndicator.health(); assertThat(actualHealth).isNotNull(); assertThat(actualHealth.getStatus()).isEqualTo(Status.DOWN); }
@Test public void shouldReportUpIfAnyPartitionLeaderIsKnown() { final BrokerClusterState mockClusterState = mock(BrokerClusterState.class); when(mockClusterState.getPartitions()).thenReturn(List.of(1, 2)); when(mockClusterState.getLeaderForPartition(1)).thenReturn(BrokerClusterState.NODE_ID_NULL); when(mockClusterState.getLeaderForPartition(2)).thenReturn(42); final Supplier<Optional<BrokerClusterState>> stateSupplier = () -> Optional.of(mockClusterState); final var sutHealthIndicator = new PartitionLeaderAwarenessHealthIndicator(stateSupplier); final var actualHealth = sutHealthIndicator.health(); assertThat(actualHealth).isNotNull(); assertThat(actualHealth.getStatus()).isEqualTo(Status.UP); }
@Test public void shouldReportDownIfListOfPartitionsIsEmpty() { final BrokerClusterState mockClusterState = mock(BrokerClusterState.class); when(mockClusterState.getPartitions()).thenReturn(emptyList()); final Supplier<Optional<BrokerClusterState>> stateSupplier = () -> Optional.of(mockClusterState); final var sutHealthIndicator = new PartitionLeaderAwarenessHealthIndicator(stateSupplier); final var actualHealth = sutHealthIndicator.health(); assertThat(actualHealth).isNotNull(); assertThat(actualHealth.getStatus()).isEqualTo(Status.DOWN); }
@Test public void shouldReportDownIfNoPartitionLeaderIsKnown() { final BrokerClusterState mockClusterState = mock(BrokerClusterState.class); when(mockClusterState.getPartitions()).thenReturn(List.of(1, 2)); when(mockClusterState.getLeaderForPartition(1)).thenReturn(BrokerClusterState.NODE_ID_NULL); when(mockClusterState.getLeaderForPartition(2)).thenReturn(BrokerClusterState.NODE_ID_NULL); final Supplier<Optional<BrokerClusterState>> stateSupplier = () -> Optional.of(mockClusterState); final var sutHealthIndicator = new PartitionLeaderAwarenessHealthIndicator(stateSupplier); final var actualHealth = sutHealthIndicator.health(); assertThat(actualHealth).isNotNull(); assertThat(actualHealth.getStatus()).isEqualTo(Status.DOWN); }