[camel] branch camel-4.0.x updated: CAMEL-18759: read the messages from the latest offset in the pausable consumer (#11471)

[orpiske]

This is an automated email from the ASF dual-hosted git repository.

orpiske pushed a commit to branch camel-4.0.x
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/camel-4.0.x by this push:
     new 5b1af66fd53 CAMEL-18759: read the messages from the latest offset in 
the pausable consumer (#11471)
5b1af66fd53 is described below

commit 5b1af66fd53377dd65a2d15338581581db68e045
Author: Otavio Rodolfo Piske <orpi...@users.noreply.github.com>
AuthorDate: Tue Sep 19 19:34:12 2023 +0200

    CAMEL-18759: read the messages from the latest offset in the pausable 
consumer (#11471)
---
 .../java/org/apache/camel/component/kafka/KafkaFetchRecords.java   | 7 +++++--
 1 file changed, 5 insertions(+), 2 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 2511137ecb9..e73ccfc8a76 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
@@ -209,9 +209,12 @@ public class KafkaFetchRecords implements Runnable {
             if (consumerListener != null) {
                 consumerListener.setConsumer(consumer);
 
-                SeekPolicy seekPolicy = 
kafkaConsumer.getEndpoint().getComponent().getConfiguration().getSeekTo();
+                SeekPolicy seekPolicy = 
kafkaConsumer.getEndpoint().getConfiguration().getSeekTo();
                 if (seekPolicy == null) {
-                    seekPolicy = SeekPolicy.BEGINNING;
+                    seekPolicy = 
kafkaConsumer.getEndpoint().getComponent().getConfiguration().getSeekTo();
+                    if (seekPolicy == null) {
+                        seekPolicy = SeekPolicy.BEGINNING;
+                    }
                 }
 
                 consumerListener.setSeekPolicy(seekPolicy);