gemini-code-assist[bot] commented on code in PR #39367:
URL: https://github.com/apache/beam/pull/39367#discussion_r3604022933
##########
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:
When catching `Exception` (which includes `InterruptedException`), the
interrupted status of the thread is cleared. If the exception is an
`InterruptedException`, we should restore the interrupted status by calling
`Thread.currentThread().interrupt()` before rethrowing or wrapping it, so that
calling code or the thread pool can handle the interruption properly.
```java
if (exn instanceof IOException) {
throw (IOException) exn;
} else if (exn instanceof InterruptedException) {
Thread.currentThread().interrupt();
throw new RuntimeException(exn);
} else if (exn instanceof RuntimeException) {
throw (RuntimeException) exn;
} else {
throw new RuntimeException(exn);
}
```
##########
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:
Calling `stagingExecutor.shutdownNow()` will shut down the shared executor
service of the `ArtifactStagingService` instance. Since
`ArtifactStagingService` is typically a long-running gRPC service (e.g., in a
JobServer), shutting down this executor will permanently prevent any future
artifact staging requests from succeeding.
Since the `StoreArtifact` task has already failed and terminated (which is
what set the exception on `totalPendingBytes` and triggered this catch block),
there is no need to shut down or cancel anything in the executor. We can safely
remove this call.
```suggestion
state = State.ERROR;
```
--
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]
