DynamoDB Enhanced Client并行扫描报无法解组响应错误如何解决

排查思路 #

• 优先确认TableSchema和DynamoDbTable实例是否全局共享
  当前代码中每个线程执行run方法时，都会重复调用TableSchema.fromBean(TfProdHmiBean.class)和enhancedClient.table()构建表实例，2.17.85版本的SDK在并发场景下重复反射解析Bean结构时存在已知并发安全问题，会偶发反序列化空指针。TableSchema和DynamoDbTable本身都是线程安全的，全局初始化一次复用即可。
• 校验并行扫描参数合法性
  确认3个线程传入的segment取值范围是0~2（必须小于totalSegments的取值3），如果参数超出范围会触发服务端返回非法响应，导致反序列化失败。单线程扫描时通常totalSegments=1、segment=0参数天然合法，所以不会触发报错。
• 调整单次扫描的limit阈值
  如果当前itemLimit设置过大，并发场景下单次响应体过大会触发SDK缓冲区截断，导致反序列化时拿到空响应，可以先将itemLimit调整为100验证是否还会报错。
• 打印完整异常栈定位根因
  当前catch块仅打印了异常message，补充打印完整异常栈可以明确是SDK反序列化逻辑的bug，还是业务Bean字段类型和DynamoDB表结构不匹配。
• 升级SDK版本修复已知bug
  2.17.85版本属于较老的2.x迭代版本，存在多个Enhanced Client并发场景下的反序列化bug，可直接将dynamodb-enhanced依赖升级到2.20.x及以上稳定版本，大部分并发相关问题会被直接修复。

优化后代码示例 #

// 全局初始化共享的TableSchema和DynamoDbTable，不要放在线程内部重复创建
private static final TableSchema<TfProdHmiBean> HMI_TABLE_SCHEMA = TableSchema.fromBean(TfProdHmiBean.class);
private static DynamoDbTable<TfProdHmiBean> hmiTable;

// 程序启动时初始化一次即可
public static void initHmiTable(DynamoDbEnhancedClient enhancedClient, String tableName) {
    hmiTable = enhancedClient.table(tableName, HMI_TABLE_SCHEMA);
}

private static class ScanSegmentTask implements Runnable {
    private int itemLimit;
    private int totalSegments;
    private int segment;
    private long scannedItemCount = 0;

    public ScanSegmentTask(int itemLimit, int totalSegments, int seg) {
        this.itemLimit = itemLimit;
        this.totalSegments = totalSegments;
        this.segment = seg;
    }

    @Override
    public void run() {
        System.out.println("Scanning segment " + segment + " out of " + totalSegments
            + " segments " + itemLimit + " items at a time...");
        try {
            ScanEnhancedRequest spec = ScanEnhancedRequest.builder()
                    .limit(itemLimit)
                    .segment(segment)
                    .totalSegments(totalSegments)
                    .build();
            Iterator<TfProdHmiBean> results = hmiTable.scan(spec).items().iterator();
            while (results.hasNext()) {
                scannedItemCount++;
                if (scannedItemCount % 1000000 == 0) {
                    System.out.println("Scanned " + scannedItemCount + " items from segment " + segment + " out of " + totalSegments);
                }
                results.next();
            }
        } catch (Exception e) {
            System.err.println("Error in segment " + segment);
            // 打印完整栈定位问题
            e.printStackTrace();
        } finally {
            System.out.println("Scanned " + scannedItemCount + " items from segment " + segment + " out of "
                + totalSegments);
        }
    }
}

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