damccorm opened a new pull request, #41066: URL: https://github.com/apache/beam/pull/41066
## Summary Fixes a flake in `StorageApiSinkRowUpdateIT` (`PreCommit Java GCP IO Direct` run https://github.com/apache/beam/actions/runs/42324970282/job/128258023987): ``` org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.RuntimeException: Append to stream projects/apache-beam-testing/datasets/storage_api_sink_rows_update.../tables/table.../streams/_default failed with Status Code FAILED_PRECONDITION. The stream may not exist. Caused by: com.google.cloud.bigquery.storage.v1.Exceptions$StreamWriterClosedException: FAILED_PRECONDITION: Connection is closed due to com.google.api.gax.rpc.PermissionDeniedException: io.grpc.StatusRuntimeException: PERMISSION_DENIED: Permission 'TABLES_UPDATE_DATA' denied on resource 'projects/apache-beam-testing/datasets/storage_api_sink_rows_update.../tables/table...' (or it may not exist). ``` ### Root Cause In `StorageApiSinkRowUpdateIT`, `Create.of(items)` writes to a non-existent destination table with `CreateDisposition.CREATE_IF_NEEDED` and `STORAGE_API_AT_LEAST_ONCE` (`_default` stream). When `DirectRunner` executes multiple bundles concurrently on separate threads, they share the same cached `StreamWriter` in `StorageApiWriteUnshardedRecords.APPEND_CLIENTS`. When the first thread's initial `AppendRows` RPC fails with `PERMISSION_DENIED ... (or it may not exist)` prior to table creation (or before table creation propagates to the Storage Write API), `ConnectionWorker` marks the shared `StreamWriter` closed. A concurrent bundle calling `append` on that shared `StreamWriter` before cache invalidation receives `StreamWriterClosedException` (which carries status code `FAILED_PRECONDITION` and is treated as a persistent error by `StorageApiWriteUnshardedRecords`). Additionally, `p.run()` in `runPipelineAndWait` was previously outside the `try` block even though `DirectRunner.run()` b locks and throws inside `p.run()`. ### Changes - Set `p.getOptions().as(DirectOptions.class).setTargetParallelism(1)` in `StorageApiSinkRowUpdateIT.runPipelineAndWait` so bundles execute sequentially on a single thread without racing on a closed `StreamWriter` during initial table creation. - Move `PipelineResult result = p.run()` inside the `try` block in `runPipelineAndWait`. ## Verification - Triggered `PreCommit Java GCP IO Direct` workflow run on this branch: https://github.com/apache/beam/actions/runs/42332692751 -- 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]
