Eliaaazzz commented on code in PR #39367:
URL: https://github.com/apache/beam/pull/39367#discussion_r3658187132
##########
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingService.java:
##########
@@ -421,8 +429,16 @@ public synchronized void
onNext(ArtifactApi.ArtifactResponseWrapper responseWrap
}
}
} catch (Exception exn) {
- LOG.error("Error submitting.", exn);
- onError(exn);
+ // The write of a previous chunk failed; surface the failure to
the client rather
+ // than leaving the stream unterminated, which would make the
client block forever.
+ LOG.error("Error staging artifacts", exn);
+ state = State.ERROR;
+ stagingExecutor.shutdownNow();
Review Comment:
I believe this one is fine as-is: `stagingExecutor` is a field of the
per-call `StreamObserver` that `reverseArtifactRetrievalService` returns — it's
created in the `START` state for each staging session, so it isn't shared
across jobs, and the existing `onError` path already shuts it down the same
way. Shutting it down here only stops the failed session's own staging tasks,
which would otherwise block on a queue nobody drains.
##########
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/artifact/ArtifactStagingService.java:
##########
@@ -282,13 +287,16 @@ public RunnerApi.ArtifactInformation call() throws
IOException {
.setTypeUrn(dest.getTypeUrn())
.setTypePayload(dest.getTypePayload())
.build();
- } catch (IOException | InterruptedException exn) {
+ } catch (Exception exn) {
// As this thread will no longer be draining the queue, we don't want
to get stuck writing
- // to it.
+ // to it. This must happen for unchecked exceptions as well:
getDestination can throw e.g.
+ // InvalidPathException, and leaving the error unset would block the
producer forever.
totalPendingBytes.setException(exn);
LOG.error("Exception staging artifacts", exn);
if (exn instanceof IOException) {
throw (IOException) exn;
+ } else if (exn instanceof RuntimeException) {
+ throw (RuntimeException) exn;
} else {
throw new RuntimeException(exn);
Review Comment:
Good catch — done. The interrupted status is now restored in both catch
blocks.
--
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]
