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/12/21 22:49:42 UTC
[08/51] [abbrv] incubator-beam git commit: Migrated the
beam-sdks-java-io-google-cloud-platform module to TestPipeline as a JUnit
rule.
Migrated the beam-sdks-java-io-google-cloud-platform module to TestPipeline as a JUnit rule.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/6dea0992
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/6dea0992
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/6dea0992
Branch: refs/heads/python-sdk
Commit: 6dea0992d9976b39232cf846906831feaa25ec43
Parents: 63331aa
Author: Stas Levin <st...@gmail.com>
Authored: Tue Dec 20 13:26:07 2016 +0200
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Dec 20 09:55:45 2016 -0800
----------------------------------------------------------------------
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 25 +++++++++++++-------
.../sdk/io/gcp/bigtable/BigtableIOTest.java | 13 ++++++----
2 files changed, 25 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6dea0992/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
index dc566d2..4ddfdea 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
@@ -643,6 +643,7 @@ public class BigQueryIOTest implements Serializable {
}
}
+ @Rule public final transient TestPipeline p = TestPipeline.create();
@Rule public transient ExpectedException thrown = ExpectedException.none();
@Rule public transient ExpectedLogs logged = ExpectedLogs.none(BigQueryIO.class);
@Rule public transient TemporaryFolder testFolder = new TemporaryFolder();
@@ -1370,7 +1371,7 @@ public class BigQueryIOTest implements Serializable {
@Test
@Category(NeedsRunner.class)
public void testBuildWriteWithoutTable() {
- Pipeline p = TestPipeline.create();
+
thrown.expect(IllegalStateException.class);
thrown.expectMessage("must set the table reference");
p.apply(Create.<TableRow>of().withCoder(TableRowJsonCoder.of()))
@@ -1591,9 +1592,11 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testWriteValidateFailsCreateNoSchema() {
+ p.enableAbandonedNodeEnforcement(false);
+
thrown.expect(IllegalArgumentException.class);
thrown.expectMessage("no schema was provided");
- TestPipeline.create()
+ p
.apply(Create.<TableRow>of())
.apply(BigQueryIO.Write
.to("dataset.table")
@@ -1602,9 +1605,11 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testWriteValidateFailsTableAndTableSpec() {
+ p.enableAbandonedNodeEnforcement(false);
+
thrown.expect(IllegalStateException.class);
thrown.expectMessage("Cannot set both a table reference and a table function");
- TestPipeline.create()
+ p
.apply(Create.<TableRow>of())
.apply(BigQueryIO.Write
.to("dataset.table")
@@ -1618,9 +1623,11 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testWriteValidateFailsNoTableAndNoTableSpec() {
+ p.enableAbandonedNodeEnforcement(false);
+
thrown.expect(IllegalStateException.class);
thrown.expectMessage("must set the table reference of a BigQueryIO.Write transform");
- TestPipeline.create()
+ p
.apply(Create.<TableRow>of())
.apply("name", BigQueryIO.Write.withoutValidation());
}
@@ -1950,7 +1957,6 @@ public class BigQueryIOTest implements Serializable {
@Test
@Category(NeedsRunner.class)
public void testPassThroughThenCleanup() throws Exception {
- Pipeline p = TestPipeline.create();
PCollection<Integer> output = p
.apply(Create.of(1, 2, 3))
@@ -1968,7 +1974,6 @@ public class BigQueryIOTest implements Serializable {
@Test
@Category(NeedsRunner.class)
public void testPassThroughThenCleanupExecuted() throws Exception {
- Pipeline p = TestPipeline.create();
p.apply(Create.<Integer>of())
.apply(new PassThroughThenCleanup<Integer>(new CleanupOperation() {
@@ -2025,6 +2030,8 @@ public class BigQueryIOTest implements Serializable {
private void testWritePartition(long numFiles, long fileSize, long expectedNumPartitions)
throws Exception {
+ p.enableAbandonedNodeEnforcement(false);
+
List<Long> expectedPartitionIds = Lists.newArrayList();
for (long i = 1; i <= expectedNumPartitions; ++i) {
expectedPartitionIds.add(i);
@@ -2044,7 +2051,7 @@ public class BigQueryIOTest implements Serializable {
new TupleTag<KV<Long, List<String>>>("singlePartitionTag") {};
PCollectionView<Iterable<KV<String, Long>>> filesView = PCollectionViews.iterableView(
- TestPipeline.create(),
+ p,
WindowingStrategy.globalDefault(),
KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()));
@@ -2164,6 +2171,8 @@ public class BigQueryIOTest implements Serializable {
@Test
public void testWriteRename() throws Exception {
+ p.enableAbandonedNodeEnforcement(false);
+
FakeBigQueryServices fakeBqServices = new FakeBigQueryServices()
.withJobService(new FakeJobService()
.startJobReturns("done", "done")
@@ -2179,7 +2188,7 @@ public class BigQueryIOTest implements Serializable {
}
PCollectionView<Iterable<String>> tempTablesView = PCollectionViews.iterableView(
- TestPipeline.create(),
+ p,
WindowingStrategy.globalDefault(),
StringUtf8Coder.of());
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6dea0992/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
index 98215df..e3d1f47 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIOTest.java
@@ -106,6 +106,7 @@ import org.junit.runners.JUnit4;
*/
@RunWith(JUnit4.class)
public class BigtableIOTest {
+ @Rule public final transient TestPipeline p = TestPipeline.create();
@Rule public ExpectedException thrown = ExpectedException.none();
@Rule public ExpectedLogs logged = ExpectedLogs.none(BigtableIO.class);
@@ -140,7 +141,7 @@ public class BigtableIOTest {
service = new FakeBigtableService();
defaultRead = defaultRead.withBigtableService(service);
defaultWrite = defaultWrite.withBigtableService(service);
- bigtableCoder = TestPipeline.create().getCoderRegistry().getCoder(BIGTABLE_WRITE_TYPE);
+ bigtableCoder = p.getCoderRegistry().getCoder(BIGTABLE_WRITE_TYPE);
}
@Test
@@ -261,6 +262,8 @@ public class BigtableIOTest {
/** Tests that when reading from a non-existent table, the read fails. */
@Test
public void testReadingFailsTableDoesNotExist() throws Exception {
+ p.enableAbandonedNodeEnforcement(false);
+
final String table = "TEST-TABLE";
BigtableIO.Read read =
@@ -273,7 +276,7 @@ public class BigtableIOTest {
thrown.expect(IllegalArgumentException.class);
thrown.expectMessage(String.format("Table %s does not exist", table));
- TestPipeline.create().apply(read);
+ p.apply(read);
}
/** Tests that when reading from an empty table, the read succeeds. */
@@ -589,7 +592,6 @@ public class BigtableIOTest {
service.createTable(table);
- TestPipeline p = TestPipeline.create();
p.apply("single row", Create.of(makeWrite(key, value)).withCoder(bigtableCoder))
.apply("write", defaultWrite.withTableId(table));
p.run();
@@ -606,10 +608,12 @@ public class BigtableIOTest {
/** Tests that when writing to a non-existent table, the write fails. */
@Test
public void testWritingFailsTableDoesNotExist() throws Exception {
+ p.enableAbandonedNodeEnforcement(false);
+
final String table = "TEST-TABLE";
PCollection<KV<ByteString, Iterable<Mutation>>> emptyInput =
- TestPipeline.create().apply(Create.<KV<ByteString, Iterable<Mutation>>>of());
+ p.apply(Create.<KV<ByteString, Iterable<Mutation>>>of());
// Exception will be thrown by write.validate() when write is applied.
thrown.expect(IllegalArgumentException.class);
@@ -625,7 +629,6 @@ public class BigtableIOTest {
final String key = "KEY";
service.createTable(table);
- TestPipeline p = TestPipeline.create();
p.apply(Create.of(makeBadWrite(key)).withCoder(bigtableCoder))
.apply(defaultWrite.withTableId(table));