camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject camel git commit: CAMEL-10901 Implemented support for Kafka consumer to seek to end of topic on startup
Date Mon, 27 Feb 2017 15:07:00 GMT
Repository: camel
Updated Branches:
  refs/heads/master 6ed2c587e -> aaf83ceaf


CAMEL-10901 Implemented support for Kafka consumer to seek to end of topic on startup


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/aaf83cea
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/aaf83cea
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/aaf83cea

Branch: refs/heads/master
Commit: aaf83ceaf1bdefb77f5cd509e60cce14ce7cacb1
Parents: 6ed2c58
Author: Sverker Abrahamsson <sverker@limetransit.com>
Authored: Mon Feb 27 01:08:14 2017 +0100
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Feb 27 16:04:31 2017 +0100

----------------------------------------------------------------------
 .../src/main/docs/kafka-component.adoc          |  2 +-
 .../component/kafka/KafkaConfiguration.java     | 18 ++++++-----
 .../camel/component/kafka/KafkaConsumer.java    | 17 +++++++---
 .../component/kafka/KafkaConsumerFullTest.java  | 33 +++++++++++++++++++-
 4 files changed, 56 insertions(+), 14 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/aaf83cea/components/camel-kafka/src/main/docs/kafka-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc b/components/camel-kafka/src/main/docs/kafka-component.adoc
index 99d804b..b6c7887 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -80,7 +80,7 @@ The Kafka component supports 80 endpoint options which are listed below:
 | offsetRepository | consumer |  | String> | The offset repository to use in order to
locally store the offset of each partition of the topic. Defining one will disable the autocommit.
 | partitionAssignor | consumer | org.apache.kafka.clients.consumer.RangeAssignor | String
| The class name of the partition assignment strategy that the client will use to distribute
partition ownership amongst consumer instances when group management is used
 | pollTimeoutMs | consumer | 5000 | Long | The timeout used when polling the KafkaConsumer.
-| seekToBeginning | consumer | false | boolean | If the option is true then KafkaConsumer
will read from beginning on startup.
+| seekTo | consumer |  | String | Set if KafkaConsumer will read from beginning or end on
startup: beginning : read from beginning end : read from end This is replacing the earlier
property seekToBeginning
 | sessionTimeoutMs | consumer | 30000 | Integer | The timeout used to detect failures when
using Kafka's group management facilities.
 | valueDeserializer | consumer | org.apache.kafka.common.serialization.StringDeserializer
| String | Deserializer class for value that implements the Deserializer interface.
 | exceptionHandler | consumer (advanced) |  | ExceptionHandler | To let the consumer use
a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options
is not in use. By default the consumer will deal with exceptions that will be logged at WARN
or ERROR level and ignored.

http://git-wip-us.apache.org/repos/asf/camel/blob/aaf83cea/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
index 57eea8f..8807d29 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
@@ -106,8 +106,8 @@ public class KafkaConfiguration {
     //fetch.max.wait.ms
     @UriParam(label = "consumer", defaultValue = "500")
     private Integer fetchWaitMaxMs = 500;
-    @UriParam(label = "consumer")
-    private boolean seekToBeginning;
+    @UriParam(label = "consumer", enums = "beginning,end")
+    private String seekTo;
 
     //Consumer configuration properties
     @UriParam(label = "consumer")
@@ -1305,15 +1305,19 @@ public class KafkaConfiguration {
         this.valueDeserializer = valueDeserializer;
     }
 
-    public boolean isSeekToBeginning() {
-        return seekToBeginning;
+    public String getSeekTo() {
+        return seekTo;
     }
 
     /**
-     * If the option is true, then KafkaConsumer will read from beginning on startup.
+     * Set if KafkaConsumer will read from beginning or end on startup:
+     * beginning : read from beginning
+     * end : read from end
+     * 
+     * This is replacing the earlier property seekToBeginning
      */
-    public void setSeekToBeginning(boolean seekToBeginning) {
-        this.seekToBeginning = seekToBeginning;
+    public void setSeekTo(String seekTo) {
+        this.seekTo = seekTo;
     }
 
     public ExecutorService getWorkerPool() {

http://git-wip-us.apache.org/repos/asf/camel/blob/aaf83cea/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java
----------------------------------------------------------------------
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 0e38543..711da6b 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
@@ -172,11 +172,18 @@ public class KafkaConsumer extends DefaultConsumer {
                             }
                         }
                     }
-                } else if (endpoint.getConfiguration().isSeekToBeginning()) {
-                    LOG.debug("{} is seeking to the beginning on topic {}", threadId, topicName);
-                    // This poll to ensures we have an assigned partition otherwise seek
won't work
-                    consumer.poll(100);
-                    consumer.seekToBeginning(consumer.assignment());
+                } else if (endpoint.getConfiguration().getSeekTo() != null) {
+                    if (endpoint.getConfiguration().getSeekTo().equals("beginning")) {
+                        LOG.debug("{} is seeking to the beginning on topic {}", threadId,
topicName);
+                        // This poll to ensures we have an assigned partition otherwise seek
won't work
+                        consumer.poll(100);
+                        consumer.seekToBeginning(consumer.assignment());
+                    } else if (endpoint.getConfiguration().getSeekTo().equals("end")) {
+                        LOG.debug("{} is seeking to the end on topic {}", threadId, topicName);
+                        // This poll to ensures we have an assigned partition otherwise seek
won't work
+                        consumer.poll(100);
+                        consumer.seekToEnd(consumer.assignment());
+                    }
                 }
                 while (isRunAllowed() && !isStoppingOrStopped() && !isSuspendingOrSuspended())
{
                     ConsumerRecords<Object, Object> allRecords = consumer.poll(pollTimeoutMs);

http://git-wip-us.apache.org/repos/asf/camel/blob/aaf83cea/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
index 183d041..f90a9c6 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerFullTest.java
@@ -115,12 +115,43 @@ public class KafkaConsumerFullTest extends BaseEmbeddedKafkaTest {
         context.stopRoute("foo");
 
         KafkaEndpoint kafkaEndpoint = (KafkaEndpoint) from;
-        kafkaEndpoint.getConfiguration().setSeekToBeginning(true);
+        kafkaEndpoint.getConfiguration().setSeekTo("beginning");
 
         context.startRoute("foo");
 
         // As wee set seek to beginning we should re-consume all messages
         to.assertIsSatisfied(3000);
     }
+
+    @Test
+    @Ignore("Currently there is a bug in kafka which leads to an uninterruptable thread so
a resub take too long (works manually)")
+    public void kafkaMessageIsConsumedByCamelSeekedToEnd() throws Exception {
+        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<String, String>(TOPIC,
"1", msg);
+            producer.send(data);
+        }
+        to.assertIsSatisfied(3000);
+
+        to.reset();
+
+        to.expectedMessageCount(0);
+
+        //Restart endpoint,
+        context.stopRoute("foo");
+
+        KafkaEndpoint kafkaEndpoint = (KafkaEndpoint) from;
+        kafkaEndpoint.getConfiguration().setSeekTo("end");
+
+        context.startRoute("foo");
+
+        // As wee set seek to end we should not re-consume any messages
+        synchronized (this) {
+            Thread.sleep(1000);
+        }
+        to.assertIsSatisfied(3000);
+    }
 }
 


Mime
View raw message