apanich commented on code in PR #35120:
URL: https://github.com/apache/beam/pull/35120#discussion_r2136068779
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ExecutionStateSampler.java:
##########
@@ -65,6 +66,7 @@ public class ExecutionStateSampler {
private static final Logger LOG =
LoggerFactory.getLogger(ExecutionStateSampler.class);
private static final int DEFAULT_SAMPLING_PERIOD_MS = 200;
Review Comment:
I see that other places also use primitives (int and long) instead of java
duration. I wonder why it is that? @Abacn would you know? Also seems like joda
time is used instead of java time :(
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ExecutionStateSampler.java:
##########
@@ -357,6 +377,19 @@ private void takeSample(long currentTimeMillis, long
millisSinceLastSample) {
transitionsAtLastSample = transitionsAtThisSample;
} else {
long lullTimeMs = currentTimeMillis - lastTransitionTimeMillis.get();
+
+ try {
+ if (lullTimeMs >
TimeUnit.MINUTES.toMillis(lullTimeMinuteForRestart)) {
+ throw new TimeoutException(
+ String.format(
+ "The ptransform has been stuck for more than %d minutes,
the SDK worker will"
+ + " restart",
+ lullTimeMinuteForRestart));
+ }
+ } catch (TimeoutException e) {
+ LOG.error(e.getMessage());
Review Comment:
I'm not sure I understand how it works. Here you handle exception (i.e. no
TimeoutException is thrown outside of this function. But in test you do verify
that exception is thrown...
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java:
##########
@@ -424,4 +424,15 @@ public String create(PipelineOptions options) {
return String.format("%s/%s", info.getName(),
info.getVersion()).replace(" ", "_");
}
}
+
+ /**
+ * The time limit (minute) that an SDK worker allows for a PTransform
operation before signaling
+ * the runner harness to restart the SDK worker.
+ */
+ @Description("The time limit (minute) that an SDK worker allows for a
PTransform operation "
+ + "before signaling the runner harness to restart the SDK worker.")
+ int getPtransformTimeoutDuration();
Review Comment:
I wonder why not using java.util.Duration
Duration getPtransformTimeout();
void setPtransformTimeout(Duration duration);
--
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
