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 2016/05/13 09:31:59 UTC
[1/5] camel git commit: CAMEL-9957: Fixed as it was not correct.
Repository: camel
Updated Branches:
refs/heads/camel-2.17.x 772b2c30e -> d5bd35a67
refs/heads/master 64e5b56c5 -> 30a2ed5bd
CAMEL-9957: Fixed as it was not correct.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/c540a426
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/c540a426
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/c540a426
Branch: refs/heads/master
Commit: c540a426485dee518ddaf1c59ef28e03469951ba
Parents: 64e5b56
Author: Claus Ibsen <da...@apache.org>
Authored: Fri May 13 10:51:57 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri May 13 10:51:57 2016 +0200
----------------------------------------------------------------------
.../camel/component/kafka/KafkaProducer.java | 27 ++++++++++++--------
.../component/kafka/KafkaProducerTest.java | 6 +----
2 files changed, 18 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/c540a426/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 0c4013f..6c432d6 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -73,6 +73,7 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
}
}
+ @SuppressWarnings("unchecked")
protected ProducerRecord createRecorder(Exchange exchange) throws CamelException {
String topic = endpoint.getTopic();
if (!endpoint.isBridgeEndpoint()) {
@@ -103,19 +104,25 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
@Override
@SuppressWarnings("unchecked")
- public void process(Exchange exchange) throws CamelException {
-
+ public void process(Exchange exchange) throws Exception {
ProducerRecord record = createRecorder(exchange);
// Just send out the record in the sync way
- try {
- kafkaProducer.send(record).get();
- } catch (Exception e) {
- throw new CamelException(e);
- }
+ kafkaProducer.send(record).get();
}
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
+ // force processing synchronously using different api
+ if (endpoint.isSynchronous()) {
+ try {
+ process(exchange);
+ } catch (Throwable e) {
+ exchange.setException(e);
+ }
+ callback.done(true);
+ return true;
+ }
+
try {
ProducerRecord record = createRecorder(exchange);
kafkaProducer.send(record, new KafkaProducerCallBack(exchange, callback));
@@ -129,10 +136,10 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
}
}
- class KafkaProducerCallBack implements Callback {
+ private final class KafkaProducerCallBack implements Callback {
- private Exchange exchange;
- private AsyncCallback callback;
+ private final Exchange exchange;
+ private final AsyncCallback callback;
KafkaProducerCallBack(Exchange exchange, AsyncCallback callback) {
this.exchange = exchange;
http://git-wip-us.apache.org/repos/asf/camel/blob/c540a426/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 68798f3..40f2113 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -79,7 +79,7 @@ public class KafkaProducerTest {
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class));
}
- @Test(expected=CamelException.class)
+ @Test(expected = Exception.class)
@SuppressWarnings({"unchecked"})
public void processSendsMessageWithException() throws Exception {
endpoint.setTopic("sometopic");
@@ -90,7 +90,6 @@ public class KafkaProducerTest {
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
producer.process(exchange);
-
}
@Test
@@ -103,10 +102,8 @@ public class KafkaProducerTest {
producer.process(exchange, callback);
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), Matchers.any(Callback.class));
-
}
-
@Test
public void processAsyncSendsMessageWithException() throws Exception {
@@ -126,7 +123,6 @@ public class KafkaProducerTest {
Mockito.verify(callback).done(Matchers.eq(true));
}
-
@Test
public void processSendsMessageWithTopicHeaderAndNoTopicInEndPoint() throws Exception {
endpoint.setTopic(null);
[5/5] camel git commit: CAMEL-9957: Use a worker pool for processing
kafka callbacks to not streal its io thread to do camel routing. See the
kafka javadoc/documentation.
Posted by da...@apache.org.
CAMEL-9957: Use a worker pool for processing kafka callbacks to not streal its io thread to do camel routing. See the kafka javadoc/documentation.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d5bd35a6
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d5bd35a6
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d5bd35a6
Branch: refs/heads/camel-2.17.x
Commit: d5bd35a67e595d08be76edb077b9595e97f90b24
Parents: 1fd632c
Author: Claus Ibsen <da...@apache.org>
Authored: Fri May 13 11:31:49 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri May 13 11:31:49 2016 +0200
----------------------------------------------------------------------
.../apache/camel/component/kafka/KafkaConfiguration.java | 11 -----------
1 file changed, 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/d5bd35a6/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 1a98b28..06f1de1 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
@@ -1154,17 +1154,6 @@ public class KafkaConfiguration {
this.valueDeserializer = valueDeserializer;
}
- public boolean isSeekToBeginning() {
- return seekToBeginning;
- }
-
- /**
- * If the option is true, then KafkaConsumer will read from beginning on startup.
- */
- public void setSeekToBeginning(boolean seekToBeginning) {
- this.seekToBeginning = seekToBeginning;
- }
-
public ExecutorService getWorkerPool() {
return workerPool;
}
[3/5] camel git commit: CAMEL-9957: Fixed as it was not correct.
Posted by da...@apache.org.
CAMEL-9957: Fixed as it was not correct.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/81e1f8a0
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/81e1f8a0
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/81e1f8a0
Branch: refs/heads/camel-2.17.x
Commit: 81e1f8a0c8a118262223c6c2c85465735fc07a17
Parents: 772b2c3
Author: Claus Ibsen <da...@apache.org>
Authored: Fri May 13 10:51:57 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri May 13 11:27:39 2016 +0200
----------------------------------------------------------------------
.../camel/component/kafka/KafkaProducer.java | 27 ++++++++++++--------
.../component/kafka/KafkaProducerTest.java | 6 +----
2 files changed, 18 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/81e1f8a0/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 0c4013f..6c432d6 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -73,6 +73,7 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
}
}
+ @SuppressWarnings("unchecked")
protected ProducerRecord createRecorder(Exchange exchange) throws CamelException {
String topic = endpoint.getTopic();
if (!endpoint.isBridgeEndpoint()) {
@@ -103,19 +104,25 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
@Override
@SuppressWarnings("unchecked")
- public void process(Exchange exchange) throws CamelException {
-
+ public void process(Exchange exchange) throws Exception {
ProducerRecord record = createRecorder(exchange);
// Just send out the record in the sync way
- try {
- kafkaProducer.send(record).get();
- } catch (Exception e) {
- throw new CamelException(e);
- }
+ kafkaProducer.send(record).get();
}
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
+ // force processing synchronously using different api
+ if (endpoint.isSynchronous()) {
+ try {
+ process(exchange);
+ } catch (Throwable e) {
+ exchange.setException(e);
+ }
+ callback.done(true);
+ return true;
+ }
+
try {
ProducerRecord record = createRecorder(exchange);
kafkaProducer.send(record, new KafkaProducerCallBack(exchange, callback));
@@ -129,10 +136,10 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
}
}
- class KafkaProducerCallBack implements Callback {
+ private final class KafkaProducerCallBack implements Callback {
- private Exchange exchange;
- private AsyncCallback callback;
+ private final Exchange exchange;
+ private final AsyncCallback callback;
KafkaProducerCallBack(Exchange exchange, AsyncCallback callback) {
this.exchange = exchange;
http://git-wip-us.apache.org/repos/asf/camel/blob/81e1f8a0/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 68798f3..40f2113 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -79,7 +79,7 @@ public class KafkaProducerTest {
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class));
}
- @Test(expected=CamelException.class)
+ @Test(expected = Exception.class)
@SuppressWarnings({"unchecked"})
public void processSendsMessageWithException() throws Exception {
endpoint.setTopic("sometopic");
@@ -90,7 +90,6 @@ public class KafkaProducerTest {
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
producer.process(exchange);
-
}
@Test
@@ -103,10 +102,8 @@ public class KafkaProducerTest {
producer.process(exchange, callback);
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), Matchers.any(Callback.class));
-
}
-
@Test
public void processAsyncSendsMessageWithException() throws Exception {
@@ -126,7 +123,6 @@ public class KafkaProducerTest {
Mockito.verify(callback).done(Matchers.eq(true));
}
-
@Test
public void processSendsMessageWithTopicHeaderAndNoTopicInEndPoint() throws Exception {
endpoint.setTopic(null);
[2/5] camel git commit: CAMEL-9957: Use a worker pool for processing
kafka callbacks to not streal its io thread to do camel routing. See the
kafka javadoc/documentation.
Posted by da...@apache.org.
CAMEL-9957: Use a worker pool for processing kafka callbacks to not streal its io thread to do camel routing. See the kafka javadoc/documentation.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/30a2ed5b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/30a2ed5b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/30a2ed5b
Branch: refs/heads/master
Commit: 30a2ed5bdcc2066c41784b878baeafb084aa4007
Parents: c540a42
Author: Claus Ibsen <da...@apache.org>
Authored: Fri May 13 11:20:03 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri May 13 11:27:18 2016 +0200
----------------------------------------------------------------------
components/camel-kafka/src/main/docs/kafka.adoc | 27 +++++++++++-
.../camel/component/kafka/KafkaComponent.java | 21 ++++++++++
.../component/kafka/KafkaConfiguration.java | 43 ++++++++++++++++++++
.../camel/component/kafka/KafkaEndpoint.java | 31 +++++++++++++-
.../camel/component/kafka/KafkaProducer.java | 40 +++++++++++++++---
5 files changed, 153 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/30a2ed5b/components/camel-kafka/src/main/docs/kafka.adoc
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/docs/kafka.adoc b/components/camel-kafka/src/main/docs/kafka.adoc
index 7fdec49..2811a62 100644
--- a/components/camel-kafka/src/main/docs/kafka.adoc
+++ b/components/camel-kafka/src/main/docs/kafka.adoc
@@ -59,16 +59,33 @@ Options (Camel 2.16 or older)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+
// component options: START
-The Kafka component has no options.
+The Kafka component supports 1 options which are listed below.
+
+
+
+{% raw %}
+[width="100%",cols="2s,1m,8",options="header"]
+|=======================================================================
+| Name | Java Type | Description
+| workerPool | ExecutorService | 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.
+|=======================================================================
+{% endraw %}
// component options: END
+
+
+
+
+
// endpoint options: START
-The Kafka component supports 71 endpoint options which are listed below:
+The Kafka component supports 74 endpoint options which are listed below:
{% raw %}
[width="100%",cols="2s,1,1m,1m,5",options="header"]
@@ -143,6 +160,9 @@ The Kafka component supports 71 endpoint options which are listed below:
| sslTruststoreLocation | producer | | String | The location of the trust store file.
| sslTruststorePassword | producer | | String | The password for the trust store file.
| sslTruststoreType | producer | JKS | String | The file format of the trust store file. Default value is JKS.
+| workerPool | producer | | ExecutorService | To use a 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.
+| workerPoolCoreSize | producer | 10 | Integer | Number of core threads for the 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.
+| workerPoolMaxSize | producer | 20 | Integer | Maximum number of threads for the 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.
| exchangePattern | advanced | InOnly | ExchangePattern | Sets the default exchange pattern when creating an exchange
| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
|=======================================================================
@@ -152,6 +172,9 @@ The Kafka component supports 71 endpoint options which are listed below:
+
+
+
For more information about Producer/Consumer configuration:
http://kafka.apache.org/documentation.html#newconsumerconfigs[http://kafka.apache.org/documentation.html#newconsumerconfigs]
http://git-wip-us.apache.org/repos/asf/camel/blob/30a2ed5b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
----------------------------------------------------------------------
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 2981b3f..76ef55f 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
@@ -17,8 +17,10 @@
package org.apache.camel.component.kafka;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.impl.UriEndpointComponent;
public class KafkaComponent extends UriEndpointComponent {
@@ -31,6 +33,8 @@ public class KafkaComponent extends UriEndpointComponent {
super(context, KafkaEndpoint.class);
}
+ private ExecutorService workerPool;
+
@Override
protected KafkaEndpoint createEndpoint(String uri, String remaining, Map<String, Object> params) throws Exception {
KafkaEndpoint endpoint = new KafkaEndpoint(uri, this);
@@ -38,8 +42,25 @@ public class KafkaComponent extends UriEndpointComponent {
if (brokers != null) {
endpoint.getConfiguration().setBrokers(brokers);
}
+
+ // configure component options before endpoint properties which can override from params
+ endpoint.getConfiguration().setWorkerPool(workerPool);
+
setProperties(endpoint, params);
return endpoint;
}
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ /**
+ * To use a shared custom worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link 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.
+ */
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/30a2ed5b/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 e0580e9..1a068c3 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
@@ -17,7 +17,9 @@
package org.apache.camel.component.kafka;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import org.apache.camel.Exchange;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
@@ -102,6 +104,13 @@ public class KafkaConfiguration {
@UriParam(label = "producer", defaultValue = "100")
private Integer retryBackoffMs = 100;
+ @UriParam(label = "producer")
+ private ExecutorService workerPool;
+ @UriParam(label = "producer", defaultValue = "10")
+ private Integer workerPoolCoreSize = 10;
+ @UriParam(label = "producer", defaultValue = "20")
+ private Integer workerPoolMaxSize = 20;
+
//Async producer config
@UriParam(label = "producer", defaultValue = "10000")
private Integer queueBufferingMaxMessages = 10000;
@@ -1158,5 +1167,39 @@ public class KafkaConfiguration {
this.seekToBeginning = seekToBeginning;
}
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ /**
+ * To use a custom worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link KafkaProducer} using asynchronous non-blocking processing.
+ */
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
+ }
+
+ public Integer getWorkerPoolCoreSize() {
+ return workerPoolCoreSize;
+ }
+
+ /**
+ * Number of core threads for the worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link KafkaProducer} using asynchronous non-blocking processing.
+ */
+ public void setWorkerPoolCoreSize(Integer workerPoolCoreSize) {
+ this.workerPoolCoreSize = workerPoolCoreSize;
+ }
+ public Integer getWorkerPoolMaxSize() {
+ return workerPoolMaxSize;
+ }
+
+ /**
+ * Maximum number of threads for the worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link KafkaProducer} using asynchronous non-blocking processing.
+ */
+ public void setWorkerPoolMaxSize(Integer workerPoolMaxSize) {
+ this.workerPoolMaxSize = workerPoolMaxSize;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/30a2ed5b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
index 1c239c8..aabd020 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
@@ -86,7 +86,13 @@ public class KafkaEndpoint extends DefaultEndpoint implements MultipleConsumersS
}
public ExecutorService createExecutor() {
- return getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, "KafkaTopic[" + configuration.getTopic() + "]", configuration.getConsumerStreams());
+ return getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, "KafkaConsumer[" + configuration.getTopic() + "]", configuration.getConsumerStreams());
+ }
+
+ public ExecutorService createProducerExecutor() {
+ int core = getConfiguration().getWorkerPoolCoreSize();
+ int max = getConfiguration().getWorkerPoolMaxSize();
+ return getCamelContext().getExecutorServiceManager().newThreadPool(this, "KafkaProducer[" + configuration.getTopic() + "]", core, max);
}
public Exchange createKafkaExchange(ConsumerRecord record) {
@@ -658,4 +664,27 @@ public class KafkaEndpoint extends DefaultEndpoint implements MultipleConsumersS
this.bridgeEndpoint = bridgeEndpoint;
}
+ public void setWorkerPool(ExecutorService workerPool) {
+ configuration.setWorkerPool(workerPool);
+ }
+
+ public void setWorkerPoolMaxSize(Integer workerPoolMaxSize) {
+ configuration.setWorkerPoolMaxSize(workerPoolMaxSize);
+ }
+
+ public Integer getWorkerPoolMaxSize() {
+ return configuration.getWorkerPoolMaxSize();
+ }
+
+ public Integer getWorkerPoolCoreSize() {
+ return configuration.getWorkerPoolCoreSize();
+ }
+
+ public ExecutorService getWorkerPool() {
+ return configuration.getWorkerPool();
+ }
+
+ public void setWorkerPoolCoreSize(Integer workerPoolCoreSize) {
+ configuration.setWorkerPoolCoreSize(workerPoolCoreSize);
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/30a2ed5b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 6c432d6..0f783b5 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.kafka;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
@@ -33,6 +34,8 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
private org.apache.kafka.clients.producer.KafkaProducer kafkaProducer;
private final KafkaEndpoint endpoint;
+ private ExecutorService workerPool;
+ private boolean shutdownWorkerPool;
public KafkaProducer(KafkaEndpoint endpoint) {
super(endpoint);
@@ -58,11 +61,12 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
this.kafkaProducer = kafkaProducer;
}
- @Override
- protected void doStop() throws Exception {
- if (kafkaProducer != null) {
- kafkaProducer.close();
- }
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
}
@Override
@@ -71,6 +75,23 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
if (kafkaProducer == null) {
kafkaProducer = new org.apache.kafka.clients.producer.KafkaProducer(props);
}
+
+ // if we are in asynchronous mode we need a worker pool
+ if (!endpoint.isSynchronous() && workerPool == null) {
+ workerPool = endpoint.createProducerExecutor();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (kafkaProducer != null) {
+ kafkaProducer.close();
+ }
+
+ if (shutdownWorkerPool && workerPool != null) {
+ endpoint.getCamelContext().getExecutorServiceManager().shutdown(workerPool);
+ workerPool = null;
+ }
}
@SuppressWarnings("unchecked")
@@ -152,7 +173,14 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
// Just set the exception back
exchange.setException(e);
}
- callback.done(false);
+ // use worker pool to continue routing the exchange
+ // as this thread is from Kafka Callback and should not be used by Camel routing
+ workerPool.submit(new Runnable() {
+ @Override
+ public void run() {
+ callback.done(false);
+ }
+ });
}
}
}
[4/5] camel git commit: CAMEL-9957: Use a worker pool for processing
kafka callbacks to not streal its io thread to do camel routing. See the
kafka javadoc/documentation.
Posted by da...@apache.org.
CAMEL-9957: Use a worker pool for processing kafka callbacks to not streal its io thread to do camel routing. See the kafka javadoc/documentation.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/1fd632cf
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/1fd632cf
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/1fd632cf
Branch: refs/heads/camel-2.17.x
Commit: 1fd632cfc3741c86cc58cfa69f5935005ddb5cfa
Parents: 81e1f8a
Author: Claus Ibsen <da...@apache.org>
Authored: Fri May 13 11:20:03 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri May 13 11:29:33 2016 +0200
----------------------------------------------------------------------
.../camel/component/kafka/KafkaComponent.java | 21 ++++++++
.../component/kafka/KafkaConfiguration.java | 55 ++++++++++++++++++++
.../camel/component/kafka/KafkaEndpoint.java | 32 +++++++++++-
.../camel/component/kafka/KafkaProducer.java | 40 +++++++++++---
4 files changed, 141 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/1fd632cf/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaComponent.java
----------------------------------------------------------------------
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 2981b3f..76ef55f 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
@@ -17,8 +17,10 @@
package org.apache.camel.component.kafka;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.impl.UriEndpointComponent;
public class KafkaComponent extends UriEndpointComponent {
@@ -31,6 +33,8 @@ public class KafkaComponent extends UriEndpointComponent {
super(context, KafkaEndpoint.class);
}
+ private ExecutorService workerPool;
+
@Override
protected KafkaEndpoint createEndpoint(String uri, String remaining, Map<String, Object> params) throws Exception {
KafkaEndpoint endpoint = new KafkaEndpoint(uri, this);
@@ -38,8 +42,25 @@ public class KafkaComponent extends UriEndpointComponent {
if (brokers != null) {
endpoint.getConfiguration().setBrokers(brokers);
}
+
+ // configure component options before endpoint properties which can override from params
+ endpoint.getConfiguration().setWorkerPool(workerPool);
+
setProperties(endpoint, params);
return endpoint;
}
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ /**
+ * To use a shared custom worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link 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.
+ */
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1fd632cf/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 6b67fc5..1a98b28 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
@@ -17,7 +17,9 @@
package org.apache.camel.component.kafka;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import org.apache.camel.Exchange;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
@@ -100,6 +102,13 @@ public class KafkaConfiguration {
@UriParam(label = "producer", defaultValue = "100")
private Integer retryBackoffMs = 100;
+ @UriParam(label = "producer")
+ private ExecutorService workerPool;
+ @UriParam(label = "producer", defaultValue = "10")
+ private Integer workerPoolCoreSize = 10;
+ @UriParam(label = "producer", defaultValue = "20")
+ private Integer workerPoolMaxSize = 20;
+
//Async producer config
@UriParam(label = "producer", defaultValue = "10000")
private Integer queueBufferingMaxMessages = 10000;
@@ -1145,4 +1154,50 @@ public class KafkaConfiguration {
this.valueDeserializer = valueDeserializer;
}
+ public boolean isSeekToBeginning() {
+ return seekToBeginning;
+ }
+
+ /**
+ * If the option is true, then KafkaConsumer will read from beginning on startup.
+ */
+ public void setSeekToBeginning(boolean seekToBeginning) {
+ this.seekToBeginning = seekToBeginning;
+ }
+
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ /**
+ * To use a custom worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link KafkaProducer} using asynchronous non-blocking processing.
+ */
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
+ }
+
+ public Integer getWorkerPoolCoreSize() {
+ return workerPoolCoreSize;
+ }
+
+ /**
+ * Number of core threads for the worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link KafkaProducer} using asynchronous non-blocking processing.
+ */
+ public void setWorkerPoolCoreSize(Integer workerPoolCoreSize) {
+ this.workerPoolCoreSize = workerPoolCoreSize;
+ }
+
+ public Integer getWorkerPoolMaxSize() {
+ return workerPoolMaxSize;
+ }
+
+ /**
+ * Maximum number of threads for the worker pool for continue routing {@link Exchange} after kafka server has acknowledge
+ * the message that was sent to it from {@link KafkaProducer} using asynchronous non-blocking processing.
+ */
+ public void setWorkerPoolMaxSize(Integer workerPoolMaxSize) {
+ this.workerPoolMaxSize = workerPoolMaxSize;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1fd632cf/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
index 369537a..d995475 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaEndpoint.java
@@ -86,7 +86,13 @@ public class KafkaEndpoint extends DefaultEndpoint implements MultipleConsumersS
}
public ExecutorService createExecutor() {
- return getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, "KafkaTopic[" + configuration.getTopic() + "]", configuration.getConsumerStreams());
+ return getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, "KafkaConsumer[" + configuration.getTopic() + "]", configuration.getConsumerStreams());
+ }
+
+ public ExecutorService createProducerExecutor() {
+ int core = getConfiguration().getWorkerPoolCoreSize();
+ int max = getConfiguration().getWorkerPoolMaxSize();
+ return getCamelContext().getExecutorServiceManager().newThreadPool(this, "KafkaProducer[" + configuration.getTopic() + "]", core, max);
}
public Exchange createKafkaExchange(ConsumerRecord record) {
@@ -649,4 +655,28 @@ public class KafkaEndpoint extends DefaultEndpoint implements MultipleConsumersS
public void setBridgeEndpoint(boolean bridgeEndpoint) {
this.bridgeEndpoint = bridgeEndpoint;
}
+
+ public void setWorkerPool(ExecutorService workerPool) {
+ configuration.setWorkerPool(workerPool);
+ }
+
+ public void setWorkerPoolMaxSize(Integer workerPoolMaxSize) {
+ configuration.setWorkerPoolMaxSize(workerPoolMaxSize);
+ }
+
+ public Integer getWorkerPoolMaxSize() {
+ return configuration.getWorkerPoolMaxSize();
+ }
+
+ public Integer getWorkerPoolCoreSize() {
+ return configuration.getWorkerPoolCoreSize();
+ }
+
+ public ExecutorService getWorkerPool() {
+ return configuration.getWorkerPool();
+ }
+
+ public void setWorkerPoolCoreSize(Integer workerPoolCoreSize) {
+ configuration.setWorkerPoolCoreSize(workerPoolCoreSize);
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/1fd632cf/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 6c432d6..0f783b5 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.kafka;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
@@ -33,6 +34,8 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
private org.apache.kafka.clients.producer.KafkaProducer kafkaProducer;
private final KafkaEndpoint endpoint;
+ private ExecutorService workerPool;
+ private boolean shutdownWorkerPool;
public KafkaProducer(KafkaEndpoint endpoint) {
super(endpoint);
@@ -58,11 +61,12 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
this.kafkaProducer = kafkaProducer;
}
- @Override
- protected void doStop() throws Exception {
- if (kafkaProducer != null) {
- kafkaProducer.close();
- }
+ public ExecutorService getWorkerPool() {
+ return workerPool;
+ }
+
+ public void setWorkerPool(ExecutorService workerPool) {
+ this.workerPool = workerPool;
}
@Override
@@ -71,6 +75,23 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
if (kafkaProducer == null) {
kafkaProducer = new org.apache.kafka.clients.producer.KafkaProducer(props);
}
+
+ // if we are in asynchronous mode we need a worker pool
+ if (!endpoint.isSynchronous() && workerPool == null) {
+ workerPool = endpoint.createProducerExecutor();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (kafkaProducer != null) {
+ kafkaProducer.close();
+ }
+
+ if (shutdownWorkerPool && workerPool != null) {
+ endpoint.getCamelContext().getExecutorServiceManager().shutdown(workerPool);
+ workerPool = null;
+ }
}
@SuppressWarnings("unchecked")
@@ -152,7 +173,14 @@ public class KafkaProducer extends DefaultProducer implements AsyncProcessor {
// Just set the exception back
exchange.setException(e);
}
- callback.done(false);
+ // use worker pool to continue routing the exchange
+ // as this thread is from Kafka Callback and should not be used by Camel routing
+ workerPool.submit(new Runnable() {
+ @Override
+ public void run() {
+ callback.done(false);
+ }
+ });
}
}
}