github-actions[bot] commented on code in PR #41000:
URL: https://github.com/apache/beam/pull/41000#discussion_r4472915653
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/TestDataflowRunner.java:
##########
@@ -347,39 +348,110 @@ private boolean waitForStreamingJobTermination(
// Getting the final state may have timed out; it may not indicate a
failure.
// This cancellation may be the second
if (finalState == null || !finalState.isTerminal()) {
- LOG.info(
- "Dataflow job {} took longer than {} seconds to complete,
cancelling.",
- job.getJobId(),
- options.getTestTimeoutSeconds());
+ if (messageHandler.hasSeenRetryableInfrastructureError()) {
+ LOG.warn(
+ "Dataflow job {} did not complete within {} seconds while the
service was retrying an"
+ + " infrastructure error; cancelling. The failure is most
likely capacity, not the"
+ + " test: {}",
+ job.getJobId(),
+ options.getTestTimeoutSeconds(),
+ messageHandler.getErrorMessage());
+ } else {
+ LOG.info(
+ "Dataflow job {} took longer than {} seconds to complete,
cancelling.",
+ job.getJobId(),
+ options.getTestTimeoutSeconds());
+ }
try {
job.cancel();
} catch (IOException e) {
throw new RuntimeException(e);
}
return false;
} else {
+ // Retryable infrastructure errors do not count: if the job still
reached DONE, the service
+ // recovered from them and the pipeline ran to completion.
return finalState == State.DONE && !messageHandler.hasSeenError();
}
}
+ /** How waiting on a merged job ended; see {@link
#waitForMergedJobTermination}. */
+ static final class MergedJobWait {
+ enum Kind {
+ /** The job reached {@link #terminalState}. */
+ TERMINATED,
+ /** The job was still running when the timeout passed. */
+ TIMED_OUT,
+ /**
+ * The job's state could not be fetched for long enough that polling
gave up before the
+ * timeout; the job itself may be perfectly healthy.
+ */
+ POLLING_FAILED,
+ /** The waiting thread was interrupted. */
+ INTERRUPTED
+ }
+
+ final Kind kind;
+ final @Nullable State terminalState;
+
+ private MergedJobWait(Kind kind, @Nullable State terminalState) {
+ this.kind = kind;
+ this.terminalState = terminalState;
+ }
+
+ static MergedJobWait terminated(State state) {
+ return new MergedJobWait(Kind.TERMINATED, state);
+ }
+
+ static MergedJobWait of(Kind kind) {
+ return new MergedJobWait(kind, null);
+ }
+
+ @Override
+ public String toString() {
+ return kind == Kind.TERMINATED ? "terminated in state " + terminalState
: kind.toString();
+ }
+ }
+
/**
- * Waits up to {@code timeout} for a merged test job to terminate, feeding
its log messages to
- * {@code messageHandler}. Returns its terminal state, or {@code null} if
the job did not
- * terminate in time or the wait was interrupted.
+ * Waits up to {@code timeout} (negative for no limit) for a merged test job
to terminate, feeding
+ * its log messages to {@code messageHandler}.
*
* <p>Unlike a standalone batch job, which this runner waits on
indefinitely, a merged job holds
- * the verdict for several tests at once, so it is never allowed to block
them forever.
+ * the verdict for several tests at once, so it is never allowed to block
them forever. A merged
+ * streaming job is additionally watched by {@link CancelOnError}, exactly
like a standalone
+ * streaming job: a failing step would otherwise retry until the timeout.
+ *
+ * <p>{@link DataflowPipelineJob#waitUntilFinish} reports both "deadline
passed" and "could not
+ * reach the service" as {@code null}; the two are told apart here by the
time spent, since
+ * polling gives up well before a sane timeout. The distinction matters
because the right reaction
+ * differs: a job that outlived its timeout should be cancelled, a job that
merely could not be
+ * observed should be left alone.
*/
- @Nullable State waitForMergedJobTermination(
+ @SuppressWarnings("FutureReturnValueIgnored") // CancelOnError's outcome is
observed via the job
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_BAD_PRACTICE")
+ MergedJobWait waitForMergedJobTermination(
DataflowPipelineJob job, Duration timeout, ErrorMonitorMessagesHandler
messageHandler) {
+ if (options.isStreaming()) {
+ options.getExecutorService().submit(new CancelOnError(job,
messageHandler));
+ }
+ long startNanos = System.nanoTime();
Review Comment:
**🟡 Medium**: The `Future<?>` returned by `submit(new CancelOnError(job,
messageHandler))` is never cancelled when `waitForMergedJobTermination` (or
`waitForStreamingJobTermination`) returns. Once `job.waitUntilFinish(timeout,
messageHandler)` exits, nothing feeds new messages into `messageHandler`, so
`CancelOnError` can no longer observe new errors. However, if the job is not in
a terminal state (for example on `POLLING_FAILED`, where the coordinator
intentionally leaves the job running and `job.getState()` returns
`State.UNKNOWN` on repeated `IOException`s, or in unit tests with mocked
`State.RUNNING` jobs), `CancelOnError` will continue looping and polling
`job.getState()` every 3 seconds in the background on
`options.getExecutorService()`.
Consider capturing the `Future<?>` and cancelling it
(`cancelOnError.cancel(true)`) in a `finally` block when the wait finishes.
--
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]
