damccorm opened a new pull request, #41142: URL: https://github.com/apache/beam/pull/41142
### Summary Fixes flaky `StorageApiSinkSchemaUpdateWithInputSchemaIT` failures in `PostCommit Java BigQueryEarlyRollout` ([run 42467812022](https://github.com/apache/beam/actions/runs/42467812022)). ### Root Cause In `StorageApiSinkSchemaUpdateITBase.runStreamingPipeline`, `PeriodicImpulse` followed by parallel `MapElements` (`GenerateRowFunc`) in `DirectRunner` can deliver bundles out of `id` order to `UpdateSchemaDoFn`. Previously, `UpdateSchemaDoFn` triggered the BigQuery table schema update based on arrival count (`current == SCHEMA_UPDATE_TRIGGER`), spaced the recognition window based on `id > SCHEMA_UPDATE_TRIGGER && id < ORIGINAL_N`, and immediately emitted any `id >= ORIGINAL_N` (`extraField`) row that arrived early before all `ORIGINAL_N` original-schema rows and the 60s `SCHEMA_RECOGNITION_WINDOW_MS` had completed. When `id >= ORIGINAL_N` rows were emitted before the Storage Write API stream recognized the updated table schema, `extraField` was dropped from those early rows. ### Fix - Buffer any out-of-order `id >= ORIGINAL_N` rows in `@StateId("deferredRows") BagState<TableRow>` while `current < ORIGINAL_N`. - Space the recognition window using `current > SCHEMA_UPDATE_TRIGGER && current < ORIGINAL_N` rather than out-of-order `id`, and flush deferred `id >= ORIGINAL_N` rows once `current == ORIGINAL_N` completes. ### Verification - Triggered `beam_PostCommit_Java_BigQueryEarlyRollout.yml` workflow run on branch: [`42733480274`](https://github.com/apache/beam/actions/runs/42733480274) -- 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]
