m-trieu commented on code in PR #32774:
URL: https://github.com/apache/beam/pull/32774#discussion_r1805813769
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java:
##########
@@ -114,50 +120,35 @@ private FanOutStreamingEngineWorkerHarness(
ChannelCachingStubFactory channelCachingStubFactory,
GetWorkBudgetDistributor getWorkBudgetDistributor,
GrpcDispatcherClient dispatcherClient,
- long clientId,
Function<WindmillStream.CommitWorkStream, WorkCommitter>
workCommitterFactory,
ThrottlingGetDataMetricTracker getDataMetricTracker) {
this.jobHeader = jobHeader;
this.getDataMetricTracker = getDataMetricTracker;
this.started = false;
this.streamFactory = streamFactory;
this.workItemScheduler = workItemScheduler;
- this.connections = new
AtomicReference<>(StreamingEngineConnectionState.EMPTY);
+ this.backends = new AtomicReference<>(StreamingEngineBackends.EMPTY);
this.channelCachingStubFactory = channelCachingStubFactory;
this.dispatcherClient = dispatcherClient;
- this.isBudgetRefreshPaused = new AtomicBoolean(false);
this.getWorkerMetadataThrottleTimer = new ThrottleTimer();
- this.newWorkerMetadataPublisher =
- singleThreadedExecutorServiceOf(PUBLISH_NEW_WORKER_METADATA_THREAD);
- this.newWorkerMetadataConsumer =
- singleThreadedExecutorServiceOf(CONSUME_NEW_WORKER_METADATA_THREAD);
- this.clientId = clientId;
- this.lastBudgetRefresh = new AtomicReference<>(Instant.EPOCH);
- this.newWindmillEndpoints =
Queues.synchronizedQueue(EvictingQueue.create(1));
- this.getWorkBudgetRefresher =
- new GetWorkBudgetRefresher(
- isBudgetRefreshPaused::get,
- () -> {
- getWorkBudgetDistributor.distributeBudget(
- connections.get().windmillStreams().values(),
totalGetWorkBudget);
- lastBudgetRefresh.set(Instant.now());
- });
- this.getWorkerMetadataStream =
- Suppliers.memoize(
- () ->
- streamFactory.createGetWorkerMetadataStream(
- dispatcherClient.getWindmillMetadataServiceStubBlocking(),
- getWorkerMetadataThrottleTimer,
- endpoints ->
- // Run this on a separate thread than the grpc stream
thread.
- newWorkerMetadataPublisher.submit(
- () -> newWindmillEndpoints.add(endpoints))));
+ this.windmillStreamManager =
+ Executors.newCachedThreadPool(
+ new
ThreadFactoryBuilder().setNameFormat(STREAM_MANAGER_THREAD_NAME).build());
+ this.workerMetadataConsumer =
+ Executors.newSingleThreadScheduledExecutor(
+ new
ThreadFactoryBuilder().setNameFormat(WORKER_METADATA_CONSUMER_THREAD_NAME).build());
+ this.getWorkBudgetDistributor = getWorkBudgetDistributor;
+ this.totalGetWorkBudget = totalGetWorkBudget;
+ this.activeMetadataVersion = Long.MIN_VALUE;
this.workCommitterFactory = workCommitterFactory;
- }
-
- private static ExecutorService singleThreadedExecutorServiceOf(String
threadName) {
- return Executors.newSingleThreadScheduledExecutor(
- new ThreadFactoryBuilder().setNameFormat(threadName).build());
+ // To satisfy CheckerFramework complaining about reference to "this" in
constructor.
+ @SuppressWarnings("methodref.receiver.bound")
+ Consumer<WindmillEndpoints> newEndpointsConsumer =
this::consumeWorkerMetadata;
+ this.getWorkerMetadataStream =
+ streamFactory.createGetWorkerMetadataStream(
+ dispatcherClient::getWindmillMetadataServiceStubBlocking,
+ getWorkerMetadataThrottleTimer,
+ newEndpointsConsumer);
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
