Fix flaky tests: Disruptor, Seda, Scheduler (batch 16)

components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorBlockWhenFullTest.java

@@ -44,7 +44,11 @@ protected RouteBuilder createRouteBuilder() {
         return new RouteBuilder() {
             @Override
             public void configure() {
-                from(DEFAULT_URI).delay(DELAY).to(MOCK_URI);
+                // syncDelayed() is required so the consumer blocks for
+                // the full delay inside onEvent(), keeping the ring buffer
+                // full long enough for the "exception when full" test to
+                // hit InsufficientCapacityException reliably.
+                from(DEFAULT_URI).delay(DELAY).syncDelayed().to(MOCK_URI);
             }
         };
     }

core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerNoPolledMessagesTest.java

@@ -16,6 +16,8 @@
  */
 package org.apache.camel.component.scheduler;
 
+import java.util.Date;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.camel.ContextTestSupport;
@@ -25,21 +27,38 @@
 import org.apache.camel.component.mock.MockEndpoint;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
 public class SchedulerNoPolledMessagesTest extends ContextTestSupport {
 
     @Test
     public void testSchedulerNoPolledMessages() throws Exception {
         MockEndpoint mock = getMockEndpoint("mock:result");
         mock.expectedMinimumMessageCount(3);
-        // the first 2 fire quickly (100ms interval), but CI environments can add
-        // significant jitter from GC pauses and CPU contention — use wide windows
-        mock.message(0).arrives().between(0, 2000).millis().beforeNext();
-        mock.message(1).arrives().between(0, 2000).millis().beforeNext();
-        // the last message should be slower as the backoff idle has kicked in
-        // (backoffMultiplier=10 × delay=100ms = ~1000ms), but allow extra margin for CI
-        mock.message(2).arrives().between(200, 5000).millis().afterPrevious();
 
         MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
+
+        // Verify backoff by checking timestamps of the first 3 received
+        // messages directly.  We do NOT use the arrives().afterPrevious()
+        // API here because it has a subtle race: afterPrevious() actually
+        // compares with the NEXT message's timestamp when one has already
+        // arrived, so a 4th message arriving ~100 ms after the 3rd can
+        // violate the lower bound intended for the backoff gap (msg 1→2).
+        List<Exchange> received = mock.getReceivedExchanges();
+        long t0 = received.get(0).getProperty(Exchange.RECEIVED_TIMESTAMP, Date.class).getTime();
+        long t1 = received.get(1).getProperty(Exchange.RECEIVED_TIMESTAMP, Date.class).getTime();
+        long t2 = received.get(2).getProperty(Exchange.RECEIVED_TIMESTAMP, Date.class).getTime();
+
+        // The first 2 messages fire quickly (~100 ms scheduler interval)
+        long gap01 = t1 - t0;
+        assertTrue(gap01 <= 2000,
+                "Gap between message 0 and 1 should be <= 2000 ms, was " + gap01 + " ms");
+
+        // After 2 idle polls the backoff kicks in
+        // (backoffMultiplier=10 × delay=100 ms ≈ 1000 ms)
+        long gap12 = t2 - t1;
+        assertTrue(gap12 >= 200,
+                "Backoff should cause gap >= 200 ms between message 1 and 2, was " + gap12 + " ms");
     }
 
     @Override

core/camel-core/src/test/java/org/apache/camel/component/seda/SedaBlockWhenFullTest.java

@@ -93,7 +93,12 @@ public void testSedaBlockingWhenFull() throws Exception {
 
     @Test
     public void testAsyncSedaBlockingWhenFull() throws Exception {
-        getMockEndpoint(MOCK_URI).setExpectedMessageCount(QUEUE_SIZE + 1);
+        // Use expectedMinimumMessageCount instead of setExpectedMessageCount:
+        // with blockWhenFull=true and async sends, all messages eventually
+        // arrive at the mock (they block and wait for queue space rather
+        // than failing). An exact-count assertion races against the
+        // remaining messages still flowing through the 130ms delay pipeline.
+        getMockEndpoint(MOCK_URI).expectedMinimumMessageCount(QUEUE_SIZE + 1);
 
         SedaEndpoint seda = context.getEndpoint(BLOCK_WHEN_FULL_URI, SedaEndpoint.class);
         assertEquals(QUEUE_SIZE, seda.getQueue().remainingCapacity());