This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new d0381bafb261 CAMEL-26881: camel-core - Aggregate EIP: an Error of the
aggregation strategy or repository no longer stops the completion timeout,
interval or recover task (#29785)
d0381bafb261 is described below
commit d0381bafb2615b1e25ebb98d6bbfa16a3432fff7
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 20:23:02 2026 +0530
CAMEL-26881: camel-core - Aggregate EIP: an Error of the aggregation
strategy or repository no longer stops the completion timeout, interval or
recover task (#29785)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/aggregate/AggregateProcessor.java | 42 +++++++++---
.../AggregateIntervalTaskExceptionTest.java | 28 +++++++-
.../AggregateOnCompletionThrowsTest.java | 40 +++++++++--
.../AggregateTimeoutRepositoryExceptionTest.java | 20 ++++++
.../AggregateTimeoutStrategyThrowsErrorTest.java | 78 ++++++++++++++++++++++
.../DefaultTimeoutMapListenerThrowsErrorTest.java | 73 ++++++++++++++++++++
.../apache/camel/support/DefaultTimeoutMap.java | 17 ++++-
7 files changed, 276 insertions(+), 22 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
index f257f354427b..11b89f86f5a5 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
@@ -1105,22 +1105,28 @@ public class AggregateProcessor extends
BaseProcessorSupport
/**
* Sends the aggregated exchanges (more than one when the batch consumer
completes all its groups). None of them is
* in the repository any more, so when one of them fails to be submitted
(the thread pool rejects it, or the
- * aggregation strategy throws an exception on completion), the others are
still sent (or failed too), and then the
- * first failure is thrown.
+ * aggregation strategy throws an exception or an Error on completion),
the others are still sent (or failed too),
+ * and then the first failure is thrown.
*/
private void submitCompletions(String key, List<Exchange> aggregated) {
- RuntimeException failure = null;
+ Throwable failure = null;
for (Exchange agg : aggregated) {
try {
onSubmitCompletion(key, agg);
- } catch (RuntimeException e) {
+ } catch (VirtualMachineError e) {
+ // left as before: a JVM-level failure is not handled here
+ throw e;
+ } catch (RuntimeException | Error e) {
+ // also an Error of the aggregation strategy, as for an
exception
if (failure == null) {
failure = e;
}
}
}
- if (failure != null) {
- throw failure;
+ if (failure instanceof Error err) {
+ throw err;
+ } else if (failure != null) {
+ throw (RuntimeException) failure;
}
}
@@ -1669,7 +1675,11 @@ public class AggregateProcessor extends
BaseProcessorSupport
try {
doOnEviction(key, exchangeId);
forgetRetry(key);
- } catch (Exception e) {
+ } catch (VirtualMachineError e) {
+ // left as before: a JVM-level failure is not handled here
+ throw e;
+ } catch (Throwable e) {
+ // also an Error of the aggregation strategy (its timeout or
onCompletion), as for an exception
// the entry is no longer in the timeout map, and the timeout
map ignores errors of its listeners, so
// without a new entry a group the repository could not read
or remove would never time out (an entry
// that a new message for the key has put in the meantime is
kept)
@@ -1788,7 +1798,11 @@ public class AggregateProcessor extends
BaseProcessorSupport
lock.unlock();
}
}
- } catch (Exception e) {
+ } catch (VirtualMachineError e) {
+ // left as before: a JVM-level failure is not handled, and
ends the interval task
+ throw e;
+ } catch (Throwable e) {
+ // also an Error (such as of the aggregation strategy), which
would otherwise end the interval task
LOG.warn("Error during completion interval task. This
exception is ignored.", e);
}
@@ -1912,7 +1926,10 @@ public class AggregateProcessor extends
BaseProcessorSupport
// not exhaust so resubmit the recovered
exchange
try {
onSubmitCompletion(key, exchange);
- } catch (RuntimeException e) {
+ } catch (VirtualMachineError e) {
+ // left as before: a JVM-level failure
is not handled, and ends the recover task
+ throw e;
+ } catch (RuntimeException | Error e) {
// it has been failed and passed to
the exception handler, and is recovered
// again on a later run, so go on with
the other exchanges
LOG.debug("Error resubmitting
recovered exchange with id: {}", exchangeId, e);
@@ -1924,8 +1941,11 @@ public class AggregateProcessor extends
BaseProcessorSupport
recoverLock.unlock();
}
}
- } catch (Exception e) {
- // must catch and log exception otherwise the executor will
not schedule next recover task
+ } catch (VirtualMachineError e) {
+ // left as before: a JVM-level failure is not handled, and
ends the recover task
+ throw e;
+ } catch (Throwable e) {
+ // must catch and log exception (also an Error) otherwise the
executor will not schedule next recover task
LOG.warn("Error during recover task. This exception is
ignored.", e);
} finally {
recoveryInProgress.set(false);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
index 67d3abac8b43..fdf309b8a90d 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
@@ -51,6 +51,18 @@ public class AggregateIntervalTaskExceptionTest extends
ContextTestSupport {
MockEndpoint.assertIsSatisfied(context, 15, TimeUnit.SECONDS);
}
+ @Test
+ public void testIntervalTaskSurvivesError() throws Exception {
+ getMockEndpoint("mock:error").expectedMinimumMessageCount(1);
+
+ template.sendBodyAndHeader("direct:error", "A", "id", "1");
+ template.sendBodyAndHeader("direct:error", "B", "id", "1");
+
+ // the first interval tick with data gets an Error (not a
VirtualMachineError) from getKeys(), and the
+ // scheduler survives it too
+ MockEndpoint.assertIsSatisfied(context, 15, TimeUnit.SECONDS);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -58,9 +70,15 @@ public class AggregateIntervalTaskExceptionTest extends
ContextTestSupport {
public void configure() {
from("direct:start")
.aggregate(header("id"), new
UseLatestAggregationStrategy())
- .aggregationRepository(new FailOnceRepository())
+ .aggregationRepository(new FailOnceRepository(false))
.completionInterval(500)
.to("mock:result");
+
+ from("direct:error")
+ .aggregate(header("id"), new
UseLatestAggregationStrategy())
+ .aggregationRepository(new FailOnceRepository(true))
+ .completionInterval(500)
+ .to("mock:error");
}
};
}
@@ -69,6 +87,11 @@ public class AggregateIntervalTaskExceptionTest extends
ContextTestSupport {
private final MemoryAggregationRepository delegate = new
MemoryAggregationRepository();
private final AtomicInteger getKeysCount = new AtomicInteger();
+ private final boolean error;
+
+ FailOnceRepository(boolean error) {
+ this.error = error;
+ }
@Override
public Exchange add(CamelContext camelContext, String key, Exchange
exchange) {
@@ -94,6 +117,9 @@ public class AggregateIntervalTaskExceptionTest extends
ContextTestSupport {
public Set<String> getKeys() {
Set<String> keys = delegate.getKeys();
if (keys != null && !keys.isEmpty() &&
getKeysCount.incrementAndGet() == 1) {
+ if (error) {
+ throw new NoClassDefFoundError("Simulated missing class");
+ }
throw new RuntimeException("Simulated repository failure");
}
return keys;
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateOnCompletionThrowsTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateOnCompletionThrowsTest.java
index 9b5943d132f0..df94ff2665d7 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateOnCompletionThrowsTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateOnCompletionThrowsTest.java
@@ -38,24 +38,34 @@ import
org.apache.camel.support.KeyValueAggregationRepository;
import org.apache.camel.support.MemoryKeyValueRepository;
import org.apache.camel.support.SynchronizationAdapter;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* An aggregation strategy whose onCompletion throws (its javadoc says it must
not): the aggregated exchange is booked
* as in progress already, and its group has been removed from the repository.
The aggregated exchange must then be
* failed, as when the thread pool rejects it, so that the aggregator no
longer counts it as in progress and a
- * recoverable repository recovers it, instead of skipping it forever as being
in progress.
+ * recoverable repository recovers it, instead of skipping it forever as being
in progress. An Error that is not a
+ * VirtualMachineError (such as a NoClassDefFoundError of a class used by the
strategy) is handled as an exception.
*/
class AggregateOnCompletionThrowsTest extends ContextTestSupport {
+ enum Failure {
+ EXCEPTION,
+ ERROR
+ }
+
private final AtomicBoolean throwOnce = new AtomicBoolean(true);
private final Set<String> alwaysThrow = ConcurrentHashMap.newKeySet();
private final List<String> failures = new CopyOnWriteArrayList<>();
private final List<Exchange> handled = new CopyOnWriteArrayList<>();
+ private Failure failure = Failure.EXCEPTION;
@Override
public boolean isUseRouteBuilder() {
@@ -105,8 +115,10 @@ class AggregateOnCompletionThrowsTest extends
ContextTestSupport {
assertMockEndpointsSatisfied();
}
- @Test
- void testFailedBatchCompletionSendsOtherGroups() throws Exception {
+ @ParameterizedTest
+ @EnumSource(Failure.class)
+ void testFailedBatchCompletionSendsOtherGroups(Failure failure) throws
Exception {
+ this.failure = failure;
AggregateProcessor aggregator = addRoute(recoverableRepository(),
true);
// the batch consumer completes the groups A and B at once: whichever
group's onCompletion throws is
@@ -118,15 +130,17 @@ class AggregateOnCompletionThrowsTest extends
ContextTestSupport {
sendBatch("A1", "A");
sendBatch("B1", "B");
Exchange last = sendBatch("A2", "A");
- assertInstanceOf(IllegalStateException.class, last.getException());
+ assertFailedWith(failure, last);
assertMockEndpointsSatisfied();
await().atMost(5, TimeUnit.SECONDS)
.untilAsserted(() -> assertEquals(0,
aggregator.getInProgressCompleteExchanges()));
}
- @Test
- void testRecoverGoesOnAfterAnExchangeThatFailsAgain() throws Exception {
+ @ParameterizedTest
+ @EnumSource(Failure.class)
+ void testRecoverGoesOnAfterAnExchangeThatFailsAgain(Failure failure)
throws Exception {
+ this.failure = failure;
addRoute(new OrderedRecoverableRepository(), false);
// onCompletion of A1+A2 throws each time, so it fails again each time
the recover task resubmits it, and it is
// the first one the recover task finds: the recover task must still
go on and recover B1+B2
@@ -141,13 +155,22 @@ class AggregateOnCompletionThrowsTest extends
ContextTestSupport {
// onCompletion of B1+B2 throws once (throwOnce), so it is in the
completed store too
send("B1", "B");
Exchange last = send("B2", "B");
- assertInstanceOf(IllegalStateException.class, last.getException());
+ assertFailedWith(failure, last);
assertMockEndpointsSatisfied();
// and A1+A2 was recovered and failed again
assertTrue(failures.stream().filter("A1+A2"::equals).count() >= 2);
}
+ private static void assertFailedWith(Failure failure, Exchange exchange) {
+ if (failure == Failure.ERROR) {
+ // the message that completed the group fails with the Error, as
before
+ assertNotNull(exchange.getException(NoClassDefFoundError.class));
+ } else {
+ assertInstanceOf(IllegalStateException.class,
exchange.getException());
+ }
+ }
+
private static KeyValueAggregationRepository recoverableRepository() {
KeyValueAggregationRepository repository = new
KeyValueAggregationRepository(new MemoryKeyValueRepository());
repository.setUseRecovery(true);
@@ -184,6 +207,9 @@ class AggregateOnCompletionThrowsTest extends
ContextTestSupport {
}
});
if (alwaysThrow.contains(body) ||
throwOnce.compareAndSet(true, false)) {
+ if (failure == Failure.ERROR) {
+ throw new NoClassDefFoundError("onCompletion failed
for " + body);
+ }
throw new IllegalStateException("onCompletion failed for "
+ body);
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
index 7bdc3ca2dfaf..90a61be096ed 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
@@ -104,6 +104,21 @@ public class AggregateTimeoutRepositoryExceptionTest
extends ContextTestSupport
assertEquals(1, repository.failures.get(), "the repository should have
failed once");
}
+ @Test
+ public void testTimeoutCompletesGroupAfterFailedGetWithError() throws
Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("B");
+
+ template.sendBodyAndHeader("direct:start", "A", "id", "1");
+ template.sendBodyAndHeader("direct:start", "B", "id", "1");
+ // the next read of the group is the one of the timeout checker, and
it throws an Error (such as a missing class
+ // of a stored body): it is tried again, as for an exception
+ repository.failNextGetWithError.set(true);
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ assertEquals(1, repository.failures.get(), "the repository should have
failed once");
+ }
+
@Test
public void testTimeoutCompletesGroupAfterFailedRemove() throws Exception {
MockEndpoint result = getMockEndpoint("mock:result");
@@ -136,6 +151,7 @@ public class AggregateTimeoutRepositoryExceptionTest
extends ContextTestSupport
private final MemoryAggregationRepository delegate = new
MemoryAggregationRepository();
private final AtomicBoolean failNextGet = new AtomicBoolean();
+ private final AtomicBoolean failNextGetWithError = new AtomicBoolean();
private final AtomicBoolean failNextRemove = new AtomicBoolean();
private final AtomicInteger failingGets = new AtomicInteger();
private final AtomicInteger failures = new AtomicInteger();
@@ -147,6 +163,10 @@ public class AggregateTimeoutRepositoryExceptionTest
extends ContextTestSupport
@Override
public Exchange get(CamelContext camelContext, String key) {
+ if (failNextGetWithError.compareAndSet(true, false)) {
+ failures.incrementAndGet();
+ throw new NoClassDefFoundError("Simulated missing class");
+ }
if (failNextGet.compareAndSet(true, false) ||
failingGets.getAndUpdate(n -> Math.max(0, n - 1)) > 0) {
failures.incrementAndGet();
throw new IllegalStateException("Simulated repository failure
in get");
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutStrategyThrowsErrorTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutStrategyThrowsErrorTest.java
new file mode 100644
index 000000000000..97c4a6b71b30
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutStrategyThrowsErrorTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.processor.aggregator;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * The timeout of the aggregation strategy throws an Error (not a
VirtualMachineError) when a group completes by its
+ * completion timeout: the other groups still complete by their timeout.
+ */
+class AggregateTimeoutStrategyThrowsErrorTest extends ContextTestSupport {
+
+ private final AtomicInteger timeouts = new AtomicInteger();
+
+ @Test
+ void testOtherGroupsStillCompleteByTimeout() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ // A is not delivered, as for an exception of timeout (the group was
removed from the repository before)
+ mock.expectedBodiesReceived("B");
+
+ template.sendBodyAndHeader("direct:start", "A", "id", "A");
+ // the timeout of A has thrown the Error
+ await().atMost(5, TimeUnit.SECONDS).until(() -> timeouts.get() >= 1);
+ template.sendBodyAndHeader("direct:start", "B", "id", "B");
+
+ // the timeout checker still runs: B completes by its timeout
+ MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .aggregate(header("id"), new AggregationStrategy() {
+ @Override
+ public Exchange aggregate(Exchange oldExchange,
Exchange newExchange) {
+ return oldExchange == null ? newExchange :
oldExchange;
+ }
+
+ @Override
+ public void timeout(Exchange exchange, int index,
int total, long timeout) {
+ if (timeouts.incrementAndGet() == 1) {
+ throw new NoClassDefFoundError("Simulated
missing class");
+ }
+ }
+ })
+
.completionTimeout(50).completionTimeoutCheckerInterval(10)
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/support/DefaultTimeoutMapListenerThrowsErrorTest.java
b/core/camel-core/src/test/java/org/apache/camel/support/DefaultTimeoutMapListenerThrowsErrorTest.java
new file mode 100644
index 000000000000..eb76f4776eb0
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/support/DefaultTimeoutMapListenerThrowsErrorTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.support;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import org.apache.camel.TimeoutMap;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A listener of a DefaultTimeoutMap throws an Error (not a
VirtualMachineError) for an evicted entry: the other entries
+ * are still evicted, and the map keeps purging.
+ */
+class DefaultTimeoutMapListenerThrowsErrorTest {
+
+ private final ScheduledExecutorService executor = new
ScheduledThreadPoolExecutor(1);
+
+ @AfterEach
+ void shutdown() {
+ executor.shutdownNow();
+ }
+
+ @Test
+ void testPurgeGoesOnAfterAListenerThrowsAnError() {
+ DefaultTimeoutMap<String, Integer> map = new
DefaultTimeoutMap<>(executor, 10);
+ List<String> evicted = new CopyOnWriteArrayList<>();
+ map.addListener((type, key, value) -> {
+ if (type == TimeoutMap.Listener.Type.Evict) {
+ evicted.add(key);
+ if ("A".equals(key)) {
+ throw new NoClassDefFoundError("Simulated missing class");
+ }
+ }
+ });
+ map.start();
+ try {
+ // A and B expire in the same purge: B is still evicted after the
Error of A
+ map.put("A", 1, 20);
+ map.put("B", 2, 20);
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
assertTrue(evicted.containsAll(List.of("A", "B"))));
+
+ // and the map keeps purging: an entry added later is evicted too
+ map.put("C", 3, 20);
+ await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
assertTrue(evicted.contains("C")));
+ assertEquals(0, map.size());
+ } finally {
+ map.stop();
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
index 3e3ac2fe785b..23310c9bbb6a 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultTimeoutMap.java
@@ -171,9 +171,13 @@ public class DefaultTimeoutMap<K, V> extends
ServiceSupport implements TimeoutMa
log.trace("Running purge task to see if any entries have been timed
out");
try {
purge();
- } catch (Exception t) {
- // must catch and log exception otherwise the executor will now
schedule next purgeTask
- log.warn("Exception occurred during purge task. This exception
will be ignored.", t);
+ } catch (VirtualMachineError t) {
+ // left as before: a JVM-level failure still ends the purge task
+ throw t;
+ } catch (Throwable t) {
+ // must catch and log exception (also an Error) otherwise the
executor will not schedule the next purgeTask,
+ // and no entry would ever time out again
+ log.warn("Exception or error occurred during purge task. This
exception or error will be ignored.", t);
}
}
@@ -243,6 +247,13 @@ public class DefaultTimeoutMap<K, V> extends
ServiceSupport implements TimeoutMa
for (Listener<K, V> listener : listeners) {
try {
listener.timeoutMapEvent(type, key, value);
+ } catch (VirtualMachineError t) {
+ // left as before: a JVM-level failure is thrown on (and ends
the purge task when it purges)
+ throw t;
+ } catch (Error t) {
+ // the other entries that expired with this one are still
evicted
+ log.warn("Error in a listener of the timeout map for key: {}
({} event). This error is ignored.", key, type,
+ t);
} catch (Exception t) {
// Ignore
}
