m-trieu commented on code in PR #32778:
URL: https://github.com/apache/beam/pull/32778#discussion_r1849052344
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -234,107 +230,199 @@ private StreamingDataflowWorker(
ID_GENERATOR,
configFetcher.getGlobalConfigHandle(),
stageInfoMap);
-
ThrottlingGetDataMetricTracker getDataMetricTracker =
new ThrottlingGetDataMetricTracker(memoryMonitor);
- WorkerStatusPages workerStatusPages =
- WorkerStatusPages.create(DEFAULT_STATUS_PORT, memoryMonitor);
- StreamingWorkerStatusPages.Builder statusPagesBuilder =
StreamingWorkerStatusPages.builder();
- int stuckCommitDurationMillis;
- GetDataClient getDataClient;
- HeartbeatSender heartbeatSender;
- if (windmillServiceEnabled) {
- WindmillStreamPool<GetDataStream> getDataStreamPool =
- WindmillStreamPool.create(
- Math.max(1, options.getWindmillGetDataStreamCount()),
- GET_DATA_STREAM_TIMEOUT,
- windmillServer::getDataStream);
- getDataClient = new StreamPoolGetDataClient(getDataMetricTracker,
getDataStreamPool);
- if (options.getUseSeparateWindmillHeartbeatStreams() != null) {
+ // Status page members. Different implementations on whether the harness
is streaming engine
+ // direct path, streaming engine cloud path, or streaming appliance.
+ @Nullable ChannelzServlet channelzServlet = null;
+ Consumer<PrintWriter> getDataStatusProvider;
+ Supplier<Long> currentActiveCommitBytesProvider;
+ if (isDirectPathPipeline(options)) {
+ WeightedSemaphore<Commit> maxCommitByteSemaphore =
Commits.maxCommitByteSemaphore();
+ FanOutStreamingEngineWorkerHarness fanOutStreamingEngineWorkerHarness =
+ FanOutStreamingEngineWorkerHarness.create(
+ createJobHeader(options, clientId),
+ GetWorkBudget.builder()
+ .setItems(chooseMaxBundlesOutstanding(options))
+ .setBytes(MAX_GET_WORK_FETCH_BYTES)
+ .build(),
+ windmillStreamFactory,
+ (workItem, watermarks, processingContext,
getWorkStreamLatencies) ->
+ computationStateCache
+ .get(processingContext.computationId())
+ .ifPresent(
+ computationState -> {
+ memoryMonitor.waitForResources("GetWork");
+ streamingWorkScheduler.scheduleWork(
+ computationState,
+ workItem,
+ watermarks,
+ processingContext,
+ getWorkStreamLatencies);
+ }),
+ createStubFactory(options),
+ GetWorkBudgetDistributors.distributeEvenly(),
+ Preconditions.checkNotNull(dispatcherClient),
+ commitWorkStream ->
+ StreamingEngineWorkCommitter.builder()
+ // Share the commitByteSemaphore across all created
workCommitters.
+ .setCommitByteSemaphore(maxCommitByteSemaphore)
+
.setBackendWorkerToken(commitWorkStream.backendWorkerToken())
+ .setOnCommitComplete(this::onCompleteCommit)
+
.setNumCommitSenders(Math.max(options.getWindmillServiceCommitThreads(), 1))
+ .setCommitWorkStreamFactory(
+ () -> CloseableStream.create(commitWorkStream, () ->
{}))
+ .build(),
+ getDataMetricTracker);
+ getDataStatusProvider = getDataMetricTracker::printHtml;
+ currentActiveCommitBytesProvider =
+ fanOutStreamingEngineWorkerHarness::currentActiveCommitBytes;
+ channelzServlet =
+ createChannelZServlet(
+ options,
fanOutStreamingEngineWorkerHarness::currentWindmillEndpoints);
+ this.streamingWorkerHarness = fanOutStreamingEngineWorkerHarness;
+ } else {
+ Windmill.GetWorkRequest 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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
