github-actions[bot] commented on code in PR #41329:
URL: https://github.com/apache/beam/pull/41329#discussion_r4572291373
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java:
##########
@@ -529,7 +532,14 @@ private PendingJob startLoad(
jobRef,
jobId.getRetryIndex());
try {
- jobService.startLoadJob(jobRef, loadConfig);
+ if (loadJobReservation == null) {
+ jobService.startLoadJob(jobRef, loadConfig);
+ } else {
+ String reservation = loadJobReservation.get();
+ checkArgument(
Review Comment:
**🟠High**: In pipeline templates, an optional `ValueProvider<String>`
parameter is passed to
`.withLoadJobReservation(options.getLoadJobReservation())` at graph
construction time (so `loadJobReservation` is a non-null
`RuntimeValueProvider`), and `loadJobReservation.get()` returns `null` (or
`""`) at runtime when the user omits the optional parameter. Throwing
`IllegalArgumentException` here breaks templates whenever the reservation
parameter is left unset at runtime.
Consistent with `loadJobProjectId` (lines 514-517) and
`customGcsTempLocation`, if `loadJobReservation == null ||
Strings.isNullOrEmpty(loadJobReservation.get())`, this should fall back to
`jobService.startLoadJob(jobRef, loadConfig)` without a reservation.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeJobService.java:
##########
@@ -187,6 +198,19 @@ public void startLoadJob(
// TODO
Review Comment:
**🟢 Low**: Now that the 4-argument `startLoadJob(jobRef, loadConfig,
streamContent, reservation)` is implemented below, this 3-argument overload can
delegate to it with `null` (just like `startLoadJob(jobRef, loadConfig)` on
line 156) so schema-update zero-load jobs without a reservation also work in
`FakeJobService`.
```suggestion
startLoadJob(jobRef, loadConfig, streamContent, null);
```
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java:
##########
@@ -318,8 +344,16 @@ public void finishBundle(FinishBundleContext context)
throws Exception {
jobRef,
jobId.getRetryIndex());
try {
- jobService.startLoadJob(
- jobRef, loadConfig, new ByteArrayContent("text/plain", new
byte[0]));
+ ByteArrayContent content = new ByteArrayContent("text/plain",
new byte[0]);
+ if (loadJobReservation == null) {
+ jobService.startLoadJob(jobRef, loadConfig, content);
+ } else {
+ String reservation = loadJobReservation.get();
+ if (Strings.isNullOrEmpty(reservation)) {
Review Comment:
**🟠High**: Similar to `WriteTables`, if `loadJobReservation` is a
`ValueProvider<String>` for an optional template parameter that is not provided
at runtime, `loadJobReservation.get()` will evaluate to `null` (or `""`).
Rather than throwing `IllegalArgumentException`, treat a `null` or empty value
from `loadJobReservation.get()` as unset and call
`jobService.startLoadJob(jobRef, loadConfig, content)` (matching
`loadJobProjectId` on lines 328-331).
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java:
##########
@@ -308,6 +308,57 @@ public void startLoadJob(
startJobStream(job, stream, errorExtractor, client, Sleeper.DEFAULT,
createDefaultBackoff());
}
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Tries executing the RPC for at most {@code MAX_RPC_RETRIES} times
until it succeeds.
+ *
+ * @throws IOException if it exceeds {@code MAX_RPC_RETRIES} attempts.
+ */
+ @Override
+ public void startLoadJob(
Review Comment:
**🟢 Low**: To avoid duplicating the `Job` and `JobConfiguration`
construction across the four `startLoadJob` methods, consider having the
existing 2-arg and 3-arg `startLoadJob` overloads (lines 281 and 300) delegate
to these new overloads with `null` as the `reservation` argument.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
