DynamoDB TransactWriteItem并发问题排查与解决方案咨询

1. B#root记录无幂等控制：当前代码用UUID.randomUUID()生成B#root的hashKey，每次触发都会生成全新主键，且未添加任何条件判断——哪怕是同一条A记录的重复流事件，两次调用都会写入新的B#root记录。
2. Parent#记录的条件表达式错误：你写的attribute_not_exists('hashKey') AND attribute_not_exists('sortKey')里，属性名加了单引号是完全错误的。DynamoDB的attribute_not_exists参数是属性名称，加单引号后，DynamoDB会认为你在检查是否存在名为'hashKey'（带单引号）的属性，而非实际的hashKey主键属性，导致这个条件永远为真，根本起不到防重复的作用。
3. 最终出现三条记录的原因：第一次调用成功写入B#root1和Parent#；第二次调用时，B#root因主键是新UUID直接写入成功，而Parent#的无效条件也可能写入成功（或部分场景下失败），最终导致总数超出预期。

1. 绑定A记录ID生成B#root主键，实现幂等 #

把B#root的hashKey与A记录的唯一ID绑定，而非随机UUID，确保同一条A记录触发的所有重复事件，生成的B#root主键完全一致：

String aRecordId = <Id given from dynamodb streams>;
rootItem.put("hashKey", new AttributeValue("B#" + aRecordId + ";Default"));
rootItem.put("sortKey", new AttributeValue("B#root"));

2. 给B#root的Put操作添加条件判断 #

和Parent#一样，加上正确的条件表达式，确保只有第一次调用能写入：

Put bRootItem = new Put()
    .withTableName("B")
    .withItem(rootItem)
    .withConditionExpression("attribute_not_exists(hashKey) AND attribute_not_exists(sortKey)");

3. 修正Parent#的条件表达式 #

去掉属性名的单引号，恢复正确的条件判断逻辑：

Put bParentItem = new Put()
    .withTableName("B")
    .withItem(parentItem)
    .withConditionExpression("attribute_not_exists(hashKey) AND attribute_not_exists(sortKey)");

4. 保留事务写入保证原子性 #

继续使用TransactWriteItem，确保B#root和Parent#要么都写入成功，要么都失败，避免出现单条记录写入的异常情况。

完整修正代码示例 #

String aRecordId = <Id given from dynamodb streams>;

// 构建B#root记录，主键绑定A记录ID
HashMap<String, AttributeValue> rootItem = new HashMap<>();
rootItem.put("hashKey", new AttributeValue("B#" + aRecordId + ";Default"));
rootItem.put("sortKey", new AttributeValue("B#root"));
rootItem.put("name", new AttributeValue("Default"));
rootItem.put("parentValue", new AttributeValue(aRecordId + ";Default"));

// 构建Parent#记录
HashMap<String, AttributeValue> parentItem = new HashMap<>();
parentItem.put("hashKey", new AttributeValue("Parent#" + aRecordId + ";Default"));
parentItem.put("sortKey", new AttributeValue("Parent#" + aRecordId + ";Default"));

// 给B#root添加条件表达式
Put bRootItem = new Put()
    .withTableName("B")
    .withItem(rootItem)
    .withConditionExpression("attribute_not_exists(hashKey) AND attribute_not_exists(sortKey)");

// 修正Parent#的条件表达式
Put bParentItem = new Put()
    .withTableName("B")
    .withItem(parentItem)
    .withConditionExpression("attribute_not_exists(hashKey) AND attribute_not_exists(sortKey)");

// 事务写入
Collection<TransactWriteItem> actions = Arrays.asList(
    new TransactWriteItem().withPut(bRootItem),
    new TransactWriteItem().withPut(bParentItem)
);

额外建议 #

• 可以维护一个幂等表（或在Table A中添加标记字段），记录已处理过的A记录ID，直接跳过重复的流事件，减少无效执行。
• 捕获ConditionalCheckFailedException，这类异常属于预期的幂等失败，无需重试。

内容的提问来源于stack exchange，提问作者Rehab Reda