Merge pull request #1131 from smallrye/backport-#1127-to-3.x

[3.x] Backport Improve Kafka source health check to handle the lazy consumption case

Author: cescoffier

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/health/KafkaSourceReadinessHealth.java

@@ -16,6 +16,7 @@
 import io.smallrye.reactive.messaging.health.HealthReport;
 import io.smallrye.reactive.messaging.kafka.KafkaConnectorIncomingConfiguration;
 import io.smallrye.reactive.messaging.kafka.impl.KafkaAdminHelper;
+import io.smallrye.reactive.messaging.kafka.impl.KafkaSource;
 import io.vertx.mutiny.core.Vertx;
 import io.vertx.mutiny.kafka.admin.KafkaAdminClient;
 
@@ -27,20 +28,22 @@ public class KafkaSourceReadinessHealth extends BaseHealth {
     private final String channel;
     private final Set<String> topics;
     private final Metric metric;
+    private final KafkaSource<?, ?> source;
 
-    public KafkaSourceReadinessHealth(Vertx vertx, KafkaConnectorIncomingConfiguration config,
+    public KafkaSourceReadinessHealth(KafkaSource<?, ?> source, Vertx vertx, KafkaConnectorIncomingConfiguration config,
             Map<String, String> kafkaConfiguration, Consumer<?, ?> consumer, Set<String> topics, Pattern pattern) {
         super(config.getChannel());
         this.config = config;
         this.channel = config.getChannel();
         this.topics = topics;
         this.pattern = pattern;
-
+        this.source = source;
         if (config.getHealthReadinessTopicVerification()) {
             // Do not create the client if the readiness health checks are disabled
             Map<String, Object> adminConfiguration = new HashMap<>(kafkaConfiguration);
             this.admin = KafkaAdminHelper.createAdminClient(vertx, adminConfiguration, config.getChannel(), true);
             this.metric = null;
+
         } else {
             this.admin = null;
             Map<MetricName, ? extends Metric> metrics = consumer.metrics();
@@ -61,7 +64,15 @@ public void close() {
 
     protected void metricsBasedHealthCheck(HealthReport.HealthReportBuilder builder) {
         if (metric != null) {
-            builder.add(channel, (double) metric.metricValue() >= 1.0);
+            boolean connected = (double) metric.metricValue() >= 1.0;
+            boolean hasSubscribers = source.hasSubscribers();
+            if (connected) {
+                builder.add(channel, true);
+            } else if (!hasSubscribers) {
+                builder.add(channel, true, "no subscription yet, so no connection to the Kafka broker yet");
+            } else {
+                builder.add(channel, false);
+            }
         } else {
             builder.add(channel, true).build();
         }

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/impl/KafkaSource.java

@@ -53,6 +53,7 @@ public class KafkaSource<K, V> {
     private final boolean isHealthEnabled;
     private final boolean isCloudEventEnabled;
     private final String channel;
+    private volatile boolean subscribed;
     private final KafkaSourceReadinessHealth health;
 
     public KafkaSource(Vertx vertx,
@@ -150,7 +151,7 @@ className, false, getDeserializationHandler(false, deserializationFailureHandler
         failureHandler = createFailureHandler(config, vertx, kafkaConfiguration, kafkaCDIEvents);
         this.consumer = kafkaConsumer;
         if (configuration.getHealthEnabled() && configuration.getHealthReadinessEnabled()) {
-            health = new KafkaSourceReadinessHealth(vertx, configuration, kafkaConfiguration,
+            health = new KafkaSourceReadinessHealth(this, vertx, configuration, kafkaConfiguration,
                     consumer.getDelegate().unwrap(), topics, pattern);
         } else {
             health = null;
@@ -189,13 +190,15 @@ className, false, getDeserializationHandler(false, deserializationFailureHandler
         }
 
         Multi<IncomingKafkaRecord<K, V>> incomingMulti = multi
+                .onCancellation().invoke(() -> subscribed = false)
                 .onSubscribe().call(s -> {
                     this.consumer.exceptionHandler(t -> reportFailure(t, false));
                     if (this.pattern != null) {
                         BiConsumer<UniEmitter<?>, AsyncResult<Void>> completionHandler = (e, ar) -> {
                             if (ar.failed()) {
                                 e.fail(ar.cause());
                             } else {
+                                subscribed = true;
                                 e.complete(null);
                             }
                         };
@@ -206,7 +209,8 @@ className, false, getDeserializationHandler(false, deserializationFailureHandler
                             delegate.subscribe(pattern, ar -> completionHandler.accept(e, ar));
                         });
                     } else {
-                        return this.consumer.subscribe(topics);
+                        return this.consumer.subscribe(topics)
+                                .onItem().invoke(() -> subscribed = true);
                     }
                 })
                 .map(rec -> commitHandler
@@ -450,4 +454,8 @@ public void isReady(HealthReport.HealthReportBuilder builder) {
     public KafkaConsumer<K, V> getConsumer() {
         return this.consumer;
     }
+
+    public boolean hasSubscribers() {
+        return subscribed;
+    }
 }

smallrye-reactive-messaging-kafka/src/test/java/io/smallrye/reactive/messaging/kafka/health/HealthCheckTest.java

@@ -3,19 +3,20 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
+import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
 import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
-import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
-import org.eclipse.microprofile.reactive.messaging.Incoming;
-import org.eclipse.microprofile.reactive.messaging.Message;
-import org.eclipse.microprofile.reactive.messaging.Outgoing;
+import org.eclipse.microprofile.reactive.messaging.*;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 
@@ -87,6 +88,41 @@ private KafkaMapBasedConfig getKafkaSinkConfigForProducingBean() {
         return builder.build();
     }
 
+    private KafkaMapBasedConfig getKafkaSourceConfig(String topic) {
+        KafkaMapBasedConfig.Builder builder = KafkaMapBasedConfig.builder("mp.messaging.incoming.input")
+                .put("value.deserializer", IntegerDeserializer.class.getName())
+                .put("topic", topic)
+                .put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        return builder.build();
+    }
+
+    @Test
+    public void testHealthOfApplicationWithChannel() {
+        KafkaMapBasedConfig config = getKafkaSourceConfig(topic);
+        LazyConsumingBean bean = runApplication(config, LazyConsumingBean.class);
+
+        AtomicInteger expected = new AtomicInteger(0);
+        usage.produceIntegers(10, null,
+                () -> new ProducerRecord<>(topic, "key", expected.getAndIncrement()));
+
+        await().until(this::isReady);
+        await().until(this::isAlive);
+
+        Multi<Integer> channel = bean.getChannel();
+        channel
+                .select().first(10)
+                .collect().asList()
+                .await().atMost(Duration.ofSeconds(10));
+
+        HealthReport liveness = getHealth().getLiveness();
+        HealthReport readiness = getHealth().getReadiness();
+
+        assertThat(liveness.isOk()).isTrue();
+        assertThat(readiness.isOk()).isTrue();
+        assertThat(liveness.getChannels()).hasSize(1);
+        assertThat(readiness.getChannels()).hasSize(1);
+    }
+
     @ApplicationScoped
     public static class ProducingBean {
 
@@ -109,4 +145,16 @@ public Publisher<Integer> source() {
 
     }
 
+    @ApplicationScoped
+    public static class LazyConsumingBean {
+
+        @Inject
+        @Channel("input")
+        Multi<Integer> channel;
+
+        public Multi<Integer> getChannel() {
+            return channel;
+        }
+    }
+
 }