gemini-code-assist[bot] commented on code in PR #38470:
URL: https://github.com/apache/beam/pull/38470#discussion_r3227443342
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java:
##########
@@ -1496,6 +1496,7 @@ public static class WriteStreamServiceImpl implements
WriteStreamService {
private final BigQueryWriteClient newWriteClient;
private final long storageWriteMaxInflightRequests;
private final long storageWriteMaxInflightBytes;
+ private final @Nullable Integer
storageWriteApiMaxRequestCallbackWaitTimeSec;
Review Comment:
If `BigQueryOptions` is updated to use `Long`, this field should also be
updated to `@Nullable Long` for consistency.
```suggestion
private final @Nullable Long
storageWriteApiMaxRequestCallbackWaitTimeSec;
```
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java:
##########
@@ -1578,6 +1583,11 @@ public StreamAppendClient getStreamAppendClient(
options.as(BigQueryOptions.class).getMaxConnectionPoolConnections())
.build());
+ if (storageWriteApiMaxRequestCallbackWaitTimeSec != null) {
+ StreamWriter.setMaxRequestCallbackWaitTime(
+
java.time.Duration.ofSeconds(storageWriteApiMaxRequestCallbackWaitTimeSec));
+ }
Review Comment:
The method `StreamWriter.setMaxRequestCallbackWaitTime` is a static method
that modifies a global setting in the `StreamWriter` class. This has
significant implications in a Beam environment:
1. **Global Side Effects**: Calling this affects all `StreamWriter`
instances in the same JVM, including those created by other pipelines or tasks
running on the same worker.
2. **Persistence**: If one pipeline sets this value, it will persist for all
subsequent pipelines running in the same worker process, even if they don't
specify this option (as the `null` check at line 1586 would skip resetting it).
3. **Race Conditions**: Multiple threads calling `getStreamAppendClient`
concurrently will race to set this global variable.
Given that the underlying library uses a static setter, this behavior is
difficult to avoid, but it should be documented in `BigQueryOptions`.
Additionally, consider if there's a way to scope this or if it's acceptable to
have such global side effects.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryOptions.java:
##########
@@ -153,6 +153,13 @@ public interface BigQueryOptions
void setStorageWriteMaxInflightBytes(Long value);
+ @Description(
+ "Maximum time in seconds a Storage Write API append request is allowed
to wait in the "
+ + "request callback queue before timing out. Overrides Storage Write
API default (5 min)")
+ Integer getStorageWriteApiMaxRequestCallbackWaitTimeSec();
+
+ void setStorageWriteApiMaxRequestCallbackWaitTimeSec(Integer value);
Review Comment:
For consistency with other duration-based options in `BigQueryOptions`
(e.g., `getStorageWriteApiTriggeringFrequencySec`), consider using `Long`
instead of `Integer` for this property.
```suggestion
Long getStorageWriteApiMaxRequestCallbackWaitTimeSec();
void setStorageWriteApiMaxRequestCallbackWaitTimeSec(Long value);
```
--
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]
