beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <>
Subject [jira] [Commented] (BEAM-3464) NexmarkLauncher should not cancel jobs once the input events has been reached
Date Thu, 18 Jan 2018 23:09:00 GMT


ASF GitHub Bot commented on BEAM-3464:

kennknowles closed pull request #4398: [ BEAM-3464] Fix NexmarkLauncher to not cancel jobs
otherwise successful jobs

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/
index a4b5c5aaf5c..1bdd5f5b7ed 100644
--- a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/
+++ b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/
@@ -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
-          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
@@ -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 {
           } catch (IOException e) {
@@ -567,7 +570,7 @@ private NexmarkPerf monitor(NexmarkQuery query) {
         case CANCELLED:
           running = false;
-          if (!waitingForShutdown) {
+          if (!cancelJob) {
             errors.add("Job was unexpectedly cancelled");


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:

> NexmarkLauncher should not cancel jobs once the input events has been reached
> -----------------------------------------------------------------------------
>                 Key: BEAM-3464
>                 URL:
>             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

View raw message