m-trieu commented on code in PR #31133:
URL: https://github.com/apache/beam/pull/31133#discussion_r1588498420
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java:
##########
@@ -931,15 +941,19 @@ void streamingDispatchLoop() {
Instant inputDataWatermark,
Instant synchronizedProcessingTime,
Windmill.WorkItem workItem,
- Collection<LatencyAttribution> getWorkStreamLatencies) -> {
- memoryMonitor.waitForResources("GetWork");
- scheduleWorkItem(
- getComputationState(computation),
- inputDataWatermark,
- synchronizedProcessingTime,
- workItem,
- getWorkStreamLatencies);
- });
+ Collection<LatencyAttribution> getWorkStreamLatencies) ->
+ computationStateCache
+ .get(computation)
+ .ifPresent(
Review Comment:
added more logging in the ComputationStateCache
```
public Optional<ComputationState> get(String computationId) {
try {
return Optional.ofNullable(computationCache.get(computationId));
} catch (ExecutionException | ComputationStateNotFoundException e) {
if (e.getCause() instanceof ComputationStateNotFoundException) {
LOG.error(
"Trying to fetch unknown computation={}, known computations are
{}.",
computationId,
getAllComputationIds());
} else {
LOG.warn("Error occurred fetching computation for computationId={}",
computationId, e);
}
}
```
--
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]
