class 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();
    }

}