m-trieu commented on code in PR #34367:
URL: https://github.com/apache/beam/pull/34367#discussion_r2023644717
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java:
##########
@@ -57,15 +64,26 @@
@Internal
@ThreadSafe
final class WindmillStreamSender implements GetWorkBudgetSpender, StreamSender
{
- private static final String STREAM_STARTER_THREAD_NAME =
"StartWindmillStreamThread-%d";
- private final AtomicBoolean started;
- private final AtomicReference<GetWorkBudget> getWorkBudget;
- private final GetWorkStream getWorkStream;
+ private static final Logger LOG =
LoggerFactory.getLogger(WindmillStreamSender.class);
+ private static final String STREAM_MANAGER_THREAD_NAME_FORMAT =
"WindmillStreamManagerThread";
+ private static final int GET_WORK_STREAM_TTL_MINUTES = 45;
+
+ private final AtomicBoolean isRunning = new AtomicBoolean(false);
private final GetDataStream getDataStream;
private final CommitWorkStream commitWorkStream;
private final WorkCommitter workCommitter;
private final StreamingEngineThrottleTimers streamingEngineThrottleTimers;
private final ExecutorService streamStarter;
+ private final String backendWorkerToken;
+
+ @GuardedBy("activeGetWorkStream")
+ private final AtomicReference<GetWorkStream> activeGetWorkStream;
+
+ @GuardedBy("activeGetWorkStream")
+ private final AtomicReference<GetWorkBudget> getWorkBudget;
+
+ @GuardedBy("activeGetWorkStream")
Review Comment:
setting the budget on a stream can affect how many items/bytes we fetch in
the header (and subsequent extension calls).
wanted to simplify the stream mechanics by preventing setBudget() from being
called mid stream creation. That way either we start the stream first, and then
call set budget which will update the internal stream budget and send an
extension (possibly) OR we set the budget first, and start the stream w/ the
new budget.
--
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
