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 47ed10257809 CAMEL-24127: Fix flaky tests (batch 15)
47ed10257809 is described below
commit 47ed10257809b2fa763c09322531895a818ba6df
Author: Guillaume Nodet <[email protected]>
AuthorDate: Sat Jul 18 14:43:18 2026 +0200
CAMEL-24127: Fix flaky tests (batch 15)
Fix 5 flaky tests: ElasticSearch disk watermark 429s,
LoopNoBreakOnShutdownTest
SEDA shutdown race, HazelcastReplicatedmapConsumerTest leaked REMOVED
events,
LRAFailuresIT periodic recovery timing, and TimerRouteAutoConfigIT
timing-dependent
metric collection. Also fixes WaitAllStrategy bug in
MicroprofileLRALocalContainerInfraService.
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../HazelcastReplicatedmapConsumerTest.java | 15 ++--
.../camel/service/lra/AbstractLRATestSupport.java | 30 +++++++-
.../apache/camel/service/lra/LRAFailuresIT.java | 14 +++-
.../integration/TimerRouteAutoConfigIT.java | 82 ++++++++++------------
.../camel/processor/LoopNoBreakOnShutdownTest.java | 14 ++--
.../ElasticSearchLocalContainerInfraService.java | 4 ++
.../MicroprofileLRALocalContainerInfraService.java | 17 +++--
7 files changed, 112 insertions(+), 64 deletions(-)
diff --git
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastReplicatedmapConsumerTest.java
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastReplicatedmapConsumerTest.java
index 140cfd903943..08055ff37cbf 100644
---
a/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastReplicatedmapConsumerTest.java
+++
b/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastReplicatedmapConsumerTest.java
@@ -102,17 +102,22 @@ public class HazelcastReplicatedmapConsumerTest extends
CamelTestSupport {
@Test
public void testRemove() throws InterruptedException {
- MockEndpoint out = getMockEndpoint("mock:removed");
- out.expectedMessageCount(1);
-
map.put("4711", "my-foo");
- // Wait for the ADDED event to be fully processed before removing,
- // otherwise the remove may execute before the async ADDED event is
delivered.
+ // Wait for the ADDED event to be fully processed. Since events for
+ // key "4711" are dispatched in partition order within Hazelcast's
+ // event system, any stale REMOVED events from resetState()'s clear()
+ // are guaranteed to have been delivered by the time we see the ADDED.
MockEndpoint added = getMockEndpoint("mock:added");
Awaitility.await().atMost(10, TimeUnit.SECONDS)
.until(() -> added.getReceivedCounter() >= 1);
+ // Reset mock:removed to discard any stale REMOVED events that leaked
+ // from clear(), then set the expectation for the test's own remove.
+ MockEndpoint out = getMockEndpoint("mock:removed");
+ out.reset();
+ out.expectedMessageCount(1);
+
map.remove("4711");
MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
this.checkHeaders(out.getExchanges().get(0).getIn().getHeaders(),
HazelcastConstants.REMOVED);
diff --git
a/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
b/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
index 3bc05e5cd341..ea8b73b14726 100644
---
a/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
+++
b/components/camel-lra/src/test/java/org/apache/camel/service/lra/AbstractLRATestSupport.java
@@ -36,7 +36,6 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
-import static org.hamcrest.Matchers.equalTo;
/**
* Base class for LRA based tests.
@@ -58,9 +57,34 @@ public abstract class AbstractLRATestSupport extends
CamelTestSupport {
@AfterEach
public void checkActiveLRAs() throws IOException, InterruptedException {
- await().atMost(20, SECONDS)
+ // After a test that exercises LRA recovery (e.g., LRAFailuresIT), the
+ // coordinator may need a recovery cycle to close the LRA. Trigger
+ // recovery explicitly rather than waiting for the periodic scan
+ // (default period: 120s).
+ await().atMost(60, SECONDS)
+ .pollInterval(2, SECONDS)
+ .pollDelay(1, SECONDS)
.alias("Some LRA have been left pending")
- .until(() -> getNumberOfActiveLRAs(), equalTo(activeLRAs));
+ .until(() -> {
+ triggerRecovery();
+ return getNumberOfActiveLRAs() == activeLRAs;
+ });
+ }
+
+ /**
+ * Triggers a recovery scan on the LRA coordinator via its REST endpoint.
+ */
+ protected void triggerRecovery() {
+ try {
+ HttpClient client = HttpClient.newHttpClient();
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(URI.create(service.getServiceAddress() +
"/lra-coordinator/recovery"))
+ .GET()
+ .build();
+ client.send(request, HttpResponse.BodyHandlers.ofString());
+ } catch (Exception e) {
+ // Best-effort — the periodic recovery manager will eventually run
+ }
}
@Override
diff --git
a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
index 5a898103f6ba..a6f70775a99a 100644
---
a/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
+++
b/components/camel-lra/src/test/java/org/apache/camel/service/lra/LRAFailuresIT.java
@@ -57,9 +57,19 @@ public class LRAFailuresIT extends AbstractLRATestSupport {
TestSupport.sendBody(template, "direct:saga-complete", "hello");
+ // The Narayana LRA coordinator retries failed completion callbacks via
+ // its periodic recovery manager (default: every 120s). Rather than
wait
+ // for the next periodic scan, explicitly trigger recovery via the
+ // coordinator's REST endpoint. Poll because the completion failure may
+ // not yet be recorded when the first trigger fires.
await().atMost(60, TimeUnit.SECONDS)
- .until(() -> complete.getReceivedCounter() >= 1
- && end.getReceivedCounter() >= 1);
+ .pollInterval(2, TimeUnit.SECONDS)
+ .pollDelay(1, TimeUnit.SECONDS)
+ .until(() -> {
+ triggerRecovery();
+ return complete.getReceivedCounter() >= 1
+ && end.getReceivedCounter() >= 1;
+ });
complete.assertIsSatisfied();
end.assertIsSatisfied();
}
diff --git
a/components/camel-opentelemetry-metrics/src/test/java/org/apache/camel/opentelemetry/metrics/integration/TimerRouteAutoConfigIT.java
b/components/camel-opentelemetry-metrics/src/test/java/org/apache/camel/opentelemetry/metrics/integration/TimerRouteAutoConfigIT.java
index 89ebd49569fd..62cd12144014 100644
---
a/components/camel-opentelemetry-metrics/src/test/java/org/apache/camel/opentelemetry/metrics/integration/TimerRouteAutoConfigIT.java
+++
b/components/camel-opentelemetry-metrics/src/test/java/org/apache/camel/opentelemetry/metrics/integration/TimerRouteAutoConfigIT.java
@@ -16,19 +16,16 @@
*/
package org.apache.camel.opentelemetry.metrics.integration;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
-import java.util.Objects;
-import java.util.logging.LogRecord;
-import java.util.logging.Logger;
+import java.util.Map;
import io.opentelemetry.api.GlobalOpenTelemetry;
-import io.opentelemetry.exporter.logging.LoggingMetricExporter;
+import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
import io.opentelemetry.sdk.metrics.data.HistogramPointData;
import io.opentelemetry.sdk.metrics.data.MetricData;
import io.opentelemetry.sdk.metrics.data.PointData;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
import org.apache.camel.CamelContext;
import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.RouteBuilder;
@@ -39,7 +36,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
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.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -47,22 +43,31 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Test class for OpenTelemetry Timer metric autoconfiguration in a Camel
route.
+ *
+ * Uses InMemoryMetricReader for deterministic, synchronous metric collection
instead of relying on the
+ * PeriodicMetricReader + LoggingMetricExporter + JUL log capture chain, which
is inherently timing-dependent and flaky.
*/
public class TimerRouteAutoConfigIT extends CamelTestSupport {
private static final long DELAY = 20L;
+ private static InMemoryMetricReader metricReader;
+
@BeforeAll
public static void init() {
GlobalOpenTelemetry.resetForTest();
- // Open telemetry autoconfiguration using an exporter that writes to
the console via logging.
- // Other possible exporters include 'logging-otlp' and 'otlp'.
- System.setProperty("otel.java.global-autoconfigure.enabled", "true");
- System.setProperty("otel.metrics.exporter", "console");
- System.setProperty("otel.traces.exporter", "none");
- System.setProperty("otel.logs.exporter", "none");
- System.setProperty("otel.propagators", "tracecontext");
- System.setProperty("otel.metric.export.interval", "300");
+ metricReader = InMemoryMetricReader.create();
+ // Still use OTel autoconfigure (the "AutoConfig" in the test name)
but with
+ // InMemoryMetricReader instead of the periodic LoggingMetricExporter.
+ AutoConfiguredOpenTelemetrySdk.builder()
+ .addPropertiesSupplier(() -> Map.of(
+ "otel.metrics.exporter", "none",
+ "otel.traces.exporter", "none",
+ "otel.logs.exporter", "none",
+ "otel.propagators", "tracecontext"))
+ .addMeterProviderCustomizer((builder, config) ->
builder.registerMetricReader(metricReader))
+ .setResultAsGlobal()
+ .build();
}
@AfterEach
@@ -82,40 +87,29 @@ public class TimerRouteAutoConfigIT extends
CamelTestSupport {
@Test
public void testOverrideMetricsName() throws Exception {
- Logger logger =
Logger.getLogger(LoggingMetricExporter.class.getName());
- MemoryLogHandler handler = new MemoryLogHandler();
- logger.addHandler(handler);
-
Object body = new Object();
MockEndpoint mockEndpoint = getMockEndpoint("mock:out");
mockEndpoint.expectedBodiesReceived(body);
template.sendBody("direct:in1", body);
- await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
- List<LogRecord> logs = new ArrayList<>(handler.getLogs());
- assertFalse(logs.isEmpty(), "No metrics were exported");
+ // Collect metrics synchronously -- no timing dependency on periodic
export
+ Collection<MetricData> allMetrics = metricReader.collectAllMetrics();
+ List<MetricData> aMetrics = allMetrics.stream()
+ .filter(md -> "A".equals(md.getName()))
+ .toList();
+ assertFalse(aMetrics.isEmpty(), "No metric data found with name A");
+
+ MetricData md = aMetrics.get(0);
+ PointData pd = md.getData()
+ .getPoints()
+ .stream()
+ .findFirst()
+ .orElseThrow();
+ assertInstanceOf(HistogramPointData.class, pd, "Expected
HistogramPointData");
+ HistogramPointData hpd = (HistogramPointData) pd;
+ assertEquals(1L, hpd.getCount());
+ assertTrue(hpd.getMin() >= DELAY);
- long dataCount = logs.stream()
- .map(LogRecord::getParameters)
- .filter(Objects::nonNull)
- .flatMap(Arrays::stream)
- .filter(MetricData.class::isInstance)
- .map(MetricData.class::cast)
- .filter(md -> "A".equals(md.getName()))
- .peek(md -> {
- PointData pd = md.getData()
- .getPoints()
- .stream()
- .findFirst()
- .orElseThrow();
- assertInstanceOf(HistogramPointData.class, pd,
"Expected HistogramPointData");
- HistogramPointData hpd = (HistogramPointData) pd;
- assertEquals(1L, hpd.getCount());
- assertTrue(hpd.getMin() >= DELAY);
- })
- .count();
- assertTrue(dataCount > 0, "No metric data found with name A");
- });
MockEndpoint.assertIsSatisfied(context);
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/LoopNoBreakOnShutdownTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/LoopNoBreakOnShutdownTest.java
index 5d735cc1f6c4..3140165ab498 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/LoopNoBreakOnShutdownTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/LoopNoBreakOnShutdownTest.java
@@ -16,8 +16,6 @@
*/
package org.apache.camel.processor;
-import java.util.concurrent.CompletableFuture;
-
import org.apache.camel.ContextTestSupport;
import org.apache.camel.ShutdownRoute;
import org.apache.camel.builder.RouteBuilder;
@@ -36,10 +34,14 @@ class LoopNoBreakOnShutdownTest extends ContextTestSupport {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedMinimumMessageCount(LOOP_COUNT);
- CompletableFuture<Object> future = template.asyncSendBody("seda:foo",
"foo");
- // asyncSendBody to seda completes almost instantly (just enqueues),
but on
- // slow CI the handoff can take longer — use a generous timeout
- await().atMost(10, SECONDS).until(future::isDone);
+ template.asyncSendBody("seda:foo", "foo");
+
+ // Wait until at least 1 loop iteration has completed and reached
mock:result.
+ // This guarantees the exchange is registered in the inflight
repository and
+ // the LoopProcessor's taskCount > 0, so the shutdown strategy will
properly
+ // wait for all 100 iterations to complete instead of seeing 0 inflight
+ // exchanges and proceeding with immediate shutdown.
+ await().atMost(10, SECONDS).until(() -> mock.getReceivedCounter() >=
1);
context.stop();
diff --git
a/test-infra/camel-test-infra-elasticsearch/src/main/java/org/apache/camel/test/infra/elasticsearch/services/ElasticSearchLocalContainerInfraService.java
b/test-infra/camel-test-infra-elasticsearch/src/main/java/org/apache/camel/test/infra/elasticsearch/services/ElasticSearchLocalContainerInfraService.java
index 772c32d1f68e..5ed456ede1dc 100644
---
a/test-infra/camel-test-infra-elasticsearch/src/main/java/org/apache/camel/test/infra/elasticsearch/services/ElasticSearchLocalContainerInfraService.java
+++
b/test-infra/camel-test-infra-elasticsearch/src/main/java/org/apache/camel/test/infra/elasticsearch/services/ElasticSearchLocalContainerInfraService.java
@@ -77,6 +77,10 @@ public class ElasticSearchLocalContainerInfraService
withPassword(PASSWORD);
+ // Disable disk watermarks to prevent "disk usage exceeded
flood-stage watermark"
+ // errors when running on CI machines with limited disk space
+ withEnv("cluster.routing.allocation.disk.threshold_enabled",
"false");
+
ContainerEnvironmentUtil.configurePort(this, fixedPort,
ELASTIC_SEARCH_PORT);
setWaitStrategy(
diff --git
a/test-infra/camel-test-infra-microprofile-lra/src/main/java/org/apache/camel/test/infra/microprofile/lra/services/MicroprofileLRALocalContainerInfraService.java
b/test-infra/camel-test-infra-microprofile-lra/src/main/java/org/apache/camel/test/infra/microprofile/lra/services/MicroprofileLRALocalContainerInfraService.java
index 03cd8520eb17..dd8eaf5ffcf8 100644
---
a/test-infra/camel-test-infra-microprofile-lra/src/main/java/org/apache/camel/test/infra/microprofile/lra/services/MicroprofileLRALocalContainerInfraService.java
+++
b/test-infra/camel-test-infra-microprofile-lra/src/main/java/org/apache/camel/test/infra/microprofile/lra/services/MicroprofileLRALocalContainerInfraService.java
@@ -26,6 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
import org.testcontainers.utility.DockerImageName;
@InfraService(service = MicroprofileLRAInfraService.class,
@@ -65,13 +66,21 @@ public class MicroprofileLRALocalContainerInfraService
withNetworkAliases(networkAlias)
// Shorten the Narayana recovery period so that failed
LRA
- // participant callbacks are retried quickly in tests
- // (default: periodicRecoveryPeriod=120s,
recoveryBackoffPeriod=10s).
+ // participant callbacks are retried quickly in tests.
+ // The default is periodicRecoveryPeriod=120s,
recoveryBackoffPeriod=10s.
+ // The container's Dockerfile sets JAVA_OPTS_APPEND
with host
+ // binding and log manager; we must not overwrite it.
Use
+ // JAVA_TOOL_OPTIONS (always processed by the JVM at
startup)
+ // to inject additional system properties.
.withEnv("JAVA_TOOL_OPTIONS",
"-Dcom.arjuna.ats.arjuna.recovery.periodicRecoveryPeriod=2 "
+
"-Dcom.arjuna.ats.arjuna.recovery.recoveryBackoffPeriod=1")
- .waitingFor(Wait.forListeningPort())
-
.waitingFor(Wait.forLogMessage(".*lra-coordinator-quarkus.*Listening on.*", 1));
+ // Use WaitAllStrategy to combine both checks (the
second
+ // waitingFor() call would silently replace the first)
+ .waitingFor(new WaitAllStrategy()
+ .withStrategy(Wait.forListeningPort())
+ .withStrategy(
+
Wait.forLogMessage(".*lra-coordinator-quarkus.*Listening on.*", 1)));
ContainerEnvironmentUtil.configurePort(this, fixedPort,
MicroprofileLRAProperties.DEFAULT_PORT);
}
