You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2024/02/01 20:54:09 UTC
(camel) branch main updated (57d113b01a7 -> 02fa048e5ca)
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
from 57d113b01a7 Camel 20281 s3 refine (#12972)
new acf7e0b9881 CAMEL-20380: added and/or improved logging messages
new 02fa048e5ca CAMEL-20380: added a note about the shutdown timeout
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../camel-kafka/src/main/docs/kafka-component.adoc | 21 +++++++++++++++++++++
.../camel/component/kafka/KafkaFetchRecords.java | 5 ++++-
.../batching/BatchingProcessingITSupport.java | 1 +
3 files changed, 26 insertions(+), 1 deletion(-)
(camel) 02/02: CAMEL-20380: added a note about the shutdown timeout
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 02fa048e5cac9e508b0282f9dbdd4649928cc074
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Thu Feb 1 13:26:13 2024 +0100
CAMEL-20380: added a note about the shutdown timeout
This should prevent unclean shutdown and lost information with batching
---
.../camel-kafka/src/main/docs/kafka-component.adoc | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc b/components/camel-kafka/src/main/docs/kafka-component.adoc
index b91231aa4f4..11478a4df05 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -616,3 +616,24 @@ public void configure() {
}
----
include::spring-boot:partial$starter.adoc[]
+
+==== Dealing with long polling timeouts
+
+In some cases, applications may want the polling process to have a long timeout (see: `pollTimeoutMs`).
+
+To properly do so, first make sure to have a max polling interval that is higher than the polling timeout (see: `maxPollIntervalMs`).
+
+Then, increase the shutdown timeout to ensure that committing, closing and other Kafka operations are not abruptly aborted. For instance:
+
+[source,java]
+----
+public void configure() {
+ // Note that this can be configured in other ways
+ getCamelContext().getShutdownStrategy().setTimeout(10000);
+
+ // route setup ...
+}
+----
+
+
+
(camel) 01/02: CAMEL-20380: added and/or improved logging messages
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit acf7e0b98813837adf263a727345df88ac87dea5
Author: Otavio Rodolfo Piske <an...@gmail.com>
AuthorDate: Thu Feb 1 11:43:30 2024 +0100
CAMEL-20380: added and/or improved logging messages
This should simplify debugging the issue
---
.../java/org/apache/camel/component/kafka/KafkaFetchRecords.java | 5 ++++-
.../kafka/integration/batching/BatchingProcessingITSupport.java | 1 +
2 files changed, 5 insertions(+), 1 deletion(-)
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 dccec83e00c..8de8517ea1e 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
@@ -352,7 +352,8 @@ public class KafkaFetchRecords implements Runnable {
safeUnsubscribe();
} catch (InterruptException e) {
- kafkaConsumer.getExceptionHandler().handleException("Interrupted while consuming " + threadId + " from kafka topic",
+ kafkaConsumer.getExceptionHandler().handleException(
+ "Thread " + threadId + " interrupted while consuming from kafka topic",
e);
commitManager.commit();
@@ -512,8 +513,10 @@ public class KafkaFetchRecords implements Runnable {
}
// As advised in the KAFKA-1894 ticket, calling this wakeup method breaks the infinite loop
+ LOG.trace("Waking up Kafka consumer");
consumer.wakeup();
} catch (InterruptedException e) {
+ LOG.trace("Interrupted while waiting for processing to finish: waking up Kafka consumer");
consumer.wakeup();
Thread.currentThread().interrupt();
} finally {
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/batching/BatchingProcessingITSupport.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/batching/BatchingProcessingITSupport.java
index f50aef3fd99..a54ccb5464f 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/batching/BatchingProcessingITSupport.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/batching/BatchingProcessingITSupport.java
@@ -92,6 +92,7 @@ abstract class BatchingProcessingITSupport extends BaseEmbeddedKafkaTestSupport
// Fourth step: We start again our route, since we have been committing the offsets from the first step,
// we will expect to consume from the latest committed offset (e.g.: from offset 5()
+ LOG.debug("Starting the fourth step");
contextExtension.getContext().getRouteController().startRoute("batching");
setupPostExecutionExpectations();