{"record":{"id":"2383d300bf8235f5","repo":"apache/beam","slug":"failed-to-flush-elements-on-window-expiration","errorCode":null,"errorMessage":"Failed to flush elements on window expiration!","messagePattern":"Failed to flush elements on window expiration!","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/SchemaUpdateHoldingFn.java","lineNumber":208,"sourceCode":"  public void onWindowExpiration(\n      @Key ShardedKey<DestinationT> key,\n      PipelineOptions pipelineOptions,\n      @StateId(\"bufferedElements\") BagState<TimestampedValue<ElementT>> bag,\n      @StateId(\"minBufferedTimestamp\") CombiningState<Long, long[], Long> minBufferedTimestamp,\n      MultiOutputReceiver o)\n      throws Exception {\n    // This can happen on test completion or drain. We can't set any more timers in window\n    // expiration, so we just have to loop until the schema is updated.\n    BackOff backoff =\n        new ExponentialBackOff.Builder()\n            .setMaxElapsedTimeMillis((int) TimeUnit.SECONDS.toMillis(10))\n            .build();\n    do {\n      if (tryFlushBuffer(key.getKey(), pipelineOptions, bag, minBufferedTimestamp, o)) {\n        return;\n      }\n    } while (BackOffUtils.next(com.google.api.client.util.Sleeper.DEFAULT, backoff));\n    throw new RuntimeException(\"Failed to flush elements on window expiration!\");\n  }\n\n  // Returns true if the buffer is completely flushed.\n  public boolean tryFlushBuffer(\n      DestinationT destination,\n      PipelineOptions pipelineOptions,\n      @StateId(\"bufferedElements\") BagState<TimestampedValue<ElementT>> bag,\n      @StateId(\"minBufferedTimestamp\") CombiningState<Long, long[], Long> minBufferedTimestamp,\n      MultiOutputReceiver o)\n      throws Exception {\n    // Force an update of the MessageConverter schema.\n    StorageApiDynamicDestinations.MessageConverter<ElementT> messageConverter =\n        convertMessagesDoFn\n            .getMessageConverters()\n            .get(\n                destination,\n                convertMessagesDoFn.getDynamicDestinations(),\n                pipelineOptions,","sourceCodeStart":190,"sourceCodeEnd":226,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/SchemaUpdateHoldingFn.java#L190-L226","documentation":"RuntimeException thrown by SchemaUpdateHoldingFn.onWindowExpiration when the function tries to flush all buffered records to the Storage Write API before the state window expires and every retry attempt fails. Elements would otherwise be lost on expiration, so the code deliberately fails the work item instead.","triggerScenarios":"tryFlushBuffer repeatedly fails (AppendRows errors such as NOT_FOUND, PERMISSION_DENIED, schema mismatches, or gRPC outages) across all BackOff attempts while the window is expiring.","commonSituations":"Destination table deleted or never created mid-stream; service account missing bigquery.tables.updateData; schema changed remotely so buffered records no longer validate; prolonged BigQuery Storage Write API outage exceeding the retry window.","solutions":["Check the preceding logs for the underlying AppendRows error (status code and stream name)","Confirm the destination table exists and the service account has bigquery.tables.updateData permission","Refresh/align the writer schema with the current table schema and rerun the failed work item","If the outage was transient, restart the pipeline from the last successful checkpoint"],"exampleFix":"null","handlingStrategy":"try-catch","validationCode":"// Before running, verify destination tables are writable\ncom.google.cloud.bigquery.Table t = bigquery.getTable(tableRef.getDataset(), tableRef.getTable());\nif (t == null) throw new IllegalStateException(\"Destination table missing: \" + tableRef);\ntestIamPermissions(tableRef, \"bigquery.tables.updateData\");","typeGuard":null,"tryCatchPattern":"try {\n  runPipeline();\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"Failed to flush elements on window expiration\")) {\n    // inspect earlier AppendRows logs for status code, fix table/permissions, resubmit\n  } else { throw e; }\n}","preventionTips":["Keep destination tables alive for the whole pipeline run","Grant bigquery.tables.updateData to the pipeline SA","Keep writer schemas in sync with table schemas (or enable auto schema updates)","Re-run failed work items after transient Storage Write API outages"],"tags":["bigquery","storage-write","windowing","flush","gcp"],"backgroundTag":"api-error-response","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}