{"record":{"id":"ffd32cf238cd5a1e","repo":"apache/beam","slug":"flush-of-stream-to-offset-failed","errorCode":null,"errorMessage":"Flush of stream {} to offset {} failed","messagePattern":"Flush of stream (.+?) to offset (.+?) failed","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java","lineNumber":215,"sourceCode":"              Duration.standardMinutes(1),\n              3,\n              BigQuerySinkMetrics.throttledTimeCounter(BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS));\n      retryManager.addOperation(\n          // runOperation\n          c -> {\n            try {\n              flushOperationsSent.inc();\n              return datasetService.flush(streamId, offset);\n            } catch (Exception e) {\n              throw new RuntimeException(e);\n            }\n          },\n          // onError\n          contexts -> {\n            Context<FlushRowsResponse> failedContext =\n                Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));\n            Throwable error = failedContext.getError();\n            LOG.warn(\"Flush of stream {} to offset {} failed\", streamId, offset, error);\n            flushOperationsFailed.inc();\n            BigQuerySinkMetrics.reportFailedRPCMetrics(\n                failedContext, BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS);\n\n            if (error instanceof ApiException) {\n              Code statusCode = ((ApiException) error).getStatusCode().getCode();\n              if (statusCode.equals(Code.ALREADY_EXISTS)) {\n                flushOperationsAlreadyExists.inc();\n                // Implies that we have already flushed up to this point, so don't retry.\n                return RetryType.DONT_RETRY;\n              }\n              if (statusCode.equals(Code.INVALID_ARGUMENT)) {\n                flushOperationsInvalidArgument.inc();\n                // Implies that the stream has already been finalized.\n                // TODO: Storage API should provide a more-specific way of identifying this failure.\n                return RetryType.DONT_RETRY;\n              }\n              if (statusCode.equals(Code.NOT_FOUND)) {","sourceCodeStart":197,"sourceCodeEnd":233,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java#L197-L233","documentation":"StorageApiFlushAndFinalizeDoFn's async onError callback fires when the FlushRows RPC to BigQuery Storage Write fails. The library logs the stream id, target offset, and error, increments flushOperationsFailed, reports failed RPC metrics, and then inspects the status code to decide retryability. This indicates rows buffered in a stream could not be committed to the given offset.","triggerScenarios":"FlushRows RPC returns an error: UNAVAILABLE (transient service issue), stream already closed/finalized, permission errors on the write stream, deadline exceeded, or gRPC channel failure.","commonSituations":"BigQuery Storage Write API transient unavailability; stream finalized while flush in flight; quota/throughput limits exceeded; IAM lacking BigQuery dataEditor on the table.","solutions":["Check the attached Throwable's status code: UNAVAILABLE/DEADLINE_EXCEEDED are typically retried automatically — verify retry policy.","Ensure the service account has BigQuery Data Editor and Storage Write API access.","Watch for streams being finalized prematurely; don't share streams across concurrent flushes.","Check BigQuery quota (throughput per stream/table) and BigQuerySinkMetrics for failure patterns."],"exampleFix":"// before: frequent UNAVAILABLE flushes with default channel\nBigQueryStorageApiSink options... \n// after: enable retries/trace and verify IAM + quota\nWriteResult result = rows.apply(\"StorageApiWrite\", StorageApiWritePayloadDestination...).withAutoSchemaUpdate(true);","handlingStrategy":"retry","validationCode":"// pre-flight: verify write access and API enablement\nboolean ok = bigquery.getTable(tableId) != null\n    && bigquery.getIamPolicy(tableId).contains(\"bigquery.tables.updateData\");","typeGuard":null,"tryCatchPattern":"// classify by status code, as the DoFn does\nif (error instanceof ApiException) {\n  Code c = ((ApiException) error).getStatusCode().getCode();\n  boolean retryable = c == Code.UNAVAILABLE || c == Code.DEADLINE_EXCEEDED || c == Code.INTERNAL;\n}","preventionTips":["Grant the service account BigQuery Data Editor on destination tables.","Enable the BigQuery Storage Write API and verify quotas.","Monitor BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS failure rates.","Avoid finalizing streams while flushes are in flight."],"tags":["bigquery","grpc","flush","retry"],"backgroundTag":"api-request-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}