What causes "Failed to close batch" in apache/beam?

Inspect e.getCause() on the resulting IOException to find the underlying gRPC status

apache/beam · error · IOException

Failed to close batch

Error message

Failed to close batch

What it means

BigtableServiceImpl.close waits for the BulkMutation batcher's completion future. If that future completes exceptionally, the ExecutionException's cause is wrapped into an IOException with this message, meaning the final flush of pending mutations failed.

Source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:569

          if (Duration.ZERO.isShorterThan(closeWaitTimeout)) {
            future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);
          } else {
            future.get();
          }
          bulkSize.update(outstandingMutations);
          outstandingMutations = 0;
          stopwatch.stop();
          latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
        } catch (BatchingException e) {
          // Ignore batching failures because element failures are tracked as is in
          // BigtableIOWriteFn.
          // TODO: Bigtable client already tracks BatchingExceptions, use BatchingExceptions
          // instead of tracking them separately in BigtableIOWriteFn.
        } catch (TimeoutException e) {
          // We fail because future.get() timed out
          throw new IOException("BulkMutation took too long to close", e);
        } catch (ExecutionException e) {
          throw new IOException("Failed to close batch", e.getCause());
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          // We fail since close() operation was interrupted.
          throw new IOException(e);
        }
        bulkMutation = null;
      }
    }

    @Override
    public CompletableFuture<MutateRowResponse> writeRecord(
        KV<ByteString, Iterable<Mutation>> record) throws IOException {

      com.google.cloud.bigtable.data.v2.models.Mutation mutation =
          com.google.cloud.bigtable.data.v2.models.Mutation.fromProtoUnsafe(record.getValue());

      RowMutationEntry entry = RowMutationEntry.createFromMutationUnsafe(record.getKey(), mutation);

View on GitHub (pinned to 12126d8942)

Solutions

Inspect e.getCause() on the resulting IOException to find the underlying gRPC status
Re-run the pipeline — element failures are tracked in BigtableIOWriteFn and written to the failure output
Validate tableId/instanceId/projectId are consistent across the pipeline
Ensure the table and column families exist before writing

Example fix

// before
table.exists(); // assume table exists, let flush fail
// after
if (!service.checkTableExists(tableId)) {
  createTableWithFamilies(projectId, instanceId, tableId, families);
}

Defensive patterns

Strategy: retry

Validate before calling

// pre-flight: confirm table exists before writing
if (!factory.tableExists(tableId)) { createTable(projectId, instanceId, tableId, families); }

Try / catch

try {
  writer.close();
} catch (IOException e) {
  Throwable cause = e.getCause(); // ExecutionException cause from flush future
  if (cause instanceof ApiException) { /* inspect gRPC status, retry */ }
}

Prevention

Ensure table and column families exist before writing
Rely on BigtableIOWriteFn per-element failure tracking and retry the failing bundles
Watch network stability between the runner and Cloud Bigtable

When it happens

Trigger: Calling close() on the Bigtable writer when the batcher's flush future fails — typically due to gRPC errors from Cloud Bigtable during the final flush (e.g. server errors, dead peer, quota limits).

Common situations: Network drop or Bigtable outage at pipeline shutdown; invalid table name not caught earlier; mutation payload rejected at flush time.

Related errors

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