Support Health Check for Kafka Streams extension

[lhauspie]

Description
As @gunnarmorling said in the #4675 issue, kafka streams doesn't expose any health checks or metrics. It could be interesting to add the health check capability (at least readiness) as the #4675 seems to be only about metrics.

Implementation ideas
Just create the @Readiness that will check if the quarkus.kafka-streams.topics are reachable or not. Maybe by reusing a part of the source code of KafkaStreamsTopologyManager#waitForTopicsToBeCreated:

        try (AdminClient adminClient = AdminClient.create(config)) {
            while (true) {
                try {
                    ListTopicsResult topics = adminClient.listTopics();
                    Set<String> topicNames = topics.names().get(10, TimeUnit.SECONDS);

                    if (topicNames.containsAll(topicsToAwait)) {
                        LOGGER.debug("All expected topics created");
                        return;
                    } else {
                        HashSet<String> missing = new HashSet<>(topicsToAwait);
                        missing.removeAll(topicNames);
                        LOGGER.debug("Waiting for topic(s) to be created: " + missing);
                    }

                    Thread.sleep(1_000);
                } catch (ExecutionException | TimeoutException e) {
                    LOGGER.error("Failed to get topic names from broker", e);
                }
            }
        }

Then add the @BuildStep in deployment module that will addHealthCheck by creating the HealthBuildItem refering to the previous @Readiness class.

[lhauspie]

I'll be glad to help on this topic if you're ok with that issue.