Skip to content

Commit 9ba2701

Browse files
mziccardmziccard
committed
Add blocking waitFor method, remove whenDone and callback
1 parent e29f635 commit 9ba2701

15 files changed

Lines changed: 333 additions & 456 deletions

File tree

‎README.md‎

Lines changed: 12 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,6 @@ Complete source code can be found at
145145
146146
```java
147147
import com.google.cloud.bigquery.BigQuery;
148-
import com.google.cloud.bigquery.BigQueryError;
149148
import com.google.cloud.bigquery.BigQueryOptions;
150149
import com.google.cloud.bigquery.Field;
151150
import com.google.cloud.bigquery.FormatOptions;
@@ -156,8 +155,6 @@ import com.google.cloud.bigquery.Table;
156155
import com.google.cloud.bigquery.TableId;
157156
import com.google.cloud.bigquery.TableInfo;
158157
159-
import java.util.List;
160-
161158
BigQuery bigquery = BigQueryOptions.defaultInstance().service();
162159
TableId tableId = TableId.of("dataset", "table");
163160
Table table = bigquery.getTable(tableId);
@@ -169,17 +166,12 @@ if (table == null) {
169166
}
170167
System.out.println("Loading data into table " + tableId);
171168
Job loadJob = table.load(FormatOptions.csv(), "gs://bucket/path");
172-
loadJob.whenDone(new Job.CompletionCallback() {
173-
@Override
174-
public void success(Job job) {
175-
System.out.println("Job succeeded");
176-
}
177-
178-
@Override
179-
public void error(BigQueryError error, List<BigQueryError> executionErrors) {
180-
System.out.println("Job completed with errors");
181-
}
182-
});
169+
loadJob = loadJob.waitFor();
170+
if (loadJob.status().error() != null) {
171+
System.out.println("Job completed with errors");
172+
} else {
173+
System.out.println("Job succeeded");
174+
}
183175
```
184176
185177
Google Cloud Compute (Alpha)
@@ -202,31 +194,19 @@ import com.google.cloud.compute.Compute;
202194
import com.google.cloud.compute.ComputeOptions;
203195
import com.google.cloud.compute.Disk;
204196
import com.google.cloud.compute.DiskId;
205-
import com.google.cloud.compute.Operation.OperationError;
206-
import com.google.cloud.compute.Operation.OperationWarning;
207197
import com.google.cloud.compute.Snapshot;
208198
209-
import java.util.List;
210-
211199
final Compute compute = ComputeOptions.defaultInstance().service();
212200
DiskId diskId = DiskId.of("us-central1-a", "disk-name");
213201
Disk disk = compute.getDisk(diskId, Compute.DiskOption.fields());
214202
if (disk != null) {
215203
final String snapshotName = "disk-name-snapshot";
216204
Operation operation = disk.createSnapshot(snapshotName);
217-
operation.whenDone(new Operation.CompletionCallback() {
218-
@Override
219-
public void success(Operation operation) {
220-
// use snapshot
221-
Snapshot snapshot = compute.getSnapshot(snapshotName);
222-
}
223-
224-
@Override
225-
public void error(List<OperationError> errors, List<OperationWarning> warnings) {
226-
// inspect errors
227-
throw new RuntimeException("Snaphsot creation failed");
228-
}
229-
});
205+
operation = operation.waitFor();
206+
if (operation.errors() == null) {
207+
// use snapshot
208+
Snapshot snapshot = compute.getSnapshot(snapshotName);
209+
}
230210
}
231211
```
232212
The second snippet shows how to create a virtual machine instance. Complete source code can be found
@@ -242,10 +222,6 @@ import com.google.cloud.compute.InstanceId;
242222
import com.google.cloud.compute.InstanceInfo;
243223
import com.google.cloud.compute.MachineTypeId;
244224
import com.google.cloud.compute.NetworkId;
245-
import com.google.cloud.compute.Operation.OperationError;
246-
import com.google.cloud.compute.Operation.OperationWarning;
247-
248-
import java.util.List;
249225
250226
Compute compute = ComputeOptions.defaultInstance().service();
251227
ImageId imageId = ImageId.of("debian-cloud", "debian-8-jessie-v20160329");
@@ -256,9 +232,7 @@ InstanceId instanceId = InstanceId.of("us-central1-a", "instance-name");
256232
MachineTypeId machineTypeId = MachineTypeId.of("us-central1-a", "n1-standard-1");
257233
Operation operation =
258234
compute.create(InstanceInfo.of(instanceId, machineTypeId, attachedDisk, networkInterface));
259-
while (!operation.isDone()) {
260-
Thread.sleep(1000L);
261-
}
235+
operation = operation.waitFor();
262236
if (operation.errors() == null) {
263237
// use instance
264238
Instance instance = compute.getInstance(instanceId);

‎gcloud-java-bigquery/src/main/java/com/google/cloud/bigquery/Job.java‎

Lines changed: 37 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020

2121
import java.io.IOException;
2222
import java.io.ObjectInputStream;
23-
import java.util.List;
2423
import java.util.Objects;
24+
import java.util.concurrent.TimeUnit;
2525

2626
/**
2727
* A Google BigQuery Job.
@@ -38,24 +38,6 @@ public class Job extends JobInfo {
3838
private final BigQueryOptions options;
3939
private transient BigQuery bigquery;
4040

41-
/**
42-
* A callback for job completion.
43-
*/
44-
public interface CompletionCallback {
45-
/**
46-
* The method called when the job completes successfully.
47-
*/
48-
void success(Job job);
49-
50-
/**
51-
* The method called when the job completes with errors. {@code error} is the final error that
52-
* caused the job to fail (see {@link JobStatus#error()}). {@code executionErrors} are all the
53-
* errors (possibly not fatal) encountered by the job during its execution (see
54-
* {@link JobStatus#executionErrors()}).
55-
*/
56-
void error(BigQueryError error, List<BigQueryError> executionErrors);
57-
}
58-
5941
/**
6042
* A builder for {@code Job} objects.
6143
*/
@@ -163,40 +145,51 @@ public boolean isDone() {
163145
}
164146

165147
/**
166-
* Waits until this job completes its execution, either failing or succeeding. If the job does not
167-
* exist, this method returns without executing any method of the provided callback. If the job
168-
* completed successfully the {@link CompletionCallback#success(Job)} method is called. If the job
169-
* completed with errors the {@link CompletionCallback#error(BigQueryError, List)} method is
170-
* called.
148+
* Blocks until this job completes its execution, either failing or succeeding. The job status is
149+
* checked every 500 milliseconds. This method returns current job's latest information. If the
150+
* job no longer exists, this method returns {@code null}.
171151
* <pre> {@code
172-
* job.whenDone(new CompletionCallback() {
173-
* void success(Job job) {
174-
* // completed successfully
175-
* }
152+
* Job completedJob = job.waitFor();
153+
* if (completedJob == null) {
154+
* // job no longer exists
155+
* } else if (completedJob.status().error() != null) {
156+
* // job failed, handle error
157+
* } else {
158+
* // job completed successfully
159+
* }}</pre>
176160
*
177-
* void error(BigQueryError error, List<BigQueryError> executionErrors) {
178-
* // handle error
179-
* }
180-
* });}</pre>
161+
* @throws BigQueryException upon failure
162+
* @throws InterruptedException if the current thread gets interrupted while waiting for the job
163+
* to complete
164+
*/
165+
public Job waitFor() throws InterruptedException {
166+
return waitFor(500, TimeUnit.MILLISECONDS);
167+
}
168+
169+
/**
170+
* Blocks until this job completes its execution, either failing or succeeding. The
171+
* {@code checkEvery} and {@code unit} parameters determine how often the job's status is checked.
172+
* This method returns current job's latest information. If the job no longer exists, this method
173+
* returns {@code null}.
174+
* <pre> {@code
175+
* Job completedJob = job.waitFor(1, TimeUnit.SECONDS);
176+
* if (completedJob == null) {
177+
* // job no longer exists
178+
* } else if (completedJob.status().error() != null) {
179+
* // job failed, handle error
180+
* } else {
181+
* // job completed successfully
182+
* }}</pre>
181183
*
182184
* @throws BigQueryException upon failure
183185
* @throws InterruptedException if the current thread gets interrupted while waiting for the job
184186
* to complete
185187
*/
186-
public void whenDone(CompletionCallback callback) throws InterruptedException {
188+
public Job waitFor(int checkEvery, TimeUnit unit) throws InterruptedException {
187189
while (!isDone()) {
188-
Thread.sleep(500L);
189-
}
190-
Job updatedJob = reload();
191-
if (updatedJob == null) {
192-
return;
193-
}
194-
BigQueryError error = updatedJob.status().error();
195-
if (error != null) {
196-
callback.error(error, updatedJob.status().executionErrors());
197-
} else {
198-
callback.success(updatedJob);
190+
unit.sleep(checkEvery);
199191
}
192+
return reload();
200193
}
201194

202195
/**

‎gcloud-java-bigquery/src/main/java/com/google/cloud/bigquery/package-info.java‎

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -33,15 +33,12 @@
3333
* }
3434
* System.out.println("Loading data into table " + tableId);
3535
* Job loadJob = table.load(FormatOptions.csv(), "gs://bucket/path");
36-
* loadJob.whenDone(new Job.CompletionCallback() {
37-
* public void success(Job job) {
38-
* System.out.println("Job succeeded");
39-
* }
40-
*
41-
* public void error(BigQueryError error, List<BigQueryError> executionErrors) {
42-
* System.out.println("Job completed with errors");
43-
* }
44-
* });}</pre>
36+
* loadJob = loadJob.waitFor();
37+
* if (loadJob.status().error() != null) {
38+
* System.out.println("Job completed with errors");
39+
* } else {
40+
* System.out.println("Job succeeded");
41+
* }}</pre>
4542
*
4643
* @see <a href="https://cloud.google.com/bigquery/">Google Cloud BigQuery</a>
4744
*/

‎gcloud-java-bigquery/src/test/java/com/google/cloud/bigquery/JobTest.java‎

Lines changed: 38 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -28,13 +28,12 @@
2828
import static org.junit.Assert.assertTrue;
2929

3030
import com.google.cloud.bigquery.JobStatistics.CopyStatistics;
31-
import com.google.common.collect.ImmutableList;
3231

3332
import org.easymock.EasyMock;
3433
import org.junit.After;
3534
import org.junit.Test;
3635

37-
import java.util.List;
36+
import java.util.concurrent.TimeUnit;
3837

3938
public class JobTest {
4039

@@ -182,60 +181,71 @@ public void testIsDone_NotExists() throws Exception {
182181
}
183182

184183
@Test
185-
public void testWhenDone_Success() throws InterruptedException {
184+
public void testWaitFor() throws InterruptedException {
186185
initializeExpectedJob(2);
187-
Job.CompletionCallback callback = EasyMock.mock(Job.CompletionCallback.class);
188186
BigQuery.JobOption[] expectedOptions = {BigQuery.JobOption.fields(BigQuery.JobField.STATUS)};
189187
JobStatus status = createStrictMock(JobStatus.class);
190188
expect(status.state()).andReturn(JobStatus.State.DONE);
191-
expect(status.error()).andReturn(null);
192189
expect(bigquery.options()).andReturn(mockOptions);
193190
Job completedJob = expectedJob.toBuilder().status(status).build();
194191
expect(bigquery.getJob(JOB_INFO.jobId(), expectedOptions)).andReturn(completedJob);
195192
expect(bigquery.getJob(JOB_INFO.jobId())).andReturn(completedJob);
196-
callback.success(completedJob);
197-
EasyMock.expectLastCall();
198-
replay(status, bigquery, callback);
193+
replay(status, bigquery);
199194
initializeJob();
200-
job.whenDone(callback);
201-
verify(status, callback);
195+
assertSame(completedJob, job.waitFor());
196+
verify(status);
202197
}
203198

204199
@Test
205-
public void testWhenDone_Error() throws InterruptedException {
206-
initializeExpectedJob(2);
207-
Job.CompletionCallback callback = EasyMock.mock(Job.CompletionCallback.class);
200+
public void testWaitFor_Null() throws InterruptedException {
201+
initializeExpectedJob(1);
202+
BigQuery.JobOption[] expectedOptions = {BigQuery.JobOption.fields(BigQuery.JobField.STATUS)};
203+
expect(bigquery.options()).andReturn(mockOptions);
204+
expect(bigquery.getJob(JOB_INFO.jobId(), expectedOptions)).andReturn(null);
205+
expect(bigquery.getJob(JOB_INFO.jobId())).andReturn(null);
206+
replay(bigquery);
207+
initializeJob();
208+
assertNull(job.waitFor());
209+
}
210+
211+
@Test
212+
public void testWaitForWithTimeUnit() throws InterruptedException {
213+
initializeExpectedJob(3);
208214
BigQuery.JobOption[] expectedOptions = {BigQuery.JobOption.fields(BigQuery.JobField.STATUS)};
209-
BigQueryError error = new BigQueryError("reason", "location", "message");
210-
List<BigQueryError> executionErrors = ImmutableList.of(error);
215+
TimeUnit timeUnit = createStrictMock(TimeUnit.class);
216+
timeUnit.sleep(42);
217+
EasyMock.expectLastCall();
211218
JobStatus status = createStrictMock(JobStatus.class);
219+
expect(status.state()).andReturn(JobStatus.State.RUNNING);
212220
expect(status.state()).andReturn(JobStatus.State.DONE);
213-
expect(status.error()).andReturn(error);
214-
expect(status.executionErrors()).andReturn(executionErrors);
215221
expect(bigquery.options()).andReturn(mockOptions);
222+
Job runningJob = expectedJob.toBuilder().status(status).build();
216223
Job completedJob = expectedJob.toBuilder().status(status).build();
224+
expect(bigquery.getJob(JOB_INFO.jobId(), expectedOptions)).andReturn(runningJob);
217225
expect(bigquery.getJob(JOB_INFO.jobId(), expectedOptions)).andReturn(completedJob);
218226
expect(bigquery.getJob(JOB_INFO.jobId())).andReturn(completedJob);
219-
callback.error(error, executionErrors);
220-
EasyMock.expectLastCall();
221-
replay(status, bigquery, callback);
227+
replay(status, bigquery, timeUnit);
222228
initializeJob();
223-
job.whenDone(callback);
224-
verify(status, callback);
229+
assertSame(completedJob, job.waitFor(42, timeUnit));
230+
verify(status, timeUnit);
225231
}
226232

227233
@Test
228-
public void testWhenDone_Null() throws InterruptedException {
229-
initializeExpectedJob(1);
230-
Job.CompletionCallback callback = EasyMock.mock(Job.CompletionCallback.class);
234+
public void testWaitForWithTimeUnit_Null() throws InterruptedException {
235+
initializeExpectedJob(2);
231236
BigQuery.JobOption[] expectedOptions = {BigQuery.JobOption.fields(BigQuery.JobField.STATUS)};
237+
TimeUnit timeUnit = createStrictMock(TimeUnit.class);
238+
timeUnit.sleep(42);
239+
EasyMock.expectLastCall();
232240
expect(bigquery.options()).andReturn(mockOptions);
241+
Job runningJob = expectedJob.toBuilder().status(new JobStatus(JobStatus.State.RUNNING)).build();
242+
expect(bigquery.getJob(JOB_INFO.jobId(), expectedOptions)).andReturn(runningJob);
233243
expect(bigquery.getJob(JOB_INFO.jobId(), expectedOptions)).andReturn(null);
234244
expect(bigquery.getJob(JOB_INFO.jobId())).andReturn(null);
235-
replay(bigquery, callback);
245+
replay(bigquery, timeUnit);
236246
initializeJob();
237-
job.whenDone(callback);
238-
verify(callback);
247+
assertNull(job.waitFor(42, timeUnit));
248+
verify(bigquery, timeUnit);
239249
}
240250

241251
@Test

0 commit comments

Comments
 (0)