(camel) branch main updated: CAMEL-25487: Fix flaky GrpcProducerStreamingTest recovery test (#27806)

[davsclaus]

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 60006fac2d9a CAMEL-25487: Fix flaky GrpcProducerStreamingTest recovery 
test (#27806)
60006fac2d9a is described below

commit 60006fac2d9a99609c9d058f01d8de008a931fad
Author: Aurélien Pupier <apup...@ibm.com>
AuthorDate: Thu Oct 15 14:02:10 2026 +0200

    CAMEL-25487: Fix flaky GrpcProducerStreamingTest recovery test (#27806)
---
 .../component/grpc/GrpcProducerStreamingTest.java  | 29 ++++++++++++++++------
 1 file changed, 21 insertions(+), 8 deletions(-)

diff --git 
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcProducerStreamingTest.java
 
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcProducerStreamingTest.java
index f29d498fb196..b69ee8d3f1c1 100644
--- 
a/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcProducerStreamingTest.java
+++ 
b/components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcProducerStreamingTest.java
@@ -82,31 +82,37 @@ public class GrpcProducerStreamingTest extends 
CamelTestSupport {
     public void testPingAsyncAsyncRecovery() throws Exception {
         int messageGroupCount = 5;
         for (int i = 1; i <= messageGroupCount; i++) {
-            template.sendBody("direct:grpc-stream-async-async-route",
+            template.sendBody("direct:grpc-stream-async-async-recovery-route",
                     
PingRequest.newBuilder().setPingName(String.valueOf(i)).build());
         }
 
-        template.sendBody("direct:grpc-stream-async-async-route", 
PingRequest.newBuilder().setPingName("error").build());
+        template.sendBody("direct:grpc-stream-async-async-recovery-route",
+                PingRequest.newBuilder().setPingName("error").build());
 
         MockEndpoint replies = getMockEndpoint("mock:grpc-replies");
-        replies.expectedMessageCount(messageGroupCount);
+        // 5 normal replies + 1 forwarded onError event (forwardOnError=true 
on this route)
+        replies.expectedMessageCount(messageGroupCount + 1);
         replies.assertIsSatisfied();
 
-        await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> 
assertListSize(pingPongServer.getLastStreamRequests(), 1));
+        // The forwarder closes its current stream before forwarding the 
onError event to the
+        // replies endpoint, so once the event is received the next batch 
opens a fresh stream.
+        // Reset the mock before sending: replies arrive asynchronously and 
must not be dropped
+        // by a late reset().
+        replies.reset();
+        replies.expectedMessageCount(messageGroupCount);
 
         for (int i = messageGroupCount + 1; i <= 2 * messageGroupCount; i++) {
-            template.sendBody("direct:grpc-stream-async-async-route",
+            template.sendBody("direct:grpc-stream-async-async-recovery-route",
                     
PingRequest.newBuilder().setPingName(String.valueOf(i)).build());
         }
 
-        replies.reset();
-        replies.expectedMessageCount(messageGroupCount);
         replies.assertIsSatisfied();
 
         context().stop();
 
         assertNotNull(pingPongServer.getLastStreamRequests());
-        await().untilAsserted(() -> 
assertListSize(pingPongServer.getLastStreamRequests(), 2));
+        await().atMost(5, TimeUnit.SECONDS)
+                .untilAsserted(() -> 
assertListSize(pingPongServer.getLastStreamRequests(), 2));
         assertListSize(pingPongServer.getLastStreamRequests().get(0), 
messageGroupCount + 1);
         assertListSize(pingPongServer.getLastStreamRequests().get(1), 
messageGroupCount);
     }
@@ -120,6 +126,13 @@ public class GrpcProducerStreamingTest extends 
CamelTestSupport {
                         .to("grpc://localhost:" + grpcServer.getPort()
                             + 
"/org.apache.camel.component.grpc.PingPong?producerStrategy=STREAMING&streamRepliesTo=direct:grpc-replies&method=pingAsyncAsync");
 
+                // Same endpoint but with forwardOnError=true so the recovery 
test receives a
+                // deterministic client-side signal that the stream has been 
torn down before
+                // sending the second batch of messages.
+                from("direct:grpc-stream-async-async-recovery-route")
+                        .to("grpc://localhost:" + grpcServer.getPort()
+                            + 
"/org.apache.camel.component.grpc.PingPong?producerStrategy=STREAMING&streamRepliesTo=direct:grpc-replies&method=pingAsyncAsync&forwardOnError=true");
+
                 from("direct:grpc-replies")
                         .to("mock:grpc-replies");
             }