{"record":{"id":"1ae191e70385c2f2","repo":"apache/beam","slug":"error-when-writing-batch","errorCode":null,"errorMessage":"Error when writing batch.","messagePattern":"Error when writing batch\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/common/AsyncBatchWriteHandler.java","lineNumber":367,"sourceCode":"                run();\n              } else {\n                throwable = new IOException(summarizeErrors(\"Exceeded retries\", results));\n              }\n            } catch (Throwable e) {\n              throwable = new IOException(summarizeErrors(\"Aborted retries\", results), e);\n            }\n          }\n        }\n      } catch (Throwable e) {\n        throwable = e;\n      }\n      if (throwable != null) {\n        setAsyncFailure(throwable);\n      }\n    }\n\n    private void setAsyncFailure(Throwable throwable) {\n      LOG.warn(\"Error when writing batch.\", throwable);\n      hasErrored.set(true);\n      asyncFailure.updateAndGet(\n          ex -> {\n            if (ex != null) {\n              throwable.addSuppressed(ex);\n            }\n            return throwable;\n          });\n      requestPermits.release(concurrentRequests); // unblock everything to fail fast\n    }\n\n    private String summarizeErrors(String prefix, List<ResT> results) {\n      Map<String, Long> countsPerError =\n          results.stream()\n              .map(errorCodeFn)\n              .filter(notNull())\n              .collect(groupingBy(identity(), counting()));\n      return countsPerError.entrySet().stream()","sourceCodeStart":349,"sourceCodeEnd":385,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/common/AsyncBatchWriteHandler.java#L349-L385","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before: default client, no retry tuning\nDynamoDbAsyncClient client = DynamoDbAsyncClient.builder().build();\n// after: retries with backoff to survive throttling\nDynamoDbAsyncClient client = DynamoDbAsyncClient.builder()\n    .overrideConfiguration(o -> o.setRetryPolicy(RetryPolicy.builder()\n        .numRetries(10)\n        .backoffStrategy(BackoffStrategy.exponentialDelay())\n        .build()))\n    .build();","handlingStrategy":"retry","validationCode":"// validate write access and capacity before batch writes\niamSimulateWrite(tableArn);\nif (items.stream().anyMatch(i -> i.sizeInBytes() > MAX_ITEM_BYTES)) throw new IllegalArgumentException(\"item exceeds size limit\");","typeGuard":null,"tryCatchPattern":"try {\n  batchWriter.write(items);\n} catch (Exception e) {\n  // asyncFailure chains suppressed causes; inspect all\n  Throwable[] suppressed = e.getSuppressed();\n  if (e.getCause() instanceof ProvisionedThroughputExceededException) { /* back off / reduce parallelism */ }\n}","preventionTips":["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"],"tags":["java","apache-beam","aws","async-write","batch"],"backgroundTag":"api-error-response","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"}