You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2016/11/10 18:49:41 UTC
[1/6] incubator-beam git commit: This closes #1320
Repository: incubator-beam
Updated Branches:
refs/heads/master ab06647f9 -> 11eaed19a
This closes #1320
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/11eaed19
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/11eaed19
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/11eaed19
Branch: refs/heads/master
Commit: 11eaed19a471ecc0c3fa1a57fb14c93ab21437a7
Parents: ab06647 bf39dc6
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Nov 10 10:49:28 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Nov 10 10:49:28 2016 -0800
----------------------------------------------------------------------
.../beam/examples/WindowedWordCountIT.java | 14 ++++++++++++--
.../dataflow/testing/TestDataflowRunner.java | 20 ++++++++++++++------
.../beam/sdk/testing/TestPipelineOptions.java | 6 ++++++
3 files changed, 32 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
[3/6] incubator-beam git commit: Fix breakage in WindowedWordCountIT
Posted by ke...@apache.org.
Fix breakage in WindowedWordCountIT
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e5b2f13f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e5b2f13f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e5b2f13f
Branch: refs/heads/master
Commit: e5b2f13f0116b0cd954b2214ca33cbf5a46a7261
Parents: ab06647
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Nov 9 10:22:51 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Nov 10 10:49:28 2016 -0800
----------------------------------------------------------------------
.../org/apache/beam/examples/WindowedWordCountIT.java | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e5b2f13f/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
----------------------------------------------------------------------
diff --git a/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java b/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
index c78fad6..ccc2d5e 100644
--- a/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
+++ b/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
@@ -18,13 +18,14 @@
package org.apache.beam.examples;
import java.io.IOException;
-import org.apache.beam.examples.WindowedWordCount.Options;
+import java.util.Date;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.testing.BigqueryMatcher;
import org.apache.beam.sdk.testing.StreamingIT;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.TestPipelineOptions;
+import org.apache.beam.sdk.util.IOChannelUtils;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -43,7 +44,7 @@ public class WindowedWordCountIT {
* Options for the {@link WindowedWordCount} Integration Test.
*/
public interface WindowedWordCountITOptions
- extends Options, TestPipelineOptions, StreamingOptions {
+ extends WindowedWordCount.Options, TestPipelineOptions, StreamingOptions {
}
@BeforeClass
@@ -67,6 +68,14 @@ public class WindowedWordCountIT {
TestPipeline.testingPipelineOptions().as(WindowedWordCountITOptions.class);
options.setStreaming(isStreaming);
+ // Note: currently unused because the example writes to BigQuery, but WindowedWordCount.Options
+ // are tightly coupled to WordCount.Options, where the option is required.
+ options.setOutput(IOChannelUtils.resolve(
+ options.getTempRoot(),
+ String.format("WindowedWordCountIT-%tF-%<tH-%<tM-%<tS-%<tL", new Date()),
+ "output",
+ "results"));
+
String query = String.format("SELECT word, SUM(count) FROM [%s:%s.%s] GROUP BY word",
options.getProject(), options.getBigQueryDataset(), options.getBigQueryTable());
options.setOnSuccessMatcher(
[6/6] incubator-beam git commit: Make TestDataflowRunner crash
message actionable
Posted by ke...@apache.org.
Make TestDataflowRunner crash message actionable
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/5653b860
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/5653b860
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/5653b860
Branch: refs/heads/master
Commit: 5653b860dbe64f61707c093fff5bf14061e772d0
Parents: 7344914
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Nov 9 14:36:54 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Nov 10 10:49:28 2016 -0800
----------------------------------------------------------------------
.../apache/beam/runners/dataflow/testing/TestDataflowRunner.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5653b860/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
index 9dacfd3..10c72b7 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
@@ -152,7 +152,9 @@ public class TestDataflowRunner extends PipelineRunner<DataflowPipelineJob> {
} else if (!success.get()) {
throw new AssertionError(
Strings.isNullOrEmpty(messageHandler.getErrorMessage())
- ? "The dataflow did not return a failure reason."
+ ? String.format(
+ "Dataflow job %s terminated in state %s but did not return a failure reason.",
+ job.getJobId(), job.getState())
: messageHandler.getErrorMessage());
} else {
assertThat(job, testPipelineOptions.getOnSuccessMatcher());
[2/6] incubator-beam git commit: Make test timeout configurable and
use in TestDataflowRunner
Posted by ke...@apache.org.
Make test timeout configurable and use in TestDataflowRunner
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7e82e35b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7e82e35b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7e82e35b
Branch: refs/heads/master
Commit: 7e82e35b2476db6b2b491861e5c2c52042ce2161
Parents: 5653b86
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Nov 9 19:24:44 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Nov 10 10:49:28 2016 -0800
----------------------------------------------------------------------
.../beam/runners/dataflow/testing/TestDataflowRunner.java | 10 +++++++---
.../org/apache/beam/sdk/testing/TestPipelineOptions.java | 6 ++++++
2 files changed, 13 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7e82e35b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
index 10c72b7..70c3f58 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
@@ -135,10 +135,14 @@ public class TestDataflowRunner extends PipelineRunner<DataflowPipelineJob> {
}
}
});
- State finalState = job.waitUntilFinish(Duration.standardMinutes(10L), messageHandler);
+ State finalState =
+ job.waitUntilFinish(
+ Duration.standardSeconds(options.getTestTimeoutSeconds()), messageHandler);
if (finalState == null || finalState == State.RUNNING) {
- LOG.info("Dataflow job {} took longer than 10 minutes to complete, cancelling.",
- job.getJobId());
+ LOG.info(
+ "Dataflow job {} took longer than {} seconds to complete, cancelling.",
+ job.getJobId(),
+ options.getTestTimeoutSeconds());
job.cancel();
}
success = resultFuture.get();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7e82e35b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipelineOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipelineOptions.java
index ff553ba..0739381 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipelineOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipelineOptions.java
@@ -17,6 +17,7 @@
*/
package org.apache.beam.sdk.testing;
+import javax.annotation.Nullable;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.DefaultValueFactory;
@@ -41,6 +42,11 @@ public interface TestPipelineOptions extends PipelineOptions {
SerializableMatcher<PipelineResult> getOnSuccessMatcher();
void setOnSuccessMatcher(SerializableMatcher<PipelineResult> value);
+ @Default.Long(10 * 60)
+ @Nullable
+ Long getTestTimeoutSeconds();
+ void setTestTimeoutSeconds(Long value);
+
/**
* Factory for {@link PipelineResult} matchers which always pass.
*/
[5/6] incubator-beam git commit: Increase WindowedWordCountIT timeout
to 20 minutes
Posted by ke...@apache.org.
Increase WindowedWordCountIT timeout to 20 minutes
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/bf39dc65
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/bf39dc65
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/bf39dc65
Branch: refs/heads/master
Commit: bf39dc6509e307a516e9d23dadceb17c78811c12
Parents: 7e82e35
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Nov 9 19:26:46 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Nov 10 10:49:28 2016 -0800
----------------------------------------------------------------------
.../src/test/java/org/apache/beam/examples/WindowedWordCountIT.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/bf39dc65/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
----------------------------------------------------------------------
diff --git a/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java b/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
index ccc2d5e..a5113c8 100644
--- a/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
+++ b/examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java
@@ -67,6 +67,7 @@ public class WindowedWordCountIT {
WindowedWordCountITOptions options =
TestPipeline.testingPipelineOptions().as(WindowedWordCountITOptions.class);
options.setStreaming(isStreaming);
+ options.setTestTimeoutSeconds(1200L);
// Note: currently unused because the example writes to BigQuery, but WindowedWordCount.Options
// are tightly coupled to WordCount.Options, where the option is required.
[4/6] incubator-beam git commit: Fix null or empty check in
TestDataflowRunner
Posted by ke...@apache.org.
Fix null or empty check in TestDataflowRunner
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/73449146
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/73449146
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/73449146
Branch: refs/heads/master
Commit: 7344914638953a78251d021f4001e404e8c1aff1
Parents: e5b2f13
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Nov 9 13:39:57 2016 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Nov 10 10:49:28 2016 -0800
----------------------------------------------------------------------
.../beam/runners/dataflow/testing/TestDataflowRunner.java | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/73449146/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
index 0f141d2..9dacfd3 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java
@@ -25,6 +25,7 @@ import com.google.api.services.dataflow.model.MetricUpdate;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Optional;
+import com.google.common.base.Strings;
import com.google.common.base.Throwables;
import java.io.IOException;
import java.math.BigDecimal;
@@ -149,9 +150,10 @@ public class TestDataflowRunner extends PipelineRunner<DataflowPipelineJob> {
throw new IllegalStateException(
"The dataflow did not output a success or failure metric.");
} else if (!success.get()) {
- throw new AssertionError(messageHandler.getErrorMessage() == null
- ? "The dataflow did not return a failure reason."
- : messageHandler.getErrorMessage());
+ throw new AssertionError(
+ Strings.isNullOrEmpty(messageHandler.getErrorMessage())
+ ? "The dataflow did not return a failure reason."
+ : messageHandler.getErrorMessage());
} else {
assertThat(job, testPipelineOptions.getOnSuccessMatcher());
}