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 2021/03/15 08:39:33 UTC
[camel] branch master updated (52b1a42 -> c7f45b8)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 52b1a42 Regen for commit ff7a92949252a92a03baa7b87e497cf744ae052e
new 98d7575 Polished camel-main waiting message
new c7f45b8 camel-core - Fixed pooled leak in redelivery error handler.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/java/org/apache/camel/processor/Pipeline.java | 17 +----------------
.../org/apache/camel/processor/PooledTaskFactory.java | 13 ++++++++++++-
.../processor/errorhandler/RedeliveryErrorHandler.java | 8 ++++++--
.../main/java/org/apache/camel/main/MainSupport.java | 8 ++++----
.../camel/support/PooledObjectFactorySupport.java | 9 +++++++++
5 files changed, 32 insertions(+), 23 deletions(-)
[camel] 01/02: Polished camel-main waiting message
Posted by da...@apache.org.
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 98d75757a3248e589e5acfda5be1a53188758e45
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Mar 15 09:21:54 2021 +0100
Polished camel-main waiting message
---
.../src/main/java/org/apache/camel/main/MainSupport.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/MainSupport.java b/core/camel-main/src/main/java/org/apache/camel/main/MainSupport.java
index ee4618e..d49d760 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/MainSupport.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/MainSupport.java
@@ -293,17 +293,17 @@ public abstract class MainSupport extends BaseMainSupport {
long sec = durationMaxSeconds;
int exit = durationHitExitCode;
if (sec > 0) {
- LOG.info("Waiting for: {} seconds", sec);
+ LOG.info("Waiting until complete: Duration max {} seconds", sec);
shutdownStrategy.await(sec, TimeUnit.SECONDS);
exitCode.compareAndSet(UNINITIALIZED_EXIT_CODE, exit);
shutdownStrategy.shutdown();
} else if (idle > 0 || max > 0) {
if (idle > 0 && max > 0) {
- LOG.info("Waiting to be idle for: {} seconds or until: {} messages has been processed", idle, max);
+ LOG.info("Waiting until complete: Duration idle {} seconds or max {} messages processed", idle, max);
} else if (idle > 0) {
- LOG.info("Waiting to be idle for: {} seconds", idle);
+ LOG.info("Waiting until complete: Duration idle {} seconds", idle);
} else {
- LOG.info("Waiting until: {} messages has been processed", max);
+ LOG.info("Waiting until complete: Duration max {} messages processed", max);
}
exitCode.compareAndSet(UNINITIALIZED_EXIT_CODE, exit);
shutdownStrategy.await();
[camel] 02/02: camel-core - Fixed pooled leak in redelivery error
handler.
Posted by da...@apache.org.
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 c7f45b87f73f9d74e9cf7d81dbbd33cf8dab6700
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Mar 15 09:38:59 2021 +0100
camel-core - Fixed pooled leak in redelivery error handler.
---
.../main/java/org/apache/camel/processor/Pipeline.java | 17 +----------------
.../org/apache/camel/processor/PooledTaskFactory.java | 13 ++++++++++++-
.../processor/errorhandler/RedeliveryErrorHandler.java | 8 ++++++--
.../camel/support/PooledObjectFactorySupport.java | 9 +++++++++
4 files changed, 28 insertions(+), 19 deletions(-)
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Pipeline.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Pipeline.java
index 5389003..557afa3 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Pipeline.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Pipeline.java
@@ -179,35 +179,20 @@ public class Pipeline extends AsyncProcessorSupport implements Navigate<Processo
boolean pooled = camelContext.adapt(ExtendedCamelContext.class).getExchangeFactory().isPooled();
if (pooled) {
- taskFactory = new PooledTaskFactory() {
+ taskFactory = new PooledTaskFactory(getId()) {
@Override
public PooledExchangeTask create(Exchange exchange, AsyncCallback callback) {
return new PipelineTask();
}
-
- @Override
- public String toString() {
- return "PooledTaskFactory[capacity: " + getCapacity() + "]";
- }
};
int capacity = camelContext.adapt(ExtendedCamelContext.class).getExchangeFactory().getCapacity();
taskFactory.setCapacity(capacity);
} else {
taskFactory = new PrototypeTaskFactory() {
@Override
- public boolean isPooled() {
- return false;
- }
-
- @Override
public PooledExchangeTask create(Exchange exchange, AsyncCallback callback) {
return new PipelineTask();
}
-
- @Override
- public String toString() {
- return "PrototypeTaskFactory";
- }
};
}
LOG.trace("Using TaskFactory: {}", taskFactory);
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PooledTaskFactory.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PooledTaskFactory.java
index 6cf306a..ac184ab 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/PooledTaskFactory.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/PooledTaskFactory.java
@@ -23,6 +23,13 @@ import org.apache.camel.support.PooledObjectFactorySupport;
public abstract class PooledTaskFactory extends PooledObjectFactorySupport<PooledExchangeTask>
implements PooledExchangeTaskFactory {
+ public PooledTaskFactory() {
+ }
+
+ public PooledTaskFactory(Object source) {
+ super(source);
+ }
+
@Override
public PooledExchangeTask acquire() {
return pool.poll();
@@ -67,6 +74,10 @@ public abstract class PooledTaskFactory extends PooledObjectFactorySupport<Poole
@Override
public String toString() {
- return "PooledTaskFactory[capacity: " + getCapacity() + "]";
+ if (source != null) {
+ return "PooledTaskFactory[source: " + source + ", capacity: " + getCapacity() + "]";
+ } else {
+ return "PooledTaskFactory[capacity: " + getCapacity() + "]";
+ }
}
}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
index 1481748..26f03b8 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/errorhandler/RedeliveryErrorHandler.java
@@ -46,6 +46,7 @@ import org.apache.camel.spi.AsyncProcessorAwaitManager;
import org.apache.camel.spi.CamelLogger;
import org.apache.camel.spi.ErrorHandlerRedeliveryCustomizer;
import org.apache.camel.spi.ExchangeFormatter;
+import org.apache.camel.spi.IdAware;
import org.apache.camel.spi.ReactiveExecutor;
import org.apache.camel.spi.ShutdownPrepared;
import org.apache.camel.spi.ShutdownStrategy;
@@ -779,7 +780,9 @@ public abstract class RedeliveryErrorHandler extends ErrorHandlerSupport
outputAsync.process(exchange, doneSync -> {
// only continue with callback if we are done
if (isDone(exchange)) {
- reactiveExecutor.schedule(callback);
+ AsyncCallback cb = callback;
+ taskFactory.release(this);
+ reactiveExecutor.schedule(cb);
} else {
// error occurred so loop back around and call ourselves
reactiveExecutor.schedule(this);
@@ -1594,7 +1597,8 @@ public abstract class RedeliveryErrorHandler extends ErrorHandlerSupport
boolean pooled = camelContext.adapt(ExtendedCamelContext.class).getExchangeFactory().isPooled();
if (pooled) {
- taskFactory = new PooledTaskFactory() {
+ String id = output instanceof IdAware ? ((IdAware) output).getId() : output.toString();
+ taskFactory = new PooledTaskFactory(id) {
@Override
public PooledExchangeTask create(Exchange exchange, AsyncCallback callback) {
return simpleTask ? new SimpleTask() : new RedeliveryTask();
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/PooledObjectFactorySupport.java b/core/camel-support/src/main/java/org/apache/camel/support/PooledObjectFactorySupport.java
index 8568832..26d2f17 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/PooledObjectFactorySupport.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/PooledObjectFactorySupport.java
@@ -29,12 +29,21 @@ import org.apache.camel.support.service.ServiceSupport;
*/
public abstract class PooledObjectFactorySupport<T> extends ServiceSupport implements PooledObjectFactory<T> {
+ protected final Object source;
protected UtilizationStatistics statistics;
protected CamelContext camelContext;
protected BlockingQueue<T> pool;
protected int capacity = 100;
protected boolean statisticsEnabled;
+ public PooledObjectFactorySupport() {
+ this.source = null;
+ }
+
+ public PooledObjectFactorySupport(Object source) {
+ this.source = source;
+ }
+
@Override
protected void doBuild() throws Exception {
super.doBuild();