m-trieu commented on code in PR #32774:
URL: https://github.com/apache/beam/pull/32774#discussion_r1817086414
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java:
##########
@@ -302,38 +324,68 @@ public void appendSpecificHtml(PrintWriter writer) {
}
private <ResponseT> ResponseT issueRequest(QueuedRequest request,
ParseFn<ResponseT> parseFn) {
- while (true) {
+ while (!isShutdown()) {
request.resetResponseStream();
try {
queueRequestAndWait(request);
return parseFn.parse(request.getResponseStream());
- } catch (CancellationException e) {
- // Retry issuing the request since the response stream was cancelled.
- continue;
+ } catch (AppendableInputStream.InvalidInputStreamStateException |
CancellationException e) {
+ handleShutdown(request, e);
+ if (!(e instanceof CancellationException)) {
+ throw e;
+ }
} catch (IOException e) {
LOG.error("Parsing GetData response failed: ", e);
- continue;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
+ handleShutdown(request, e);
throw new RuntimeException(e);
} finally {
pending.remove(request.id());
}
}
+
+ throw new WindmillStreamShutdownException(
+ "Cannot send request=[" + request + "] on closed stream.");
+ }
+
+ private void handleShutdown(QueuedRequest request, Throwable... causes) {
+ if (isShutdown()) {
+ WindmillStreamShutdownException shutdownException =
+ new WindmillStreamShutdownException(
+ "Cannot send request=[" + request + "] on closed stream.");
+
+ for (Throwable cause : causes) {
+ shutdownException.addSuppressed(cause);
+ }
+
+ throw shutdownException;
+ }
+ }
+
+ private void handleShutdown(QueuedBatch batch) {
+ if (isShutdown()) {
+ throw new WindmillStreamShutdownException(
+ "Stream was closed when attempting to send " + batch.requestsCount()
+ " requests.");
+ }
}
private void queueRequestAndWait(QueuedRequest request) throws
InterruptedException {
QueuedBatch batch;
boolean responsibleForSend = false;
- CountDownLatch waitForSendLatch = null;
- synchronized (batches) {
+ @Nullable QueuedBatch prevBatch = null;
+ synchronized (shutdownLock) {
+ if (isShutdown()) {
+ handleShutdown(request);
Review Comment:
done
--
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]
