m-trieu commented on code in PR #34367:
URL: https://github.com/apache/beam/pull/34367#discussion_r2025713645
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java:
##########
@@ -165,4 +201,39 @@ long getAndResetThrottleTime() {
long getCurrentActiveCommitBytes() {
return workCommitter.currentActiveCommitBytes();
}
+
+ /**
+ * Creates, starts, and gracefully terminates {@link GetWorkStream} before
the clientside deadline
+ * to prevent {@link
org.apache.beam.vendor.grpc.v1p69p0.io.grpc.Status#DEADLINE_EXCEEDED} errors.
+ * If at any point the server closes the stream, reconnects immediately.
+ */
+ private void getWorkStreamLoop(CountDownLatch waitForInitialStream) {
+ @Nullable GetWorkStream newStream = null;
+ while (isRunning.get()) {
+ synchronized (activeGetWorkStream) {
+ newStream = getWorkStreamFactory.get();
Review Comment:
in the current case the new budget maybe reflected in the old and new stream
which is ok, since the new stream will send the header + new budget in the
request this seems like it could be simpler?
--
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]
