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/10/25 08:58:06 UTC
[camel] 02/02: CAMEL-11933: camel-kafka - added support for manual
commits
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit c7e39575e34496492e2cf7802afd76a45fa3e072
Author: Claus Ibsen <da...@apache.org>
AuthorDate: Wed Oct 25 10:56:45 2017 +0200
CAMEL-11933: camel-kafka - added support for manual commits
---
.../camel-kafka/src/main/docs/kafka-component.adoc | 22 +++++++++-
.../kafka/DefaultKafkaManualCommitFactory.java | 31 ++++++++++++++
.../camel/component/kafka/KafkaComponent.java | 31 ++++++++++++++
.../camel/component/kafka/KafkaConfiguration.java | 16 +++++++
.../camel/component/kafka/KafkaConsumer.java | 8 +++-
.../component/kafka/KafkaManualCommitFactory.java | 34 +++++++++++++++
.../kafka/KafkaConsumerManualCommitTest.java | 2 +-
.../springboot/KafkaComponentConfiguration.java | 50 ++++++++++++++++++++++
8 files changed, 189 insertions(+), 5 deletions(-)
diff --git a/components/camel-kafka/src/main/docs/kafka-component.adoc b/components/camel-kafka/src/main/docs/kafka-component.adoc
index 415b732..866bd8f 100644
--- a/components/camel-kafka/src/main/docs/kafka-component.adoc
+++ b/components/camel-kafka/src/main/docs/kafka-component.adoc
@@ -32,7 +32,7 @@ kafka:topic[?options]
// component options: START
-The Kafka component supports 6 options which are listed below.
+The Kafka component supports 8 options which are listed below.
@@ -44,6 +44,8 @@ The Kafka component supports 6 options which are listed below.
| *workerPool* (advanced) | To use a shared custom worker pool for continue routing Exchange after kafka server has acknowledge the message that was sent to it from KafkaProducer using asynchronous non-blocking processing. If using this option then you must handle the lifecycle of the thread pool to shut the pool down when no longer needed. | | ExecutorService
| *useGlobalSslContext Parameters* (security) | Enable usage of global SSL context parameters. | false | boolean
| *breakOnFirstError* (consumer) | This options controls what happens when a consumer is processing an exchange and it fails. If the option is false then the consumer continues to the next message and processes it. If the option is true then the consumer breaks out and will seek back to offset of the message that caused a failure and then re-attempt to process this message. However this can lead to endless processing of the same message if its bound to fail every time eg a poison message [...]
+| *allowManualCommit* (consumer) | Whether to allow doing manual commits via KafkaManualCommit. If this option is enabled then an instance of KafkaManualCommit is stored on the Exchange message header which allows end users to access this API and perform manual offset commits via the Kafka consumer. | false | boolean
+| *kafkaManualCommit Factory* (consumer) | Factory to use for creating KafkaManualCommit instances. This allows to plugin a custom factory to create custom KafkaManualCommit instances in case special logic is needed when doing manual commits that deviates from the default implementation that comes out of the box. | | KafkaManualCommit Factory
| *resolveProperty Placeholders* (advanced) | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders. | true | boolean
|===
// component options: END
@@ -67,13 +69,14 @@ with the following path and query parameters:
| *topic* | *Required* Name of the topic to use. On the consumer you can use comma to separate multiple topics. A producer can only send a message to a single topic. | | String
|===
-==== Query Parameters (86 parameters):
+==== Query Parameters (87 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
|===
| Name | Description | Default | Type
| *brokers* (common) | URL of the Kafka brokers to use. The format is host1:port1host2:port2 and the list can be a subset of brokers or a VIP pointing to a subset of brokers. This option is known as bootstrap.servers in the Kafka documentation. | | String
| *clientId* (common) | The client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request. | | String
+| *allowManualCommit* (consumer) | Whether to allow doing manual commits via KafkaManualCommit. If this option is enabled then an instance of KafkaManualCommit is stored on the Exchange message header which allows end users to access this API and perform manual offset commits via the Kafka consumer. | false | boolean
| *autoCommitEnable* (consumer) | If true periodically commit to ZooKeeper the offset of messages already fetched by the consumer. This committed offset will be used when the process fails as the position from which the new consumer will begin. | true | Boolean
| *autoCommitIntervalMs* (consumer) | The frequency in ms that the consumer offsets are committed to zookeeper. | 5000 | Integer
| *autoCommitOnStop* (consumer) | Whether to perform an explicit auto commit when the consumer stops to ensure the broker has a commit from the last consumed message. This requires the option autoCommitEnable is turned on. The possible values are: sync async or none. And sync is the default value. | sync | String
@@ -383,9 +386,20 @@ In XML:
By default the Kafka consumer will use auto commit, where the offset will be committed automaticcally in the background using a given interval.
In case you want to force manual commits, you can use `KafkaManualCommit` API from the Camel Exchange, stored on the message header.
+This requires to turn on manual commits by either setting the option `allowManualCommit` to `true` on the `KafkaComponent`
+or on the endpoint, for example:
[source,java]
----
+KafkaComponent kafka = new KafkaComponent();
+kafka.setAllowManualCommit(true);
+...
+camelContext.addComponent("kafka", kafka);
+----
+
+You can then use the `KafkaManualCommit` from Java code such as a Camel `Processor`:
+[source,java]
+----
public void process(Exchange exchange) {
KafkaManualCommit manual = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
manual.commitSync();
@@ -393,3 +407,7 @@ public void process(Exchange exchange) {
----
This will force a synchronous commit which will block until the commit is acknowledge on Kafka, or if it fails an exception is thrown.
+
+If you want to use a custom implementation of `KafkaManualCommit` then you can configure a custom `KafkaManualCommitFactory`
+on the `KafkaComponent` that creates instances of your custom implementation.
+
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/DefaultKafkaManualCommitFactory.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/DefaultKafkaManualCommitFactory.java
new file mode 100644
index 0000000..3760f90
--- /dev/null
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/DefaultKafkaManualCommitFactory.java
@@ -0,0 +1,31 @@
+/**
+ * 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 org.apache.camel.Exchange;
+import org.apache.camel.spi.StateRepository;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+
+public class DefaultKafkaManualCommitFactory implements KafkaManualCommitFactory {
+
+ @Override
+ public KafkaManualCommit newInstance(Exchange exchange, KafkaConsumer consumer, String topicName,
+ String threadId, StateRepository<String, String> offsetRepository, TopicPartition partition, long partitionLastOffset) {
+ return new DefaultKafkaManualCommit(consumer, topicName, threadId, offsetRepository, partition, partitionLastOffset);
+ }
+}
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
index 175dda1..09e2299 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
@@ -36,6 +36,10 @@ public class KafkaComponent extends UriEndpointComponent implements SSLContextPa
private boolean useGlobalSslContextParameters;
@Metadata(label = "consumer", defaultValue = "false")
private boolean breakOnFirstError;
+ @Metadata(label = "consumer", defaultValue = "false")
+ private boolean allowManualCommit;
+ @Metadata(label = "consumer,advanced")
+ private KafkaManualCommitFactory kafkaManualCommitFactory = new DefaultKafkaManualCommitFactory();
public KafkaComponent() {
super(KafkaEndpoint.class);
@@ -61,6 +65,7 @@ public class KafkaComponent extends UriEndpointComponent implements SSLContextPa
endpoint.getConfiguration().setTopic(remaining);
endpoint.getConfiguration().setWorkerPool(getWorkerPool());
endpoint.getConfiguration().setBreakOnFirstError(isBreakOnFirstError());
+ endpoint.getConfiguration().setAllowManualCommit(isAllowManualCommit());
// brokers can be configured on either component or endpoint level
// and the consumer and produce is aware of this and act accordingly
@@ -147,4 +152,30 @@ public class KafkaComponent extends UriEndpointComponent implements SSLContextPa
}
+ public boolean isAllowManualCommit() {
+ return allowManualCommit;
+ }
+
+ /**
+ * Whether to allow doing manual commits via {@link KafkaManualCommit}.
+ * <p/>
+ * If this option is enabled then an instance of {@link KafkaManualCommit} is stored on the {@link Exchange} message header,
+ * which allows end users to access this API and perform manual offset commits via the Kafka consumer.
+ */
+ public void setAllowManualCommit(boolean allowManualCommit) {
+ this.allowManualCommit = allowManualCommit;
+ }
+
+ public KafkaManualCommitFactory getKafkaManualCommitFactory() {
+ return kafkaManualCommitFactory;
+ }
+
+ /**
+ * Factory to use for creating {@link KafkaManualCommit} instances. This allows to plugin a custom factory
+ * to create custom {@link KafkaManualCommit} instances in case special logic is needed when doing manual commits
+ * that deviates from the default implementation that comes out of the box.
+ */
+ public void setKafkaManualCommitFactory(KafkaManualCommitFactory kafkaManualCommitFactory) {
+ this.kafkaManualCommitFactory = kafkaManualCommitFactory;
+ }
}
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 e724ef6..edb0e18 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
@@ -113,6 +113,8 @@ public class KafkaConfiguration implements Cloneable {
//Consumer configuration properties
@UriParam(label = "consumer", defaultValue = "true")
private Boolean autoCommitEnable = true;
+ @UriParam(label = "consumer")
+ private boolean allowManualCommit;
@UriParam(label = "consumer", defaultValue = "sync", enums = "sync,async,none")
private String autoCommitOnStop = "sync";
@UriParam(label = "consumer")
@@ -604,6 +606,20 @@ public class KafkaConfiguration implements Cloneable {
this.autoCommitEnable = autoCommitEnable;
}
+ public boolean isAllowManualCommit() {
+ return allowManualCommit;
+ }
+
+ /**
+ * Whether to allow doing manual commits via {@link KafkaManualCommit}.
+ * <p/>
+ * If this option is enabled then an instance of {@link KafkaManualCommit} is stored on the {@link Exchange} message header,
+ * which allows end users to access this API and perform manual offset commits via the Kafka consumer.
+ */
+ public void setAllowManualCommit(boolean allowManualCommit) {
+ this.allowManualCommit = allowManualCommit;
+ }
+
public StateRepository<String, String> getOffsetRepository() {
return offsetRepository;
}
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 a26bf81..0fca9e7 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
@@ -244,8 +244,12 @@ public class KafkaConsumer extends DefaultConsumer {
if (!isAutoCommitEnabled()) {
exchange.getIn().setHeader(KafkaConstants.LAST_RECORD_BEFORE_COMMIT, !recordIterator.hasNext());
}
- // allow Camel users to access the Kafka consumer API to be able to do for example manual commits
- exchange.getIn().setHeader(KafkaConstants.MANUAL_COMMIT, new DefaultKafkaManualCommit(consumer, topicName, threadId, offsetRepository, partition, partitionLastOffset));
+ if (endpoint.getComponent().isAllowManualCommit()) {
+ // allow Camel users to access the Kafka consumer API to be able to do for example manual commits
+ KafkaManualCommit manual = endpoint.getComponent().getKafkaManualCommitFactory().newInstance(exchange, consumer, topicName, threadId,
+ offsetRepository, partition, partitionLastOffset);
+ exchange.getIn().setHeader(KafkaConstants.MANUAL_COMMIT, manual);
+ }
try {
processor.process(exchange);
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaManualCommitFactory.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaManualCommitFactory.java
new file mode 100644
index 0000000..e900c10
--- /dev/null
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaManualCommitFactory.java
@@ -0,0 +1,34 @@
+/**
+ * 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 org.apache.camel.Exchange;
+import org.apache.camel.spi.StateRepository;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ * Factory to create a new {@link KafkaManualCommit} to store on the {@link Exchange}.
+ */
+public interface KafkaManualCommitFactory {
+
+ /**
+ * Creates a new instance
+ */
+ KafkaManualCommit newInstance(Exchange exchange, KafkaConsumer consumer, String topicName, String threadId,
+ StateRepository<String, String> offsetRepository, TopicPartition partition, long partitionLastOffset);
+}
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java
index 7e8ca1f..c69f3b0 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaConsumerManualCommitTest.java
@@ -36,7 +36,7 @@ public class KafkaConsumerManualCommitTest extends BaseEmbeddedKafkaTest {
public static final String TOPIC = "test";
@EndpointInject(uri = "kafka:" + TOPIC
- + "?groupId=group1&sessionTimeoutMs=30000&autoCommitEnable=false&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
+ + "?groupId=group1&sessionTimeoutMs=30000&autoCommitEnable=false&allowManualCommit=true&interceptorClasses=org.apache.camel.component.kafka.MockConsumerInterceptor")
private Endpoint from;
@EndpointInject(uri = "mock:result")
diff --git a/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
index 03c1f48..d7d4c83 100644
--- a/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-kafka-starter/src/main/java/org/apache/camel/component/kafka/springboot/KafkaComponentConfiguration.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.kafka.springboot;
import java.util.concurrent.ExecutorService;
import javax.annotation.Generated;
+import org.apache.camel.component.kafka.KafkaManualCommitFactory;
import org.apache.camel.spi.StateRepository;
import org.apache.camel.spring.boot.ComponentConfigurationPropertiesCommon;
import org.apache.camel.util.jsse.SSLContextParameters;
@@ -72,6 +73,21 @@ public class KafkaComponentConfiguration
*/
private Boolean breakOnFirstError = false;
/**
+ * Whether to allow doing manual commits via KafkaManualCommit. If this
+ * option is enabled then an instance of KafkaManualCommit is stored on the
+ * Exchange message header which allows end users to access this API and
+ * perform manual offset commits via the Kafka consumer.
+ */
+ private Boolean allowManualCommit = false;
+ /**
+ * Factory to use for creating KafkaManualCommit instances. This allows to
+ * plugin a custom factory to create custom KafkaManualCommit instances in
+ * case special logic is needed when doing manual commits that deviates from
+ * the default implementation that comes out of the box.
+ */
+ @NestedConfigurationProperty
+ private KafkaManualCommitFactory kafkaManualCommitFactory;
+ /**
* Whether the component should resolve property placeholders on itself when
* starting. Only properties which are of String type can use property
* placeholders.
@@ -120,6 +136,23 @@ public class KafkaComponentConfiguration
this.breakOnFirstError = breakOnFirstError;
}
+ public Boolean getAllowManualCommit() {
+ return allowManualCommit;
+ }
+
+ public void setAllowManualCommit(Boolean allowManualCommit) {
+ this.allowManualCommit = allowManualCommit;
+ }
+
+ public KafkaManualCommitFactory getKafkaManualCommitFactory() {
+ return kafkaManualCommitFactory;
+ }
+
+ public void setKafkaManualCommitFactory(
+ KafkaManualCommitFactory kafkaManualCommitFactory) {
+ this.kafkaManualCommitFactory = kafkaManualCommitFactory;
+ }
+
public Boolean getResolvePropertyPlaceholders() {
return resolvePropertyPlaceholders;
}
@@ -187,6 +220,15 @@ public class KafkaComponentConfiguration
*/
private Boolean autoCommitEnable = true;
/**
+ * Whether to allow doing manual commits via {@link KafkaManualCommit} .
+ * <p/>
+ * If this option is enabled then an instance of
+ * {@link KafkaManualCommit} is stored on the {@link Exchange} message
+ * header, which allows end users to access this API and perform manual
+ * offset commits via the Kafka consumer.
+ */
+ private Boolean allowManualCommit = false;
+ /**
* The offset repository to use in order to locally store the offset of
* each partition of the topic. Defining one will disable the
* autocommit.
@@ -760,6 +802,14 @@ public class KafkaComponentConfiguration
this.autoCommitEnable = autoCommitEnable;
}
+ public Boolean getAllowManualCommit() {
+ return allowManualCommit;
+ }
+
+ public void setAllowManualCommit(Boolean allowManualCommit) {
+ this.allowManualCommit = allowManualCommit;
+ }
+
public StateRepository getOffsetRepository() {
return offsetRepository;
}
--
To stop receiving notification emails like this one, please contact
"commits@camel.apache.org" <co...@camel.apache.org>.