Consume the rest of the response stream for Watch RPC before returning. This necessary for the call to close correctly (e.g. for listeners to receive Status/trailers).
RELNOTES:
@@ -135,6 +135,7 @@ class GrpcRemoteExecutor {
return retrier.execute(
() -> {
Iterator<ChangeBatch> replies = watcherBlockingStub().watch(wr);
+ try {
while (replies.hasNext()) {
ChangeBatch cb = replies.next();
for (Change ch : cb.getChangesList()) {
@@ -151,7 +152,8 @@ class GrpcRemoteExecutor {
case DOES_NOT_EXIST:
// TODO(olaola): either make this retriable, or use a different exception.
throw new IOException(
- String.format("Operation %s lost on the remote server.", op.getName()));
+ String.format(
+ "Operation %s lost on the remote server.", op.getName()));
case EXISTS:
Operation o;
try {
@@ -176,6 +178,21 @@ class GrpcRemoteExecutor {
}
}
}
+ } finally {
+ // The blocking streaming call closes correctly only when trailers and a Status
+ // are received from the server so that onClose() is called on this call's
+ // CallListener. Under normal circumstances (no cancel/errors), these are
+ // guaranteed to be sent by the server only if replies.hasNext() has been called
+ // after all replies from the stream have been consumed.
+ try {
+ while (replies.hasNext()) {
+ replies.next();
+ }
+ } catch (StatusRuntimeException e) {
+ // Cleanup: ignore exceptions, because the meaningful errors have already been
+ // propagated.
+ }
+ }
throw new IOException(
String.format("Watch request for %s terminated with no result.", op.getName()));
});
