scwhittle commented on code in PR #35120:
URL: https://github.com/apache/beam/pull/35120#discussion_r2243662010
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ExecutionStateSampler.java:
##########
@@ -167,9 +198,15 @@ private Void stateSampler() throws Exception {
Thread.sleep(difference);
} else {
long millisSinceLastSample = currentTimeMillis - lastSampleTimeMillis;
+ Optional<String> timeoutMsg = Optional.empty();
synchronized (activeStateTrackers) {
for (ExecutionStateTracker activeTracker : activeStateTrackers) {
- activeTracker.takeSample(currentTimeMillis, millisSinceLastSample);
+ timeoutMsg =
+ timeoutMsg.or(
+ () -> activeTracker.takeSample(currentTimeMillis,
millisSinceLastSample));
+ if (timeoutMsg.isPresent() && this.onTimeoutExceededCallback !=
null) {
Review Comment:
move this outside the synchronized block so we're not calling the callback
under it.
we're only going to look at the first timeoutMsg but that is fine.
--
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
