This is an automated email from the ASF dual-hosted git repository.
apupier 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 fadef82e50b5 CAMEL-26684: Fix flaky GrpcRouteControlledReplyTypeTest
fadef82e50b5 is described below
commit fadef82e50b538269527c432648b2f71fc654a2b
Author: Aurélien Pupier <[email protected]>
AuthorDate: Fri Nov 6 09:58:09 2026 +0100
CAMEL-26684: Fix flaky GrpcRouteControlledReplyTypeTest
delegationUnchangedRequestIsNotSentAsResponse sends onNext and onCompleted
back to back and expects the rc-delegation-unchanged route to process two
exchanges: the onNext one, which fails with the reply type error, and the
forwarded onCompleted one (forwardOnCompleted=true).
The reply type error closes the call. When it happens before the server
handled the half-close of the client, the onCompleted event is never
forwarded, so the second exchange never exists and awaitDoneExchanges times
out with a ConditionTimeoutException.
Make the order of the events deterministic instead of relaxing the
assertions, which still check for the two exchanges, the failure of the
first one and that the second one does not fail:
- the onNext exchange waits, on another thread than the one of the call, for
the onCompleted event to reach the route, so the call is not closed before
the half-close is handled;
- the onCompleted exchange waits for the onNext exchange to have failed
before it replies, so it is ignored on the closed call as expected.
Reproduced under CPU load with the test repeated 600 times: 4 failures
without the change (and 1 in a previous run of 300), none with it.
Co-authored-by: IBM Bob 2.2.1
Signed-off-by: Aurélien Pupier <[email protected]>
---
.../grpc/GrpcRouteControlledReplyTypeTest.java | 28 ++++++++++++++++++++++
1 file changed, 28 insertions(+)
diff --git
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
index d388a40f353d..919388f545ab 100644
---
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
+++
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcRouteControlledReplyTypeTest.java
@@ -22,6 +22,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import io.grpc.ManagedChannel;
@@ -66,6 +67,11 @@ public class GrpcRouteControlledReplyTypeTest extends
GrpcTestSupport {
private final Map<String, ManagedChannel> channels = new
ConcurrentHashMap<>();
// the exception of each exchange done by a consumer route (null when it
completed), by route id
private final Map<String, List<Exception>> doneExchanges = new
ConcurrentHashMap<>();
+ // rc-delegation-unchanged: the onCompleted event reached the route, and
the onNext exchange failed. They order the
+ // events, as the call must not be closed by the reply type error before
the server handles the half-close of the
+ // client, otherwise there is no onCompleted event to forward
+ private final CountDownLatch delegationOnCompletedReceived = new
CountDownLatch(1);
+ private final CountDownLatch delegationOnNextFailed = new
CountDownLatch(1);
@AfterEach
public void stopGrpcChannels() {
@@ -284,6 +290,10 @@ public class GrpcRouteControlledReplyTypeTest extends
GrpcTestSupport {
doneExchanges.computeIfAbsent(exchange.getFromRouteId(),
id -> Collections.synchronizedList(new
ArrayList<>()))
.add(exchange.getException());
+ if (event instanceof CamelEvent.ExchangeFailedEvent
+ &&
"rc-delegation-unchanged".equals(exchange.getFromRouteId())) {
+ delegationOnNextFailed.countDown();
+ }
}
}
@@ -354,6 +364,20 @@ public class GrpcRouteControlledReplyTypeTest extends
GrpcTestSupport {
// client and bidirectional streaming calls: the consumer
forwards each event with its type header
from(CONSUMER +
"&hash=5&consumerStrategy=DELEGATION&forwardOnCompleted=true&muteException=false")
.routeId("rc-delegation-unchanged")
+ .choice()
+ .when(header(GrpcConstants.GRPC_EVENT_TYPE_HEADER)
+
.isEqualTo(GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED))
+ // the onCompleted reply is sent once the call is
closed by the failure of the onNext reply
+ .process(e -> {
+ delegationOnCompletedReceived.countDown();
+ awaitLatch(delegationOnNextFailed);
+ })
+ .otherwise()
+ // another thread than the one of the call, which has
to deliver the onCompleted event
+ .threads(1)
+ .process(e ->
awaitLatch(delegationOnCompletedReceived))
+ .end() // threads
+ .end() // choice
.to(REPLY);
from(CONSUMER +
"&hash=6&consumerStrategy=DELEGATION&forwardOnCompleted=true")
@@ -369,6 +393,10 @@ public class GrpcRouteControlledReplyTypeTest extends
GrpcTestSupport {
};
}
+ private static void awaitLatch(CountDownLatch latch) throws
InterruptedException {
+ assertTrue(latch.await(5, TimeUnit.SECONDS), "the latch was not
released within 5 seconds");
+ }
+
private static Pong toPong(PingRequest request, String suffix) {
return new Pong(request.getPingName() + suffix, request.getPingId());
}
