apache/beam · error

Flush of stream to offset failed

Error message

Flush of stream {} to offset {} failed

What it means

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.

Solutions

1. Check the attached Throwable's status code: UNAVAILABLE/DEADLINE_EXCEEDED are typically retried automatically — verify retry policy.
2. Ensure the service account has BigQuery Data Editor and Storage Write API access.
3. Watch for streams being finalized prematurely; don't share streams across concurrent flushes.
4. Check BigQuery quota (throughput per stream/table) and BigQuerySinkMetrics for failure patterns.

Example fix

// before: frequent UNAVAILABLE flushes with default channel
BigQueryStorageApiSink options... 
// after: enable retries/trace and verify IAM + quota
WriteResult result = rows.apply("StorageApiWrite", StorageApiWritePayloadDestination...).withAutoSchemaUpdate(true);

// pre-flight: verify write access and API enablement
boolean ok = bigquery.getTable(tableId) != null
    && bigquery.getIamPolicy(tableId).contains("bigquery.tables.updateData");

// classify by status code, as the DoFn does
if (error instanceof ApiException) {
  Code c = ((ApiException) error).getStatusCode().getCode();
  boolean retryable = c == Code.UNAVAILABLE || c == Code.DEADLINE_EXCEEDED || c == Code.INTERNAL;
}

When it happens

Trigger: 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.

Common situations: BigQuery Storage Write API transient unavailability; stream finalized while flush in flight; quota/throughput limits exceeded; IAM lacking BigQuery dataEditor on the table.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

Related errors

• Append to stream by client # failed with error, operations…
• Append to stream failed with invalid offset of
• Append to stream failed with Status Code . The stream may…
• Caught exception whilw trying to close append client…
• expansion failed

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/ffd32cf238cd5a1e. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java:215

              Duration.standardMinutes(1),
              3,
              BigQuerySinkMetrics.throttledTimeCounter(BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS));
      retryManager.addOperation(
          // runOperation
          c -> {
            try {
              flushOperationsSent.inc();
              return datasetService.flush(streamId, offset);
            } catch (Exception e) {
              throw new RuntimeException(e);
            }
          },
          // onError
          contexts -> {
            Context<FlushRowsResponse> failedContext =
                Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));
            Throwable error = failedContext.getError();
            LOG.warn("Flush of stream {} to offset {} failed", streamId, offset, error);
            flushOperationsFailed.inc();
            BigQuerySinkMetrics.reportFailedRPCMetrics(
                failedContext, BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS);

            if (error instanceof ApiException) {
              Code statusCode = ((ApiException) error).getStatusCode().getCode();
              if (statusCode.equals(Code.ALREADY_EXISTS)) {
                flushOperationsAlreadyExists.inc();
                // Implies that we have already flushed up to this point, so don't retry.
                return RetryType.DONT_RETRY;
              }
              if (statusCode.equals(Code.INVALID_ARGUMENT)) {
                flushOperationsInvalidArgument.inc();
                // Implies that the stream has already been finalized.
                // TODO: Storage API should provide a more-specific way of identifying this failure.
                return RetryType.DONT_RETRY;
              }
              if (statusCode.equals(Code.NOT_FOUND)) {

View on GitHub (pinned to 12126d8942)