{"record":{"id":"6322b3ded6c0f909","repo":"apache/beam","slug":"error-writing-to-dynamodb-unprocessed-items-remaining","errorCode":null,"errorMessage":"Error writing to DynamoDB. Unprocessed items remaining","messagePattern":"Error writing to DynamoDB\\. Unprocessed items remaining","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"critical","filePath":"sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/dynamodb/DynamoDBIO.java","lineNumber":463,"sourceCode":"          // Group values KV<tableName, writeRequest> by tableName\n          // Note: The original order of arrival is lost reading the map entries.\n          Map<String, List<WriteRequest>> writesPerTable =\n              batch.values().stream()\n                  .collect(groupingBy(KV::getKey, mapping(KV::getValue, toList())));\n\n          // Backoff used to resume from partial failures\n          BackOff resume = resumeBackoff.backoff();\n          do {\n            BatchWriteItemRequest batchRequest =\n                BatchWriteItemRequest.builder().requestItems(writesPerTable).build();\n            // If unprocessed items remain, we have to resume the operation (with backoff)\n            writesPerTable = client.batchWriteItem(batchRequest).unprocessedItems();\n          } while (!writesPerTable.isEmpty() && BackOffUtils.next(Sleeper.DEFAULT, resume));\n\n          if (!writesPerTable.isEmpty()) {\n            DYNAMO_DB_WRITE_FAILURES.inc();\n            LOG.error(RESUME_ERROR_LOG, writesPerTable);\n            throw new IOException(ERROR_UNPROCESSED_ITEMS);\n          }\n        } finally {\n          batch.clear();\n        }\n      }\n\n      @Teardown\n      public void tearDown() {\n        if (client != null) {\n          client.close();\n        }\n      }\n    }\n  }\n}\n","sourceCodeStart":445,"sourceCodeEnd":479,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/dynamodb/DynamoDBIO.java#L445-L479","documentation":"DynamoDBIO's batch write retries unprocessed items using the configured backoff policy; after exhausting retries, if some items are still unprocessed, it increments the failure counter, logs them, and throws an IOException. This means DynamoDB persistently rejected part of the batch (throughput limits, item size, throttling).","triggerScenarios":"Calling flushBatch (from processElement or finishBundle) when batchWriteItem keeps returning non-empty unprocessedItems after all BackOff retries are exhausted.","commonSituations":"DynamoDB table provisioned throughput exceeded; hot partitions; very large batches near 25-item/16MB limits; sustained write rates above capacity; AWS throttling during traffic spikes.","solutions":["Enable DynamoDB auto-scaling or increase write capacity on target tables","Reduce batch size / rate limit the write via a rate limit policy in the pipeline","Check the RESUME_ERROR_LOG output to identify which items failed and why","Retry the failed pipeline range / use Beam retry plus a dead-letter queue for persistently failing items"],"exampleFix":"// before\nDynamoDBIO.<T>write().withWriteResultMapper(...) // default backoff, large batches\n// after\n.apply(\"DynamoDB write\", DynamoDBIO.<T>write()\n    .withMaxBatchSize(20)\n    .withRetryPolicy(_RETRY_POLICY)); // longer exponential backoff","handlingStrategy":"retry","validationCode":"// pre-check capacity: use DescribeTable / CloudWatch WriteThrottleEvents before large writes","typeGuard":null,"tryCatchPattern":"try { result = p.run().waitUntilFinish(); } catch (Exception e) { inspect DYNAMO_DB_WRITE_FAILURES metric; backfill failed range; }","preventionTips":["Enable auto-scaling / on-demand capacity on target tables","Limit batch size and add rate limiting upstream","Monitor CloudWatch WriteThrottleEvents for the table","Route persistent failures to a dead-letter queue"],"tags":["dynamodb","io","throttling","retry-exhausted"],"backgroundTag":"database-write-failed","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"}