What causes "Failed to write the mutation group: {}" in apache/beam?

Inspect the returned failures PCollection (failedTag) and replay the MutationGroups after fixing the root cause.

Failed to write the mutation group

Error message

Failed to write the mutation group: {}

What it means

In the fault-tolerant write path of SpannerIO's grouping transform, a MutationGroup failed to be written to Cloud Spanner with a SpannerException after the configured retries. The failure is counted, logged, and the MutationGroup is emitted to the failure tag (withWriteFailureFn / failedTag output) instead of failing the pipeline. This is a warning-level signal that some writes were dropped from the main output and routed to the failures PCollection.

Solutions

Inspect the returned failures PCollection (failedTag) and replay the MutationGroups after fixing the root cause.
Check the attached SpannerException for the specific cause (permissions, schema mismatch, timeouts) and fix accordingly.
Reduce mutation group size or batch count to avoid timeouts, or increase the retry deadline configuration.
Verify table schemas and that all mutations reference existing tables/columns with valid keys.

Example fix

// before: failures silently dropped by ignoring the failed tag
spannerWrites.apply("Write", SpannerIO.write().to(config));

// after: handle the failures output
SpannerWriteResult result =
    rows.apply(SpannerIO.write()
        .withInstanceId(instanceId)
        .withDatabaseId(databaseId)
        .withFailureTag());
result.getFailedMutations().apply("LogFailures", ParDo.of(new LogMutationGroupDoFn()));

Defensive patterns

Strategy: try-catch

Try / catch

// Consume the failure-tag output and persist/replay failed MutationGroups:
SpannerWriteResult result = rows.apply(SpannerIO.write()
    .withInstanceId(instanceId).withDatabaseId(dbId).withFailureTag());
result.getFailedMutations()
    .apply("PersistFailures", Write.to(failureSink));

Prevention

Always configure and monitor the failure-tag PCollection when using SpannerIO.write().
Keep mutation groups within Spanner size/timeout limits; split large batches.
Validate schemas and permissions before large write jobs.
Set up alerting on mutationGroupsWriteFail counters.

When it happens

Trigger: Applying SpannerIO.write().to(...) in a pipeline configured with a failure output (failureTag), and writeMutations(mg) throws SpannerException — e.g. DeadlineExceeded, Aborted beyond retries, permission errors, or schema violations for a mutation group.

Common situations: Large mutation groups exceeding size/timeout limits; transient Spanner outages; writes violating the database schema (missing table/column); interleaved-key ordering violations.

Related errors

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

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java:2879

          // fall through and retry individual mutationGroups.
        } else if (failureMode == FailureMode.FAIL_FAST) {
          mutationGroupsWriteFail.inc(mutations.size());
          LOG.error("Failed to write a batch of mutation groups", e);
          throw e;
        } else {
          throw new IllegalArgumentException("Unknown failure mode " + failureMode);
        }
      }

      // If we are here, writing a batch has failed, retry individual mutations.
      for (MutationGroup mg : mutations) {
        try {
          spannerWriteRetries.inc();
          writeMutations(mg);
          mutationGroupsWriteSuccess.inc();
        } catch (SpannerException e) {
          mutationGroupsWriteFail.inc();
          LOG.warn("Failed to write the mutation group: {}", mg, e);
          c.output(failedTag, mg);
        }
      }
    }

    /*
     Spanner aborts all inflight transactions during a schema change. Client is expected
     to retry silently. These must not be counted against retry backoff.
    */
    private void spannerWriteWithRetryIfSchemaChange(List<Mutation> batch) throws SpannerException {
      Set<String> tableNames = batch.stream().map(Mutation::getTable).collect(Collectors.toSet());
      for (int retry = 1; ; retry++) {
        try {
          spannerAccessor
              .getDatabaseClient()
              .writeAtLeastOnceWithOptions(batch, getTransactionOptions());
          // Get names of all tables in batch of mutations.
          reportServiceCallMetricsForBatch(tableNames, "ok");

View on GitHub (pinned to 12126d8942)