m-trieu commented on code in PR #34148:
URL: https://github.com/apache/beam/pull/34148#discussion_r2045204818
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -393,6 +408,41 @@ private StreamingDataflowWorker(
LOG.debug("LocalWindmillHostport: {}", options.getLocalWindmillHostport());
}
+ private GetDataClient createGetDataClient(
+ DataflowWorkerHarnessOptions options,
+ WindmillServerStub windmillServer,
+ ThrottlingGetDataMetricTracker getDataMetricTracker,
+ @Nullable WindmillStreamPool<GetDataStream> getDataStreamPool) {
+ if (options.isEnableStreamingEngine()) {
+ Preconditions.checkNotNull(
+ getDataStreamPool, "getDataStreamPool must be initialized for
Streaming Engine");
+ return new StreamPoolGetDataClient(getDataMetricTracker,
getDataStreamPool);
+ } else {
+ return new ApplianceGetDataClient(windmillServer, getDataMetricTracker);
+ }
+ }
+
+ private WorkCommitter createWorkCommitter(
+ DataflowWorkerHarnessOptions options,
+ WindmillServerStub windmillServer,
+ int numCommitThreads,
+ Consumer<CompleteCommit> onCommitComplete) {
+ WeightedSemaphore<Commit> maxCommitByteSemaphore =
Commits.maxCommitByteSemaphore();
Review Comment:
move into `if (options.isEnableStreamingEngine())` block
--
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
