damccorm commented on code in PR #41066:
URL: https://github.com/apache/beam/pull/41066#discussion_r4512695623
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiSinkRowUpdateIT.java:
##########
@@ -202,21 +202,10 @@ private void assertRowsWritten(String tableSpec,
Iterable<TableRow> expected)
}
private void runPipelineAndWait(Pipeline p) {
+ // Run on a single DirectRunner thread so concurrent bundles do not share
a closed
+ // StreamWriter while the destination table is being created on the first
write retry.
+ p.getOptions().as(DirectOptions.class).setTargetParallelism(1);
Review Comment:
Yes — this is working around a concurrency bug in
`StorageApiWriteUnshardedRecords` when `useDefaultStream = true` (used by
`STORAGE_API_AT_LEAST_ONCE` / row-mutation updates):
1. Multiple `DirectRunner` threads share the same cached `AppendClientInfo`
/ `StreamWriter` instance from `APPEND_CLIENTS` for the `_default` stream
(since `numStorageWriteApiStreamAppendClients` defaults to `1`, so
`clientNumber = 0` for all threads).
2. On the first write to a newly created table (`CREATE_IF_NEEDED`), or when
BigQuery closes the gRPC stream on an error before `tryCreateTable.call()` runs
in `RetryManager`'s `onError` callback
(`StorageApiWriteUnshardedRecords.java:983,1032`),
`invalidateAppendClient(true)` (or `ConnectionWorker` inside `StreamWriter`)
closes that shared `StreamWriter`.
3. When a concurrent bundle on another thread calls
`writeStream.appendRows(c.offset, c.protoRows)`
(`StorageApiWriteUnshardedRecords.java:821`) on the still-referenced
`StreamWriter`, `StreamWriter.append()` throws
`Exceptions.StreamWriterClosedException` (`FAILED_PRECONDITION: Connection is
already closed`) **synchronously**.
4. In `StorageApiWriteUnshardedRecords.java:831-833`, the synchronous `catch
(Exception e) { throw new RuntimeException(e); }` inside `RetryManager`'s
`runOperation` lambda escapes `RetryManager.Operation.run()` directly instead
of returning `ApiFutures.immediateFailedFuture(e)`, and
`StorageApiWriteUnshardedRecords.java:1020` also treats
`Status.Code.FAILED_PRECONDITION` as a non-retryable persistent error even on
the `_default` stream.
5. Previously, `StorageApiSinkRowUpdateIT.runPipelineAndWait` tried to mask
this by catching and ignoring `StreamWriterClosedException` at pipeline
completion, which caused silent data loss (missing rows in `assertRowsWritten`)
whenever the exception aborted a bundle whose rows hadn't been flushed yet.
I initially restricted this PR to test-only changes (`targetParallelism(1)`)
to avoid modifying core sink behavior, but we can fix
`StorageApiWriteUnshardedRecords.java` directly (converting synchronous
`StreamWriterClosedException` into a failed future and allowing `RetryManager`
to invalidate/recreate the client and retry on `StreamWriterClosedException`)
if you prefer!
--
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]
