davsclaus commented on code in PR #27806:
URL: https://github.com/apache/camel/pull/27806#discussion_r4275133981
##########
components/camel-grpc/src/main/java/org/apache/camel/component/grpc/client/GrpcStreamingExchangeForwarder.java:
##########
@@ -37,9 +37,15 @@ class GrpcStreamingExchangeForwarder implements
GrpcExchangeForwarder {
private final Object grpcStub;
private final Lock lock = new ReentrantLock();
- private volatile StreamObserver<Object> currentStream;
- private volatile StreamObserver<Object> currentResponseObserver;
+ /**
+ * Holds the current open stream together with a closed flag. The holder
is replaced atomically (under lock) each
+ * time the stream is recreated. The closed flag is set by the gRPC
response thread when the server terminates the
+ * call, signalling that any concurrent or subsequent forward() must open
a fresh stream.
+ */
+ private StreamHolder currentHolder;
+
+ private StreamObserver<Object> currentResponseObserver;
Review Comment:
This used to be `volatile`. `shutdown()` (line 82) reads it without the
lock, from the producer stop thread, so that thread can see a stale `null` and
skip `onCompleted()`. If the class is not reverted, please keep the field
volatile:
```suggestion
private volatile StreamObserver<Object> currentResponseObserver;
```
##########
components/camel-grpc/src/test/java/org/apache/camel/component/grpc/GrpcProducerStreamingTest.java:
##########
@@ -82,20 +82,25 @@ public void testPingAsyncAsync() throws Exception {
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));
-
+ // At this point closeCurrentHolder() has already completed on the
client side:
+ // closeCurrentHolder() runs *before* streamObserver.onError() is
called, and
+ // streamObserver.onError() is what ultimately delivers the onError
event to this mock.
+ // MockEndpoint.assertIsSatisfied() above confirms that delivery has
occurred, so the
+ // next batch is guaranteed to open a fresh stream — no
timing-dependent wait needed.
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());
}
Review Comment:
`replies.reset()` and `replies.expectedMessageCount(messageGroupCount)`
(lines 107-108) run after this loop. The replies to 6..10 arrive asynchronously
on the gRPC thread and can reach the mock before `reset()`, which drops them
from the count. The test then fails with `Expected: <5> but was: <n>`, the same
symptom as in the JIRA. Please move the reset and the expectation above this
loop.
##########
components/camel-grpc/src/main/java/org/apache/camel/component/grpc/client/GrpcStreamingExchangeForwarder.java:
##########
@@ -74,38 +80,44 @@ public void forward(Exchange exchange) {
@Override
public void shutdown() {
if (this.currentResponseObserver != null) {
-
checkAndRecreateStreamObserver(this.currentResponseObserver).onCompleted();
+ getOrCreateStream(this.currentResponseObserver).onCompleted();
}
- doCloseStream();
+ closeCurrentHolder();
}
- private StreamObserver<Object>
checkAndRecreateStreamObserver(StreamObserver<Object> responseObserver) {
- StreamObserver<Object> curStream = this.currentStream;
- if (curStream == null) {
- lock.lock();
- try {
- if (this.currentStream == null) {
- this.currentResponseObserver = responseObserver;
- this.currentStream = doCreateStream(responseObserver);
- }
-
- curStream = this.currentStream;
- } finally {
- lock.unlock();
+ /**
+ * Returns the current open stream, creating a new one if there is none or
if the current one has been closed by the
+ * server (onError / onCompleted). Always called under the lock so that a
concurrent doCloseHolder() cannot sneak in
+ * between the null-check and the return.
+ */
+ private StreamObserver<Object> getOrCreateStream(StreamObserver<Object>
responseObserver) {
+ lock.lock();
+ try {
+ if (this.currentHolder == null || this.currentHolder.closed) {
+ this.currentResponseObserver = responseObserver;
+ this.currentHolder = new
StreamHolder(doCreateStream(responseObserver));
}
- }
- StreamObserver<Object> curResponseObserver =
this.currentResponseObserver;
- if (curResponseObserver != null &&
!curResponseObserver.equals(responseObserver)) {
- throw new IllegalArgumentException("This forwarder must always use
the same response observer");
+ StreamObserver<Object> curResponseObserver =
this.currentResponseObserver;
+ if (curResponseObserver != null &&
!curResponseObserver.equals(responseObserver)) {
+ throw new IllegalArgumentException("This forwarder must always
use the same response observer");
+ }
+ return this.currentHolder.stream;
+ } finally {
+ lock.unlock();
}
Review Comment:
The check is done under the lock, but the stream is used after `unlock()`:
`forward()` calls `streamObserver.onNext(...)` at line 69 without holding the
lock. `closeCurrentHolder()` on the gRPC response thread can still run between
this return and that `onNext`, which is the same window as the old volatile
fast path. So this does not remove the TOCTOU, and it adds a lock acquisition
per message. The test change already makes the test deterministic without this
(the old `doCloseStream()` also ran before `streamObserver.onError`). Could
this file be reverted?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
