This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch exchange-factory in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/exchange-factory by this push: new d03e874 CAMEL-16222: PooledExchangeFactory experiment d03e874 is described below commit d03e87461f1ffcc346cfc8fb56bfc28ccf37adb6 Author: Claus Ibsen <claus.ib...@gmail.com> AuthorDate: Sun Feb 21 19:54:21 2021 +0100 CAMEL-16222: PooledExchangeFactory experiment --- .../instance/HazelcastInstanceConsumer.java | 4 +- .../hazelcast/list/HazelcastListConsumer.java | 6 --- .../hazelcast/listener/CamelListener.java | 4 +- .../hazelcast/map/HazelcastMapConsumer.java | 6 --- .../multimap/HazelcastMultimapConsumer.java | 6 --- .../hazelcast/queue/HazelcastQueueConsumer.java | 4 +- .../HazelcastReplicatedmapConsumer.java | 6 --- .../hazelcast/seda/HazelcastSedaConsumer.java | 32 ++++++++-------- .../hazelcast/set/HazelcastSetConsumer.java | 6 --- .../hazelcast/topic/HazelcastTopicConsumer.java | 6 --- .../camel/component/hbase/HBaseConsumer.java | 5 ++- .../apache/camel/component/hdfs/HdfsConsumer.java | 44 ++++++++++++---------- 12 files changed, 51 insertions(+), 78 deletions(-) diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/instance/HazelcastInstanceConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/instance/HazelcastInstanceConsumer.java index 3dc0450..6a9b392 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/instance/HazelcastInstanceConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/instance/HazelcastInstanceConsumer.java @@ -49,7 +49,7 @@ public class HazelcastInstanceConsumer extends DefaultConsumer { } private void sendExchange(MembershipEvent event, String action) { - Exchange exchange = getEndpoint().createExchange(); + Exchange exchange = createExchange(false); HazelcastComponentHelper.setListenerHeaders(exchange, HazelcastConstants.INSTANCE_LISTENER, action); @@ -71,6 +71,8 @@ public class HazelcastInstanceConsumer extends DefaultConsumer { "Error processing exchange for Hazelcast consumer on your Hazelcast cluster.", exchange, exchange.getException()); } + + releaseExchange(exchange, false); } } diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/list/HazelcastListConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/list/HazelcastListConsumer.java index f22a3bb..6178333 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/list/HazelcastListConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/list/HazelcastListConsumer.java @@ -42,9 +42,6 @@ public class HazelcastListConsumer extends HazelcastDefaultConsumer { queue = hazelcastInstance.getList(cacheName); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStart() - */ @Override protected void doStart() throws Exception { super.doStart(); @@ -52,9 +49,6 @@ public class HazelcastListConsumer extends HazelcastDefaultConsumer { listener = queue.addItemListener(new CamelItemListener(this, cacheName), true); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStop() - */ @Override protected void doStop() throws Exception { queue.removeItemListener(listener); diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/listener/CamelListener.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/listener/CamelListener.java index 9e330be..d7cdffa 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/listener/CamelListener.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/listener/CamelListener.java @@ -32,7 +32,7 @@ public class CamelListener { } protected void sendExchange(String operation, Object key, Object value) { - Exchange exchange = consumer.getEndpoint().createExchange(); + Exchange exchange = consumer.createExchange(false); // set object to body exchange.getIn().setBody(value); @@ -57,6 +57,8 @@ public class CamelListener { exchange, exchange.getException()); } + + consumer.releaseExchange(exchange, false); } public String getCacheName() { diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/map/HazelcastMapConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/map/HazelcastMapConsumer.java index 1e8fbff..2d16262 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/map/HazelcastMapConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/map/HazelcastMapConsumer.java @@ -37,9 +37,6 @@ public class HazelcastMapConsumer extends HazelcastDefaultConsumer { cache = hazelcastInstance.getMap(cacheName); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStart() - */ @Override protected void doStart() throws Exception { super.doStart(); @@ -47,9 +44,6 @@ public class HazelcastMapConsumer extends HazelcastDefaultConsumer { listener = cache.addEntryListener(new CamelMapListener(this, cacheName), true); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStop() - */ @Override protected void doStop() throws Exception { cache.removeEntryListener(listener); diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/multimap/HazelcastMultimapConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/multimap/HazelcastMultimapConsumer.java index 29c0d6a..a506fcb 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/multimap/HazelcastMultimapConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/multimap/HazelcastMultimapConsumer.java @@ -38,9 +38,6 @@ public class HazelcastMultimapConsumer extends HazelcastDefaultConsumer { cache = hazelcastInstance.getMultiMap(cacheName); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStart() - */ @Override protected void doStart() throws Exception { super.doStart(); @@ -48,9 +45,6 @@ public class HazelcastMultimapConsumer extends HazelcastDefaultConsumer { listener = cache.addEntryListener(new CamelEntryListener(this, cacheName), true); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStop() - */ @Override protected void doStop() throws Exception { cache.removeEntryListener(listener); diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueConsumer.java index 7794bda..e177126 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/queue/HazelcastQueueConsumer.java @@ -86,12 +86,14 @@ public class HazelcastQueueConsumer extends HazelcastDefaultConsumer { final Object body = queue.poll(config.getPollingTimeout(), TimeUnit.MILLISECONDS); // CAMEL-16035 - If the polling timeout is exceeded with nothing to poll from the queue, the queue.poll() method return NULL if (body != null) { - Exchange exchange = getEndpoint().createExchange(); + Exchange exchange = createExchange(false); exchange.getIn().setBody(body); try { processor.process(exchange); } catch (Exception e) { getExceptionHandler().handleException("Error during processing", exchange, e); + } finally { + releaseExchange(exchange, false); } } } catch (InterruptedException e) { diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/replicatedmap/HazelcastReplicatedmapConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/replicatedmap/HazelcastReplicatedmapConsumer.java index 2a215fd..380def8 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/replicatedmap/HazelcastReplicatedmapConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/replicatedmap/HazelcastReplicatedmapConsumer.java @@ -38,9 +38,6 @@ public class HazelcastReplicatedmapConsumer extends HazelcastDefaultConsumer { cache = hazelcastInstance.getReplicatedMap(cacheName); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStart() - */ @Override protected void doStart() throws Exception { super.doStart(); @@ -48,9 +45,6 @@ public class HazelcastReplicatedmapConsumer extends HazelcastDefaultConsumer { listener = cache.addEntryListener(new CamelEntryListener(this, cacheName)); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStop() - */ @Override protected void doStop() throws Exception { cache.removeEntryListener(listener); diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java index a73aafe..c58db47 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java @@ -76,7 +76,7 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { BaseQueue<?> queue = endpoint.getHazelcastInstance().getQueue(endpoint.getConfiguration().getQueueName()); while (queue != null && isRunAllowed()) { - final Exchange exchange = this.getEndpoint().createExchange(); + final Exchange exchange = createExchange(true); TransactionContext transactionCtx = null; try { @@ -100,22 +100,25 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { exchange.getIn().setBody(body); } try { + final TransactionContext txc = transactionCtx; // process using the asynchronous routing engine processor.process(exchange, new AsyncCallback() { public void done(boolean asyncDone) { - // noop + if (exchange.getException() != null) { + // Rollback + if (txc != null) { + txc.rollbackTransaction(); + } + getExceptionHandler().handleException("Error processing exchange", exchange, + exchange.getException()); + } + // It's OK, I commit + if (exchange.getException() == null && txc != null) { + LOG.trace("Commit transaction: {}", txc.getTxnId()); + txc.commitTransaction(); + } } }); - - if (exchange.getException() != null) { - // Rollback - if (transactionCtx != null) { - transactionCtx.rollbackTransaction(); - } - getExceptionHandler().handleException("Error processing exchange", exchange, - exchange.getException()); - } - } catch (Exception e) { LOG.error("Hzlq Exception caught: {}", e, e); // Rollback @@ -125,11 +128,6 @@ public class HazelcastSedaConsumer extends DefaultConsumer implements Runnable { } } } - // It's OK, I commit - if (exchange.getException() == null && transactionCtx != null) { - LOG.trace("Commit transaction: {}", transactionCtx.getTxnId()); - transactionCtx.commitTransaction(); - } } catch (InterruptedException e) { if (LOG.isDebugEnabled()) { LOG.debug("Hzlq Consumer Interrupted: {}", e, e); diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/set/HazelcastSetConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/set/HazelcastSetConsumer.java index 8a17cf9..4cf75e3 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/set/HazelcastSetConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/set/HazelcastSetConsumer.java @@ -40,9 +40,6 @@ public class HazelcastSetConsumer extends HazelcastDefaultConsumer { set = hazelcastInstance.getSet(cacheName); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStart() - */ @Override protected void doStart() throws Exception { super.doStart(); @@ -50,9 +47,6 @@ public class HazelcastSetConsumer extends HazelcastDefaultConsumer { listener = set.addItemListener(new CamelItemListener(this, cacheName), true); } - /** - * @see org.apache.camel.support.DefaultConsumer#doStop() - */ @Override protected void doStop() throws Exception { set.removeItemListener(listener); diff --git a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicConsumer.java b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicConsumer.java index 1c816c6..67a1a2a 100644 --- a/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicConsumer.java +++ b/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/topic/HazelcastTopicConsumer.java @@ -25,9 +25,6 @@ import org.apache.camel.Processor; import org.apache.camel.component.hazelcast.HazelcastDefaultConsumer; import org.apache.camel.component.hazelcast.listener.CamelMessageListener; -/** - * - */ public class HazelcastTopicConsumer extends HazelcastDefaultConsumer { private ITopic<Object> topic; @@ -44,9 +41,6 @@ public class HazelcastTopicConsumer extends HazelcastDefaultConsumer { } } - /** - * @see org.apache.camel.support.DefaultConsumer#doStart() - */ @Override protected void doStart() throws Exception { super.doStart(); diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConsumer.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConsumer.java index 59efd5a..d2c3867 100644 --- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConsumer.java +++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConsumer.java @@ -29,6 +29,7 @@ import org.apache.camel.component.hbase.mapping.CellMappingStrategyFactory; import org.apache.camel.component.hbase.model.HBaseCell; import org.apache.camel.component.hbase.model.HBaseData; import org.apache.camel.component.hbase.model.HBaseRow; +import org.apache.camel.support.DefaultExchange; import org.apache.camel.support.ScheduledBatchPollingConsumer; import org.apache.camel.util.CastUtils; import org.apache.camel.util.ObjectHelper; @@ -94,7 +95,7 @@ public class HBaseConsumer extends ScheduledBatchPollingConsumer { ResultScanner scanner = table.getScanner(scan); int exchangeCount = 0; // The next three statements are used just to get a reference to the BodyCellMappingStrategy instance. - Exchange exchange = endpoint.createExchange(); + Exchange exchange = new DefaultExchange(endpoint); exchange.getIn().setHeader(CellMappingStrategyFactory.STRATEGY, CellMappingStrategyFactory.BODY); CellMappingStrategy mappingStrategy = endpoint.getCellMappingStrategyFactory().getStrategy(exchange.getIn()); for (Result result = scanner.next(); @@ -137,7 +138,7 @@ public class HBaseConsumer extends ScheduledBatchPollingConsumer { } data.getRows().add(resultRow); - exchange = endpoint.createExchange(); + exchange = createExchange(true); // Probably overkill but kept it here for consistency. exchange.getIn().setHeader(CellMappingStrategyFactory.STRATEGY, CellMappingStrategyFactory.BODY); mappingStrategy.applyScanResults(exchange.getIn(), data); diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java index e25ac4e..7e197a4 100644 --- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java +++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java @@ -179,36 +179,40 @@ public final class HdfsConsumer extends ScheduledPollConsumer { private void processHdfsInputStream( HdfsInputStream hdfsFile, Holder<Object> key, Holder<Object> value, AtomicInteger messageCount, AtomicInteger totalMessageCount) { - Exchange exchange = this.getEndpoint().createExchange(); - Message message = exchange.getIn(); - String fileName = StringUtils.substringAfterLast(hdfsFile.getActualPath(), "/"); - message.setHeader(Exchange.FILE_NAME, fileName); - message.setHeader(Exchange.FILE_NAME_CONSUMED, fileName); - message.setHeader("CamelFileAbsolutePath", hdfsFile.getActualPath()); - if (key.getValue() != null) { - message.setHeader(HdfsHeader.KEY.name(), key.getValue()); - } + Exchange exchange = createExchange(false); + try { + Message message = exchange.getIn(); + String fileName = StringUtils.substringAfterLast(hdfsFile.getActualPath(), "/"); + message.setHeader(Exchange.FILE_NAME, fileName); + message.setHeader(Exchange.FILE_NAME_CONSUMED, fileName); + message.setHeader("CamelFileAbsolutePath", hdfsFile.getActualPath()); + if (key.getValue() != null) { + message.setHeader(HdfsHeader.KEY.name(), key.getValue()); + } - if (hdfsFile.getNumOfReadBytes() >= 0) { - message.setHeader(Exchange.FILE_LENGTH, hdfsFile.getNumOfReadBytes()); - } + if (hdfsFile.getNumOfReadBytes() >= 0) { + message.setHeader(Exchange.FILE_LENGTH, hdfsFile.getNumOfReadBytes()); + } - message.setBody(value.getValue()); + message.setBody(value.getValue()); - updateNewExchange(exchange, messageCount.get(), hdfsFile); + updateNewExchange(exchange, messageCount.get(), hdfsFile); + + LOG.debug("Processing file [{}]", fileName); - LOG.debug("Processing file [{}]", fileName); - try { processor.process(exchange); totalMessageCount.incrementAndGet(); + } catch (Exception e) { exchange.setException(e); + } finally { + // in case of unhandled exceptions then let the exception handler handle them + if (exchange.getException() != null) { + getExceptionHandler().handleException(exchange.getException()); + } + releaseExchange(exchange, false); } - // in case of unhandled exceptions then let the exception handler handle them - if (exchange.getException() != null) { - getExceptionHandler().handleException(exchange.getException()); - } } private boolean normalFileIsDirectoryHasSuccessFile(FileStatus fileStatus, HdfsInfo info) {