[damccorm]

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() blocks 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

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR mitigates flakiness in StorageApiSinkRowUpdateIT by configuring DirectOptions.setTargetParallelism(1) in runPipelineAndWait, preventing concurrent DirectRunner bundles from sharing and racing on a closed StreamWriter while the destination table is lazily created on initial write failure.

🔍 General Feedback

• Moving p.run() into the try block activates the catch (Pipeline.PipelineExecutionException e) handler that swallows Exceptions.StreamWriterClosedException (previously unreachable because DirectOptions.isBlockOnRun() defaults to true). If a StreamWriterClosedException causes p.run() to abort after exhausting bundle retries, those rows were never written, so swallowing the exception masks the real failure and causes a confusing downstream assertion failure in assertRowsWritten.
• Beyond the test workaround, the underlying concurrency issue in StorageApiWriteUnshardedRecords (STORAGE_API_AT_LEAST_ONCE with CREATE_IF_NEEDED) occurs because FAILED_PRECONDITION from StreamWriterClosedException is treated as a fatal hasPersistentErrors condition before tryCreateTable.call() / RetryType.RETRY_ALL_OPERATIONS can be reached, and synchronous StreamWriterClosedException throws in runOperation bypass RetryManager.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

🟡 Medium: Moving p.run() inside the try block makes the catch (Pipeline.PipelineExecutionException e) block active (since DirectOptions.isBlockOnRun() is true by default, p.run() blocks and throws PipelineExecutionException directly).

However, StreamWriterClosedException is thrown when writeStream.appendRows(...) fails during DestinationState.flush(), and only surfaces as a PipelineExecutionException after the DirectRunner exhausts all 4 bundle retries and aborts the pipeline. If this exception is caught and ignored here, the failed bundle's rows were never written to BigQuery, and the test will immediately proceed to assertRowsWritten and fail with a misleading row-mismatch AssertionError (or 404 table not found) while discarding the actual pipeline failure stack trace.

With setTargetParallelism(1) preventing the concurrent bundle race, consider removing the try/catch block that swallows StreamWriterClosedException so that any unexpected failure surfaces with its full exception cause.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR resolves flakiness in StorageApiSinkRowUpdateIT by configuring DirectOptions.setTargetParallelism(1) in runPipelineAndWait to prevent concurrent DirectRunner bundles from racing on a shared StreamWriter during initial table creation. It also removes the obsolete StreamWriterClosedException catch block so unexpected pipeline errors surface directly.

🔍 General Feedback

• The changes are clean, well-scoped, and address the previous review feedback regarding exception handling.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR resolves flakiness in StorageApiSinkRowUpdateIT and BigQueryIOIcebergManagedTableIT by setting DirectOptions.setTargetParallelism(1) for pipelines writing with STORAGE_API_AT_LEAST_ONCE and CREATE_IF_NEEDED, preventing concurrent DirectRunner bundles from racing on a shared closed StreamWriter during initial table creation. It also cleans up the obsolete StreamWriterClosedException catch block in StorageApiSinkRowUpdateIT.

🔍 General Feedback

• The changes across both integration tests are clean, focused, and properly address the concurrent StreamWriter race condition in DirectRunner.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.