BigQueryOutput extends TaskOutput<StagingTableId, TableId> { public static BigQueryOutput create(String project, String dataset, String table) { return create(TableId.of(project, dataset, table)); } BigQueryOutput(F0<FloBigQueryClient> bigQuery, TableId tableId); static BigQueryOutput create(String project, String dataset, String table); static BigQueryOutput create(TableId tableId); TableId tableId(); @Override StagingTableId provide(EvalContext evalContext); @Override Optional<TableId> lookup(Task<TableId> task); static BigQueryMocking mock();  }
@Test public void shouldBeSerializable() throws SerializationException { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); final BigQueryOutput context = BigQueryOutput.create("foo", "bar", "baz"); Serialization.serialize(context, baos); final ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray()); final BigQueryOutput deserializedContext = Serialization.deserialize(bais); assertThat(deserializedContext, is(notNullValue())); }
@Test public void shouldBeRunnable() throws Exception { final String nonExistentProject = UUID.randomUUID().toString(); final Task<TableId> task = Task.named("task").ofType(TableId.class) .output(BigQueryOutput.create(nonExistentProject, "foo", "bar")) .process(StagingTableId::tableId); final Future<TableId> future = FloRunner.runTask(task).future(); try { future.get(30, TimeUnit.SECONDS); } catch (ExecutionException e) { final Throwable rootCause = Throwables.getRootCause(e); if (rootCause instanceof GoogleJsonResponseException) { } else if (rootCause instanceof IllegalArgumentException && rootCause.getMessage().startsWith("A project ID is required")) { } else if (rootCause instanceof IllegalArgumentException && rootCause.getMessage().startsWith("Dataset does not exist.")) { } else if (rootCause instanceof BigQueryException && rootCause.getMessage().equals("The project " + nonExistentProject + " has not enabled BigQuery.")) { } else { throw new AssertionError("Unknown error, might be serialization problem that needs fixing?", e); } } }