### Question:
KafkaStreamsTopicsHealthCheck implements HealthCheck { @Override public HealthCheckResponse call() { HealthCheckResponseBuilder builder = HealthCheckResponse.named("Kafka Streams topics health check").up(); try { Set<String> missingTopics = manager.getMissingTopics(trimmedTopics); List<String> availableTopics = new ArrayList<>(trimmedTopics); availableTopics.removeAll(missingTopics); if (!availableTopics.isEmpty()) { builder.withData("available_topics", String.join(",", availableTopics)); } if (!missingTopics.isEmpty()) { builder.down().withData("missing_topics", String.join(",", missingTopics)); } } catch (InterruptedException e) { LOGGER.error("error when retrieving missing topics", e); builder.down().withData("technical_error", e.getMessage()); } return builder.build(); }  @PostConstruct void init(); @Override HealthCheckResponse call();  }### Answer:
@Test public void shouldBeUpIfNoMissingTopic() throws InterruptedException { Mockito.when(manager.getMissingTopics(anyList())).thenReturn(Collections.emptySet()); HealthCheckResponse response = healthCheck.call(); assertThat(response.getState()).isEqualTo(HealthCheckResponse.State.UP); }
@Test public void shouldBeDownIfMissingTopic() throws InterruptedException { Mockito.when(manager.getMissingTopics(anyList())).thenReturn(Collections.singleton("topic")); HealthCheckResponse response = healthCheck.call(); assertThat(response.getState()).isEqualTo(HealthCheckResponse.State.DOWN); }