You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2024/05/07 12:45:03 UTC
(camel) 01/01: CAMEL-19358: camel-kafka - Add test for suspend/resume kafka consumer. Add JMX operation to expose pause state.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch kafka-pause
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 15f308f17de85cde42eb6fe0dffc87ad3cf710da
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue May 7 14:44:49 2024 +0200
CAMEL-19358: camel-kafka - Add test for suspend/resume kafka consumer. Add JMX operation to expose pause state.
---
.../camel/component/kafka/KafkaConsumer.java | 13 +++++-
.../camel/component/kafka/KafkaFetchRecords.java | 4 +-
.../kafka/integration/KafkaConsumerFullIT.java | 52 ++++++++++++++++++++++
3 files changed, 67 insertions(+), 2 deletions(-)
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
index d2a9d080015..f0c75bd3eb6 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
@@ -27,6 +27,8 @@ import java.util.stream.Collectors;
import org.apache.camel.Processor;
import org.apache.camel.Suspendable;
+import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.component.kafka.consumer.errorhandler.KafkaConsumerListener;
import org.apache.camel.health.HealthCheckAware;
import org.apache.camel.health.HealthCheckHelper;
@@ -46,6 +48,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@ManagedResource(description = "Managed KafkaConsumer")
public class KafkaConsumer extends DefaultConsumer
implements ResumeAware<ResumeStrategy>, HealthCheckAware, ConsumerListenerAware<KafkaConsumerListener>,
Suspendable {
@@ -242,7 +245,15 @@ public class KafkaConsumer extends DefaultConsumer
}
public List<TaskHealthState> healthStates() {
- return tasks.stream().map(t -> t.healthState()).collect(Collectors.toList());
+ return tasks.stream().map(KafkaFetchRecords::healthState).collect(Collectors.toList());
+ }
+
+ /**
+ * Whether the Kafka client is currently paused
+ */
+ @ManagedAttribute(description = "Whether the Kafka client is currently paused")
+ public boolean isKafkaPaused() {
+ return tasks.stream().allMatch(KafkaFetchRecords::isPaused);
}
@Override
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 26ee3cf3330..a40a29ebbd6 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
@@ -549,7 +549,9 @@ public class KafkaFetchRecords implements Runnable {
}
public boolean isPaused() {
- return !consumer.paused().isEmpty();
+ // cannot use consumer directly as you can have ConcurrentModificationException as kafka client does not permit
+ // multiple threads to use the client consumer, so we check the state only
+ return state == State.PAUSED;
}
public void setConnected(boolean connected) {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
index 9398f17a35d..318ecc59e22 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaConsumerFullIT.java
@@ -25,6 +25,7 @@ import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.component.kafka.KafkaConsumer;
import org.apache.camel.component.kafka.KafkaEndpoint;
import org.apache.camel.component.kafka.MockConsumerInterceptor;
import org.apache.camel.component.kafka.SeekPolicy;
@@ -43,6 +44,7 @@ import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.TestMethodOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -223,6 +225,56 @@ public class KafkaConsumerFullIT extends BaseKafkaTestSupport {
assertInstanceOf(MyKafkaHeaderDeserializer.class, kafkaEndpoint.getConfiguration().getHeaderDeserializer());
}
+ @Order(6)
+ @Test
+ public void kafkaMessageIsConsumedByCamelAfterSuspendResume() throws Exception {
+ MockEndpoint to = contextExtension.getMockEndpoint(KafkaTestUtil.MOCK_RESULT);
+
+ to.expectedMessageCount(5);
+ to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
+
+ for (int k = 0; k < 5; k++) {
+ String msg = "message-" + k;
+ ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", msg);
+ producer.send(data);
+ }
+
+ to.assertIsSatisfied(3000);
+
+ assertEquals(5, MockConsumerInterceptor.recordsCaptured.stream()
+ .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
+
+ // suspend route
+ CamelContext context = contextExtension.getContext();
+ context.getRouteController().suspendRoute("full-it");
+
+ // wait until the kafka client is really paused
+ KafkaConsumer kc = (KafkaConsumer) context.getRoute("full-it").getConsumer();
+ Awaitility.await().until(() -> {
+ boolean paused = kc.isKafkaPaused();
+ LOG.info("Waiting for kafka client to be paused: {}", paused);
+ return paused;
+ });
+
+ context.getRouteController().resumeRoute("full-it");
+
+ to.reset();
+
+ to.expectedMessageCount(3);
+ to.expectedBodiesReceivedInAnyOrder("message-5", "message-6", "message-7");
+
+ for (int k = 5; k < 8; k++) {
+ String msg = "message-" + k;
+ ProducerRecord<String, String> data = new ProducerRecord<>(TOPIC, "1", msg);
+ producer.send(data);
+ }
+
+ to.assertIsSatisfied(3000);
+
+ assertEquals(5 + 3, MockConsumerInterceptor.recordsCaptured.stream()
+ .flatMap(i -> StreamSupport.stream(i.records(TOPIC).spliterator(), false)).count());
+ }
+
private static class MyKafkaHeaderDeserializer extends DefaultKafkaHeaderDeserializer {
}
}