[
https://issues.apache.org/jira/browse/BEAM-3464?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16331404#comment-16331404
]
ASF GitHub Bot commented on BEAM-3464:
--------------------------------------
kennknowles closed pull request #4398: [ BEAM-3464] Fix NexmarkLauncher to not
cancel jobs otherwise successful jobs
URL: https://github.com/apache/beam/pull/4398
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
index a4b5c5aaf5c..1bdd5f5b7ed 100644
---
a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
+++
b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
@@ -481,6 +481,7 @@ private NexmarkPerf monitor(NexmarkQuery query) {
long lastActivityMsSinceEpoch = -1;
NexmarkPerf perf = null;
boolean waitingForShutdown = false;
+ boolean cancelJob = false;
boolean publisherCancelled = false;
List<String> errors = new ArrayList<>();
@@ -527,15 +528,17 @@ private NexmarkPerf monitor(NexmarkQuery query) {
NexmarkUtils.console("job has fatal errors, cancelling.");
errors.add(String.format("Pipeline reported %s fatal errors",
fatalCount));
waitingForShutdown = true;
+ cancelJob = true;
} else if (configuration.debug && configuration.numEvents > 0
&& currPerf.numEvents == configuration.numEvents
&& currPerf.numResults >= 0 &&
quietFor.isLongerThan(DONE_DELAY)) {
- NexmarkUtils.console("streaming query appears to have finished,
cancelling job.");
+ NexmarkUtils.console("streaming query appears to have finished
waiting for completion.");
waitingForShutdown = true;
} else if (quietFor.isLongerThan(STUCK_TERMINATE_DELAY)) {
NexmarkUtils.console("streaming query appears to have gotten stuck,
cancelling job.");
- errors.add("Streaming job was cancelled since appeared stuck");
+ errors.add("Cancelling streaming job since it appeared stuck");
waitingForShutdown = true;
+ cancelJob = true;
} else if (quietFor.isLongerThan(STUCK_WARNING_DELAY)) {
NexmarkUtils.console("WARNING: streaming query appears to have been
stuck for %d min.",
quietFor.getStandardMinutes());
@@ -543,7 +546,7 @@ private NexmarkPerf monitor(NexmarkQuery query) {
String.format("Streaming query was stuck for %d min",
quietFor.getStandardMinutes()));
}
- if (waitingForShutdown) {
+ if (cancelJob) {
try {
job.cancel();
} catch (IOException e) {
@@ -567,7 +570,7 @@ private NexmarkPerf monitor(NexmarkQuery query) {
break;
case CANCELLED:
running = false;
- if (!waitingForShutdown) {
+ if (!cancelJob) {
errors.add("Job was unexpectedly cancelled");
}
break;
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> NexmarkLauncher should not cancel jobs once the input events has been reached
> -----------------------------------------------------------------------------
>
> Key: BEAM-3464
> URL: https://issues.apache.org/jira/browse/BEAM-3464
> Project: Beam
> Issue Type: Bug
> Components: testing
> Reporter: Sam Whittle
> Assignee: Sam Whittle
> Priority: Trivial
>
> NexmarkLauncher monitors the nexmark job and cancels it in several cases: if
> it gets stuck, has fatal errors or the input processed counter equals the
> desired amount.
> Cancelling based upon the processed counter is problematic for the following
> reasons:
> - Dataflow (and other runners?) streaming counters are non-determinisitic and
> thus the cancellation is deterministic
> - The event counter is from the beginning stage of the pipeline and the
> pipeline may have internally buffered data
> This cancellation is also unnessary as once all the events are processed, the
> nexmark source watermark goes to infinity and the pipeline will terminate due
> to all watermarks going to infinity after it completes draining.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
