[Java] Propagate downstream errors from BigQuery success output

[bvolpato]

With withPropagateSuccessfulStorageApiWrites(true), the unsharded BigQuery Storage Write API writer catches exceptions from fused downstream transforms as if they were row-parsing failures. The SDK can then report bundle success despite missing downstream output.

For example, write rows A and B using STORAGE_API_AT_LEAST_ONCE, then attach a fused ParDo to getSuccessfulStorageApiInserts() that throws on A and emits B. Both rows reach the fake BigQuery service, but the original SDK harness logs A's exception, processes B, and completes the bundle successfully. A has disappeared from downstream processing without a bundle failure to trigger recovery.

Move outputWithTimestamp outside the row-parsing exception handler. A downstream exception now fails the bundle. Parsing failures retain their existing handling. This changes failure propagation; it does not roll back rows already accepted by BigQuery or change at-least-once write semantics.

The regression builds a public BigQueryIO pipeline, verifies that the writer and downstream consumer can fuse, and executes them through the real Java SDK ProcessBundleHandler. It covers both the record-threshold flush during ProcessElement and the flush during FinishBundle. Assertions check the downstream exception, instruction poisoning, and the exact rows stored before failure. A classic DirectRunner pipeline alone does not exercise this synchronous fused-output call.

Validation:

• Both regression cases fail against unchanged upstream and pass with the fix.
• A native Gradle run combining the new regression with BigQueryIOWriteTest.testStorageApiErrorsWriteTableRows* passes 5 cases, with 2 inapplicable parameter combinations skipped. This used JDK 21 with the repository's default Java compilation target.
• Spotless, validateChanges, and git diff --check pass. CHANGES.md includes the fix.
• The reproduction uses an in-process fake BigQuery service, not a live BigQuery incident.

./gradlew :sdks:java:io:google-cloud-platform:test \
  --tests org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecordsTest

The production change adds no dependencies and does not alter the append or retry protocol.

[bvolpato]

The GCP retry passed on the same commit: 2,229 tests passed, 237 were skipped, and none failed. Both new fused-harness regression cases passed, as did the streaming-insert KMS test that previously received a BigQuery HTTP 500.

All current-head checks are now complete: 18 successful and one neutral.