What causes "Error when writing batch." in apache/beam?

Inspect the logged throwable cause — address throttling by reducing write parallelism/batch size or increasing table capacity.

apache/beam · error

Error when writing batch.

Error message

Error when writing batch.

What it means

AsyncBatchWriteHandler (AWS SDK v2 S3/DynamoDB async batching in Beam's aws2 IO) records an asynchronous write failure via setAsyncFailure: the CompletableFuture callback received a throwable, logs it, sets hasErrored, and chains it into asyncFailure (preserving prior failures as suppressed exceptions). The failure typically surfaces later when the pipeline checks the write result, aborting the write.

Solutions

Inspect the logged throwable cause — address throttling by reducing write parallelism/batch size or increasing table capacity.
Verify IAM credentials/roles include write permissions for the target table/bucket.
Enable retry configuration on the AWS client builder (retryPolicy, backoff) to absorb transient throttling.
Fix data-shape issues (oversized items, invalid keys) indicated by the wrapped SDK exception.

Example fix

// before: default client, no retry tuning
DynamoDbAsyncClient client = DynamoDbAsyncClient.builder().build();
// after: retries with backoff to survive throttling
DynamoDbAsyncClient client = DynamoDbAsyncClient.builder()
    .overrideConfiguration(o -> o.setRetryPolicy(RetryPolicy.builder()
        .numRetries(10)
        .backoffStrategy(BackoffStrategy.exponentialDelay())
        .build()))
    .build();

Defensive patterns

Strategy: retry

Validate before calling

// validate write access and capacity before batch writes
iamSimulateWrite(tableArn);
if (items.stream().anyMatch(i -> i.sizeInBytes() > MAX_ITEM_BYTES)) throw new IllegalArgumentException("item exceeds size limit");

Try / catch

try {
  batchWriter.write(items);
} catch (Exception e) {
  // asyncFailure chains suppressed causes; inspect all
  Throwable[] suppressed = e.getSuppressed();
  if (e.getCause() instanceof ProvisionedThroughputExceededException) { /* back off / reduce parallelism */ }
}

Prevention

Configure SDK retry policy with exponential backoff for throttling
Right-size table capacity or use on-demand mode for spiky bulk loads
Validate IAM write permissions before launching write jobs
Bound batch sizes and in-flight request counts

When it happens

Trigger: An async AWS batch request (e.g. DynamoDB batchWriteItem / S3 batched writes) completes exceptionally: throttling (ProvisionedThroughputExceededException), item validation errors, access denied, or SDK client shutdown mid-flight; the failure callback invokes setAsyncFailure.

Common situations: Write throughput exceeding table/capacity limits, IAM policies missing the required write permissions, oversized items, or network issues during bulk loads with the AWS2 IO connectors.

Related errors

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

Appendix: source

Thrown at sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/common/AsyncBatchWriteHandler.java:367

                run();
              } else {
                throwable = new IOException(summarizeErrors("Exceeded retries", results));
              }
            } catch (Throwable e) {
              throwable = new IOException(summarizeErrors("Aborted retries", results), e);
            }
          }
        }
      } catch (Throwable e) {
        throwable = e;
      }
      if (throwable != null) {
        setAsyncFailure(throwable);
      }
    }

    private void setAsyncFailure(Throwable throwable) {
      LOG.warn("Error when writing batch.", throwable);
      hasErrored.set(true);
      asyncFailure.updateAndGet(
          ex -> {
            if (ex != null) {
              throwable.addSuppressed(ex);
            }
            return throwable;
          });
      requestPermits.release(concurrentRequests); // unblock everything to fail fast
    }

    private String summarizeErrors(String prefix, List<ResT> results) {
      Map<String, Long> countsPerError =
          results.stream()
              .map(errorCodeFn)
              .filter(notNull())
              .collect(groupingBy(identity(), counting()));
      return countsPerError.entrySet().stream()

View on GitHub (pinned to 12126d8942)