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 <[email protected]> Authored: Tue Feb 21 14:55:45 2017 +0100 Committer: Claus Ibsen <[email protected]> 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"); + } + }; + } +}
