This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 6ec13d5d4e66 CAMEL-24473: Fix Kafka async consumer readiness health
check (#25844)
6ec13d5d4e66 is described below
commit 6ec13d5d4e66a7599ab598e3284392b4653869b7
Author: Omar Atie <[email protected]>
AuthorDate: Fri Aug 28 06:28:22 2026 -0700
CAMEL-24473: Fix Kafka async consumer readiness health check (#25844)
* CAMEL-24473: Fix Kafka async consumer readiness health check
Resolve network readiness for AsyncKafkaConsumer (group.protocol=consumer)
via NetworkClientDelegate instead of failing open when ClassicKafkaConsumer
reflection path is unavailable. Fail closed when connectivity cannot be
verified.
Adds KafkaNetworkHealthHelper, unit tests, and group-protocol health check
IT.
Co-authored-by: Cursor Agent <[email protected]>
* CAMEL-24473: Address review feedback on Kafka health check helper
Preserve fail-open behavior for custom Kafka clients and producers.
Add null guards on async reflection chain and real-client unit tests.
Co-authored-by: Cursor Agent <[email protected]>
---------
Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Cursor Agent <[email protected]>
---
.../camel/component/kafka/KafkaFetchRecords.java | 26 +---
.../component/kafka/KafkaNetworkHealthHelper.java | 140 +++++++++++++++++++++
.../camel/component/kafka/KafkaProducer.java | 26 +---
.../kafka/KafkaNetworkHealthHelperTest.java | 88 +++++++++++++
.../KafkaConsumerGroupProtocolHealthCheckIT.java | 138 ++++++++++++++++++++
5 files changed, 368 insertions(+), 50 deletions(-)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
index d371189a8025..40455caddfc0 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java
@@ -56,7 +56,6 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.WakeupException;
@@ -569,30 +568,7 @@ public class KafkaFetchRecords implements Runnable {
if (!connected) {
return false;
}
-
- boolean ready = true;
- try {
- if (consumer instanceof
org.apache.kafka.clients.consumer.KafkaConsumer) {
- // need to use reflection to access the network client which
has API to check if the client has ready
- // connections
- org.apache.kafka.clients.consumer.KafkaConsumer kc =
(org.apache.kafka.clients.consumer.KafkaConsumer) consumer;
- Object client =
ReflectionHelper.getField(kc.getClass().getDeclaredField("delegate"), kc);
- if (client != null) {
- ConsumerNetworkClient nc
- = (ConsumerNetworkClient)
ReflectionHelper.getField(client.getClass().getDeclaredField("client"),
- client);
- LOG.trace(
- "Health-Check calling
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.hasReadyNode");
- ready = nc.hasReadyNodes(System.currentTimeMillis());
- }
- }
- } catch (Exception e) {
- // ignore
- LOG.debug("Cannot check hasReadyNodes on KafkaConsumer client
(ConsumerNetworkClient) due to: "
- + e.getMessage() + ". This exception is ignored.",
- e);
- }
- return ready;
+ return KafkaNetworkHealthHelper.consumerHasReadyNodes(consumer);
}
private Properties getKafkaProps() {
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaNetworkHealthHelper.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaNetworkHealthHelper.java
new file mode 100644
index 000000000000..87f63dc81626
--- /dev/null
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaNetworkHealthHelper.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+import org.apache.camel.util.ReflectionHelper;
+import org.apache.kafka.clients.KafkaClient;
+import org.apache.kafka.clients.NetworkClient;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
+import org.apache.kafka.clients.producer.internals.Sender;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Resolves Kafka client network readiness across classic and async consumer
protocols.
+ * <p/>
+ * Uses reflection against kafka-clients internals (same approach as
CAMEL-20592) because the public consumer/producer
+ * APIs do not expose connectivity state.
+ */
+final class KafkaNetworkHealthHelper {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KafkaNetworkHealthHelper.class);
+
+ private KafkaNetworkHealthHelper() {
+ }
+
+ static boolean consumerHasReadyNodes(Consumer<?, ?> consumer) {
+ if (!(consumer instanceof
org.apache.kafka.clients.consumer.KafkaConsumer<?, ?> kafkaConsumer)) {
+ // Custom consumer implementations keep legacy fail-open behavior
+ return true;
+ }
+ try {
+ Object delegate =
ReflectionHelper.getField(kafkaConsumer.getClass().getDeclaredField("delegate"),
kafkaConsumer);
+ if (delegate == null) {
+ LOG.warn("KafkaConsumer delegate is null; treating consumer
readiness as not ready");
+ return false;
+ }
+ long now = System.currentTimeMillis();
+
+ Boolean classicReady = classicConsumerHasReadyNodes(delegate, now);
+ if (classicReady != null) {
+ return classicReady;
+ }
+
+ Boolean asyncReady = asyncConsumerHasReadyNodes(delegate, now);
+ if (asyncReady != null) {
+ return asyncReady;
+ }
+
+ LOG.warn("Cannot resolve Kafka consumer network client for
readiness check; treating as not ready");
+ return false;
+ } catch (Exception e) {
+ LOG.debug("Cannot check hasReadyNodes on KafkaConsumer due to: {}.
Treating as not ready.", e.getMessage(), e);
+ return false;
+ }
+ }
+
+ static boolean
producerHasReadyNodes(org.apache.kafka.clients.producer.Producer<?, ?>
producer) {
+ if (!(producer instanceof
org.apache.kafka.clients.producer.KafkaProducer<?, ?> kafkaProducer)) {
+ // Custom producer implementations keep legacy fail-open behavior
+ return true;
+ }
+ try {
+ Sender sender
+ = (Sender)
ReflectionHelper.getField(kafkaProducer.getClass().getDeclaredField("sender"),
kafkaProducer);
+ if (sender == null) {
+ return true;
+ }
+ NetworkClient networkClient
+ = (NetworkClient)
ReflectionHelper.getField(sender.getClass().getDeclaredField("client"), sender);
+ if (networkClient == null) {
+ return true;
+ }
+ LOG.trace("Health-Check calling NetworkClient.hasReadyNodes");
+ return networkClient.hasReadyNodes(System.currentTimeMillis());
+ } catch (Exception e) {
+ LOG.debug("Cannot check hasReadyNodes on KafkaProducer due to: {}.
This exception is ignored.", e.getMessage(), e);
+ return true;
+ }
+ }
+
+ private static Boolean classicConsumerHasReadyNodes(Object delegate, long
now) throws Exception {
+ try {
+ Object client =
ReflectionHelper.getField(delegate.getClass().getDeclaredField("client"),
delegate);
+ if (client instanceof ConsumerNetworkClient networkClient) {
+ LOG.trace("Health-Check calling
ConsumerNetworkClient.hasReadyNodes");
+ return networkClient.hasReadyNodes(now);
+ }
+ if (client instanceof KafkaClient kafkaClient) {
+ return kafkaClient.hasReadyNodes(now);
+ }
+ } catch (NoSuchFieldException e) {
+ // not classic layout
+ }
+ return null;
+ }
+
+ private static Boolean asyncConsumerHasReadyNodes(Object delegate, long
now) throws Exception {
+ try {
+ Object applicationEventHandler
+ =
ReflectionHelper.getField(delegate.getClass().getDeclaredField("applicationEventHandler"),
delegate);
+ if (applicationEventHandler == null) {
+ return null;
+ }
+ Object networkThread = ReflectionHelper.getField(
+
applicationEventHandler.getClass().getDeclaredField("networkThread"),
applicationEventHandler);
+ if (networkThread == null) {
+ return false;
+ }
+ Object networkClientDelegate = ReflectionHelper.getField(
+
networkThread.getClass().getDeclaredField("networkClientDelegate"),
networkThread);
+ if (networkClientDelegate == null) {
+ return false;
+ }
+ KafkaClient kafkaClient = (KafkaClient) ReflectionHelper.getField(
+
networkClientDelegate.getClass().getDeclaredField("client"),
networkClientDelegate);
+ if (kafkaClient == null) {
+ return false;
+ }
+ LOG.trace("Health-Check calling KafkaClient.hasReadyNodes on async
consumer delegate");
+ return kafkaClient.hasReadyNodes(now);
+ } catch (NoSuchFieldException e) {
+ return null;
+ }
+ }
+}
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 7c8c7d722e66..7334c4b93756 100755
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -50,12 +50,10 @@ import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ReflectionHelper;
import org.apache.camel.util.URISupport;
import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.clients.producer.internals.Sender;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.slf4j.Logger;
@@ -111,29 +109,7 @@ public class KafkaProducer extends DefaultAsyncProducer
implements RouteIdAware
}
public boolean isReady() {
- boolean ready = true;
- try {
- if (kafkaProducer instanceof
org.apache.kafka.clients.producer.KafkaProducer) {
- // need to use reflection to access the network client which
has API to check if the client has ready
- // connections
- org.apache.kafka.clients.producer.KafkaProducer kp
- = (org.apache.kafka.clients.producer.KafkaProducer)
kafkaProducer;
- Sender sender
- = (Sender) ReflectionHelper
-
.getField(kp.getClass().getDeclaredField("sender"), kp);
- NetworkClient nc
- = (NetworkClient)
ReflectionHelper.getField(sender.getClass().getDeclaredField("client"), sender);
- LOG.trace(
- "Health-Check calling
org.apache.kafka.clients.NetworkClient.hasReadyNode");
- ready = nc.hasReadyNodes(System.currentTimeMillis());
- }
- } catch (Exception e) {
- // ignore
- LOG.debug("Cannot check hasReadyNodes on KafkaProducer client
(NetworkClient) due to "
- + e.getMessage() + ". This exception is ignored.",
- e);
- }
- return ready;
+ return KafkaNetworkHealthHelper.producerHasReadyNodes(kafkaProducer);
}
@SuppressWarnings("rawtypes")
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaNetworkHealthHelperTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaNetworkHealthHelperTest.java
new file mode 100644
index 000000000000..52ba941da1b8
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaNetworkHealthHelperTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka;
+
+import java.util.Properties;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class KafkaNetworkHealthHelperTest {
+
+ @Test
+ void consumerHasReadyNodesShouldFailOpenForCustomConsumer() {
+
assertTrue(KafkaNetworkHealthHelper.consumerHasReadyNodes(mock(org.apache.kafka.clients.consumer.Consumer.class)));
+ }
+
+ @Test
+ void consumerHasReadyNodesShouldFailOpenForNullConsumer() {
+ assertTrue(KafkaNetworkHealthHelper.consumerHasReadyNodes(null));
+ }
+
+ @Test
+ void producerHasReadyNodesShouldFailOpenForCustomProducer() {
+
assertTrue(KafkaNetworkHealthHelper.producerHasReadyNodes(mock(org.apache.kafka.clients.producer.Producer.class)));
+ }
+
+ @Test
+ void producerHasReadyNodesShouldFailOpenForNullProducer() {
+ assertTrue(KafkaNetworkHealthHelper.producerHasReadyNodes(null));
+ }
+
+ @Test
+ void
consumerHasReadyNodesShouldResolveClassicAndAsyncLayoutsWithoutException() {
+ Properties classicProps = consumerProps("classic");
+ Properties asyncProps = consumerProps("consumer");
+
+ try (KafkaConsumer<String, String> classic = new
KafkaConsumer<>(classicProps);
+ KafkaConsumer<String, String> async = new
KafkaConsumer<>(asyncProps)) {
+ assertDoesNotThrow(() ->
KafkaNetworkHealthHelper.consumerHasReadyNodes(classic));
+ assertDoesNotThrow(() ->
KafkaNetworkHealthHelper.consumerHasReadyNodes(async));
+ }
+ }
+
+ @Test
+ void producerHasReadyNodesShouldResolveLayoutWithoutException() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:1");
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(props)) {
+ assertDoesNotThrow(() ->
KafkaNetworkHealthHelper.producerHasReadyNodes(producer));
+ }
+ }
+
+ private static Properties consumerProps(String groupProtocol) {
+ Properties props = new Properties();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:1");
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, "health-check-test");
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);
+ return props;
+ }
+}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerGroupProtocolHealthCheckIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerGroupProtocolHealthCheckIT.java
new file mode 100644
index 000000000000..1ef43a285920
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/health/KafkaConsumerGroupProtocolHealthCheckIT.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.integration.health;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.integration.common.KafkaAdminUtil;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.FeatureMetadata;
+import org.apache.kafka.clients.admin.FinalizedVersionRange;
+import org.apache.kafka.common.Uuid;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Tags;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Readiness health check coverage for {@code group.protocol=consumer}
(KIP-848 / AsyncKafkaConsumer).
+ */
+@Timeout(60)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@DisabledIfSystemProperty(named = "kafka.instance.type", matches =
"local-strimzi-container",
+ disabledReason = "Test infra Kafka runs the Strimzi
containers in a way that conflicts with multiple concurrent images")
+@Tags({ @Tag("health") })
+@EnabledOnOs(value = { OS.LINUX, OS.MAC, OS.FREEBSD, OS.OPENBSD, OS.WINDOWS },
+ architectures = { "amd64", "aarch64", "s390x" },
+ disabledReason = "This test does not run reliably on ppc64le")
+public class KafkaConsumerGroupProtocolHealthCheckIT extends
KafkaHealthCheckTestSupport {
+
+ public static final String TOPIC = "test-health-group-protocol-" +
Uuid.randomUuid();
+
+ @BeforeAll
+ static void checkConsumerProtocolSupport() {
+ try (AdminClient adminClient =
KafkaAdminUtil.createAdminClient(service)) {
+ FeatureMetadata metadata =
adminClient.describeFeatures().featureMetadata().get(10, TimeUnit.SECONDS);
+ Map<String, FinalizedVersionRange> finalizedFeatures =
metadata.finalizedFeatures();
+ FinalizedVersionRange groupVersion =
finalizedFeatures.get("group.version");
+ Assumptions.assumeTrue(
+ groupVersion != null && groupVersion.maxVersionLevel() >=
1,
+ "Broker does not support the consumer group protocol
(KIP-848), requires Kafka 4.0+ with group.version >= 1");
+ } catch (Exception e) {
+ Assumptions.assumeTrue(false,
+ "Could not determine broker feature support: " +
e.getMessage());
+ }
+ }
+
+ @Override
+ public void configureContext(CamelContext context) {
+ // NO-OP
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ String from = "kafka:" + TOPIC + "?brokers=" +
service.getBootstrapServers()
+ +
"&groupId=KafkaConsumerGroupProtocolHealthCheckIT&autoOffsetReset=earliest"
+ +
"&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true"
+ + "&groupProtocol=consumer";
+
+ from(from)
+ .routeId("test-health-group-protocol-it")
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Order(1)
+ @Test
+ @DisplayName("Readiness reports UP with group.protocol=consumer when
broker is healthy")
+ public void testReportReadyWhenReady() {
+ CamelContext context = contextExtension.getContext();
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ Collection<HealthCheck.Result> results =
HealthCheckHelper.invokeReadiness(context);
+ boolean up = results.stream().allMatch(r ->
r.getState().equals(HealthCheck.State.UP));
+ Assertions.assertTrue(up, "readiness check with async consumer
protocol");
+ });
+ }
+
+ @Order(2)
+ @Test
+ @DisplayName("Readiness reports DOWN with group.protocol=consumer when
broker is shut down")
+ public void testReadinessWhenDown() {
+ CamelContext context = contextExtension.getContext();
+ service.shutdown();
+ serviceShutdown = true;
+
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ Collection<HealthCheck.Result> results =
HealthCheckHelper.invokeReadiness(context);
+ Optional<HealthCheck.Result> down
+ = results.stream().filter(r ->
r.getState().equals(HealthCheck.State.DOWN)).findFirst();
+ Assertions.assertTrue(down.isPresent());
+ String msg = down.get().getMessage().get();
+ Assertions.assertTrue(msg.contains("KafkaConsumer is not ready"));
+ Map<String, Object> details = down.get().getDetails();
+ Assertions.assertEquals(TOPIC, details.get("topic"));
+ Assertions.assertEquals("test-health-group-protocol-it",
details.get("route.id"));
+ });
+ }
+}
