m-trieu commented on code in PR #32778:
URL: https://github.com/apache/beam/pull/32778#discussion_r1857063226
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -526,60 +535,72 @@ public static StreamingDataflowWorker
fromOptions(DataflowWorkerHarnessOptions o
WorkUnitClient dataflowServiceClient,
GrpcWindmillStreamFactory.Builder windmillStreamFactoryBuilder,
Function<ComputationConfig.Fetcher, ComputationStateCache>
computationStateCacheFactory) {
- ComputationConfig.Fetcher configFetcher;
- WindmillServerStub windmillServer;
- ComputationStateCache computationStateCache;
- GrpcWindmillStreamFactory windmillStreamFactory;
- ConfigFetcherComputationStateCacheAndWindmillClient.Builder builder =
- ConfigFetcherComputationStateCacheAndWindmillClient.builder();
if (options.isEnableStreamingEngine()) {
GrpcDispatcherClient dispatcherClient =
GrpcDispatcherClient.create(options, new
WindmillStubFactoryFactoryImpl(options));
- configFetcher =
+ ComputationConfig.Fetcher configFetcher =
StreamingEngineComputationConfigFetcher.create(
options.getGlobalConfigRefreshPeriod().getMillis(),
dataflowServiceClient);
configFetcher.getGlobalConfigHandle().registerConfigObserver(dispatcherClient::onJobConfig);
- computationStateCache =
computationStateCacheFactory.apply(configFetcher);
- windmillStreamFactory =
+ ComputationStateCache computationStateCache =
+ computationStateCacheFactory.apply(configFetcher);
+ GrpcWindmillStreamFactory windmillStreamFactory =
windmillStreamFactoryBuilder
.setProcessHeartbeatResponses(
new
WorkHeartbeatResponseProcessor(computationStateCache::get))
.setHealthCheckIntervalMillis(
options.getWindmillServiceStreamingRpcHealthCheckPeriodMs())
.build();
- windmillServer = GrpcWindmillServer.create(options,
windmillStreamFactory, dispatcherClient);
- builder.setWindmillDispatcherClient(dispatcherClient);
+ return ConfigFetcherComputationStateCacheAndWindmillClient.builder()
+ .setWindmillDispatcherClient(dispatcherClient)
+ .setConfigFetcher(configFetcher)
+ .setComputationStateCache(computationStateCache)
+ .setWindmillStreamFactory(windmillStreamFactory)
+ .setWindmillServer(
+ GrpcWindmillServer.create(options, windmillStreamFactory,
dispatcherClient))
+ .build();
} else {
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
