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 2017/02/21 16:09:23 UTC
camel git commit: CAMEL-10871 - When consuming from Kafka indicates
whether or not the record is the last one in the batch
Repository: camel
Updated Branches:
refs/heads/master ac1d6eca0 -> 6d8ba8e89
CAMEL-10871 - When consuming from Kafka indicates whether or not the record is the last one in the batch
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/6d8ba8e8
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/6d8ba8e8
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/6d8ba8e8
Branch: refs/heads/master
Commit: 6d8ba8e89d013f23781cabca190d03aa8d141542
Parents: ac1d6ec
Author: Antoine DESSAIGNE <an...@gmail.com>
Authored: Tue Feb 21 14:55:45 2017 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Tue Feb 21 17:06:28 2017 +0100
----------------------------------------------------------------------
.../src/main/docs/kafka-component.adoc | 15 ++--
.../camel/component/kafka/KafkaConstants.java | 1 +
.../camel/component/kafka/KafkaConsumer.java | 45 +++++-----
.../component/kafka/KafkaConsumerFullTest.java | 4 +-
.../KafkaConsumerLastRecordHeaderTest.java | 92 ++++++++++++++++++++
5 files changed, 129 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/6d8ba8e8/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 211d592..99d804b 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -152,13 +152,14 @@ http://kafka.apache.org/documentation.html#producerconfigs[http://kafka.apache.o
The following headers are available when consuming messages from Kafka.
[width="100%",cols="2m,2m,1m,5",options="header"]
-|=======================================================================================================
-| Header constant | Header value | Type | Description
-| KafkaConstants.TOPIC | "kafka.TOPIC" | String | The topic from where the message originated
-| KafkaConstants.PARTITION | "kafka.PARTITION" | Integer | The partition where the message was stored
-| KafkaConstants.OFFSET | "kafka.OFFSET" | Long | The offset of the message
-| KafkaConstants.KEY | "kafka.KEY" | Object | The key of the message if configured
-|=======================================================================================================
+|=================================================================================================================================================================================================================
+| Header constant | Header value | Type | Description
+| KafkaConstants.TOPIC | "kafka.TOPIC" | String | The topic from where the message originated
+| KafkaConstants.PARTITION | "kafka.PARTITION" | Integer | The partition where the message was stored
+| KafkaConstants.OFFSET | "kafka.OFFSET" | Long | The offset of the message
+| KafkaConstants.KEY | "kafka.KEY" | Object | The key of the message if configured
+| KafkaConstants.LAST_RECORD_BEFORE_COMMIT | "kafka.LAST_RECORD_BEFORE_COMMIT" | Boolean | Whether or not it's the last record before commit (only available if `autoCommitEnable` endpoint parameter is `false`)
+|=================================================================================================================================================================================================================
#### Producer headers
http://git-wip-us.apache.org/repos/asf/camel/blob/6d8ba8e8/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
index 938cb28..49371a6 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
@@ -23,6 +23,7 @@ public final class KafkaConstants {
public static final String KEY = "kafka.KEY";
public static final String TOPIC = "kafka.TOPIC";
public static final String OFFSET = "kafka.OFFSET";
+ public static final String LAST_RECORD_BEFORE_COMMIT = "kafka.LAST_RECORD_BEFORE_COMMIT";
public static final String KAFKA_DEFAULT_ENCODER = "kafka.serializer.DefaultEncoder";
public static final String KAFKA_STRING_ENCODER = "kafka.serializer.StringEncoder";
http://git-wip-us.apache.org/repos/asf/camel/blob/6d8ba8e8/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 0327a76..0e38543 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
@@ -19,6 +19,7 @@ package org.apache.camel.component.kafka;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.Set;
@@ -180,29 +181,33 @@ public class KafkaConsumer extends DefaultConsumer {
while (isRunAllowed() && !isStoppingOrStopped() && !isSuspendingOrSuspended()) {
ConsumerRecords<Object, Object> allRecords = consumer.poll(pollTimeoutMs);
for (TopicPartition partition : allRecords.partitions()) {
- List<ConsumerRecord<Object, Object>> partitionRecords = allRecords
- .records(partition);
- for (ConsumerRecord<Object, Object> record : partitionRecords) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("partition = {}, offset = {}, key = {}, value = {}", record.partition(), record.offset(), record.key(), record.value());
+ Iterator<ConsumerRecord<Object, Object>> recordIterator = allRecords.records(partition).iterator();
+ if (recordIterator.hasNext()) {
+ ConsumerRecord<Object, Object> record = null;
+ while (recordIterator.hasNext()) {
+ record = recordIterator.next();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("partition = {}, offset = {}, key = {}, value = {}", record.partition(), record.offset(), record.key(),
+ record.value());
+ }
+ Exchange exchange = endpoint.createKafkaExchange(record);
+ if (endpoint.getConfiguration().isAutoCommitEnable() != null && !endpoint.getConfiguration().isAutoCommitEnable()) {
+ exchange.getIn().setHeader(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, !recordIterator.hasNext());
+ }
+ try {
+ processor.process(exchange);
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Error during processing", exchange, e);
+ }
}
- Exchange exchange = endpoint.createKafkaExchange(record);
- try {
- processor.process(exchange);
- } catch (Exception e) {
- getExceptionHandler().handleException("Error during processing", exchange, e);
+ long partitionLastOffset = record.offset();
+ if (offsetRepository != null) {
+ offsetRepository.setState(serializeOffsetKey(partition), serializeOffsetValue(partitionLastOffset));
+ // if autocommit is false
+ } else if (endpoint.getConfiguration().isAutoCommitEnable() != null && !endpoint.getConfiguration().isAutoCommitEnable()) {
+ consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(partitionLastOffset + 1)));
}
}
- if (offsetRepository != null) {
- long partitionLastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
- offsetRepository.setState(serializeOffsetKey(partition), serializeOffsetValue(partitionLastOffset));
- } else if (endpoint.getConfiguration().isAutoCommitEnable() != null
- && !endpoint.getConfiguration().isAutoCommitEnable()) {
- // if autocommit is false
- long partitionLastoffset = partitionRecords.get(partitionRecords.size() - 1).offset();
- consumer.commitSync(Collections.singletonMap(
- partition, new OffsetAndMetadata(partitionLastoffset + 1)));
- }
}
}
LOG.info("Unsubscribing {} from topic {}", threadId, topicName);
http://git-wip-us.apache.org/repos/asf/camel/blob/6d8ba8e8/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 5b449bd..183d041 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
@@ -80,6 +80,9 @@ public class KafkaConsumerFullTest extends BaseEmbeddedKafkaTest {
public void kafkaMessageIsConsumedByCamel() throws InterruptedException, IOException {
to.expectedMessageCount(5);
to.expectedBodiesReceivedInAnyOrder("message-0", "message-1", "message-2", "message-3", "message-4");
+ // The LAST_RECORD_BEFORE_COMMIT header should not be configured on any exchange because autoCommitEnable=true
+ to.expectedHeaderValuesReceivedInAnyOrder(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, null, null, null, null, null);
+
for (int k = 0; k < 5; k++) {
String msg = "message-" + k;
ProducerRecord<String, String> data = new ProducerRecord<String, String>(TOPIC, "1", msg);
@@ -119,6 +122,5 @@ public class KafkaConsumerFullTest extends BaseEmbeddedKafkaTest {
// As wee set seek to beginning we should re-consume all messages
to.assertIsSatisfied(3000);
}
-
}
http://git-wip-us.apache.org/repos/asf/camel/blob/6d8ba8e8/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java
new file mode 100644
index 0000000..e772127
--- /dev/null
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerLastRecordHeaderTest.java
@@ -0,0 +1,92 @@
+/**
+ * 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.List;
+import java.util.Properties;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class KafkaConsumerLastRecordHeaderTest extends BaseEmbeddedKafkaTest {
+ private static final String TOPIC = "last-record";
+
+ @EndpointInject(uri = "mock:result")
+ private MockEndpoint result;
+
+ private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
+
+ @Before
+ public void before() {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" + getKafkaPort());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaConstants.KAFKA_DEFAULT_SERIALIZER);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaConstants.KAFKA_DEFAULT_SERIALIZER);
+ props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, KafkaConstants.KAFKA_DEFAULT_PARTITIONER);
+ props.put(ProducerConfig.ACKS_CONFIG, "1");
+
+ producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
+ }
+
+ @After
+ public void after() {
+ if (producer != null) {
+ producer.close();
+ }
+ }
+
+ /**
+ * When consuming data with autoCommitEnable=false
+ * Then the LAST_RECORD_BEFORE_COMMIT header must be always defined
+ * And it should be true only for the last one
+ */
+ @Test
+ public void shouldStartFromBeginningWithEmptyOffsetRepository() throws InterruptedException {
+ result.expectedMessageCount(5);
+ result.expectedBodiesReceived("message-0", "message-1", "message-2", "message-3", "message-4");
+
+ for (int i = 0; i < 5; i++) {
+ producer.send(new ProducerRecord<>(TOPIC, "1", "message-" + i));
+ }
+
+ result.assertIsSatisfied(3000);
+
+ List<Exchange> exchanges = result.getExchanges();
+ for (int i = 0; i < exchanges.size(); i++) {
+ Boolean header = exchanges.get(i).getIn().getHeader(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, Boolean.class);
+ assertNotNull("Header not set for #" + i, header);
+ assertEquals("Header invalid for #" + i, header, i == exchanges.size() - 1);
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("kafka:" + TOPIC + "?groupId=A&autoOffsetReset=earliest&autoCommitEnable=false").to("mock:result");
+ }
+ };
+ }
+}