m-trieu commented on code in PR #32774:
URL: https://github.com/apache/beam/pull/32774#discussion_r1806054336
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcCommitWorkStream.java:
##########
@@ -317,29 +350,71 @@ public boolean commitWorkItem(
if (!canAccept(commitRequest.getSerializedSize() +
computation.length())) {
return false;
}
- PendingRequest request = new PendingRequest(computation, commitRequest,
onDone);
+
+ PendingRequest request = PendingRequest.create(computation,
commitRequest, onDone);
add(idGenerator.incrementAndGet(), request);
return true;
}
/** Flushes any pending work items to the wire. */
@Override
public void flush() {
- flushInternal(queue);
- queuedBytes = 0;
- queue.clear();
+ try {
+ if (!isShutdown()) {
+ flushInternal(queue);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ queuedBytes = 0;
+ queue.clear();
+ }
}
void add(long id, PendingRequest request) {
- assert (canAccept(request.getBytes()));
+ Preconditions.checkState(canAccept(request.getBytes()));
Review Comment:
we call `request.onDone` with the error code when we shutdown and drain the
queue
the check of isShutdown() in canAccept is to help us reject any more commits
going into the stream if we are in a shutdown state
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
