java - DynamoDB 仅部分更新

Question

我想使用 dynamoDBMapper 库通过 java 更新我的 DynamoDB。我所做的是将消息(我想要执行的更新)推送到一个 SQS，并让我的 java 代码使用这些消息并更新我的 dynamoDB。我发现，当我使用脚本在短时间内推送超过 150 条消息时，所有数据都可以被消耗，但 DynamoDB 中只有部分记录被更新。更新 DynamoDB 的代码如下:

@Service
public class PersistenceMessageProcessingServiceImpl implements PersistenceMessageProcessingService{

    @Override
    public void process(TextMessage textMessage){
        String eventData = textMessage.getText();
        updateEventStatus(eventData);
    }

    /* 
    each input is a caseDetail messages in Event Table

    get data, parse data and update relative records partially in dynamodb.
    finally check if still any open cases, if not change state of event
    */
    private void updateEventStatus(String eventData) throws ParseException, IOException {
        RetryUtils retryUtils = new RetryUtils(maxRetries, waitTimeInMilliSeconds, influxService);
        SNowResponse serviceNowResponse = parseData(eventData);
        EventCaseMap eventCaseMap = eventCaseMapRepository.findBySysId(sysId);
        if (eventCaseMap != null) {
            Event event = eventRepository.findByEventId(eventCaseMap.getSecurityEventManagerId());
            CaseManagementDetails caseManagementDetails = event.getCaseManagementDetails();

            Case existingCaseDetails = getCaseByCaseSystemId(caseManagementDetails, sysId);

            caseDetails.setCaseStatus('Resolved');
            caseDetails.setResolution(serviceNowResponse.getCloseCode());
            caseDetails.setResolvedBy("A");
            caseDetails.setAssessment(serviceNowResponse.getAssessment());
            caseDetails.setResolutionSource("SEM");
            retryUtils.run(() -> {
                    return eventRepository.updateEvent(event); }, RETRY_MEASUREMENT);

        }

        boolean stillOpen = false;
        for(Case existingCase : caseManagementDetails.getCases()){
            if(("OPEN").equals(existingCase.getCaseStatus().toString())){
                stillOpen = true;
                break;
            }
        }
        if(!stillOpen){
            event.setState('CLOSED');
        }


    }


    private Case getCaseByCaseSystemId(CaseManagementDetails caseManagementDetails, String sysId) {
        Case caseDetails = null;
        if (caseManagementDetails != null) {
            List<Case> caseList = caseManagementDetails.getCases();
            for (Case c : caseList) {
                if (c.getCaseSystemId() != null && c.getCaseSystemId().equalsIgnoreCase(sysId)) {
                    caseDetails = c;
                    break;
                }
            }
        }
        return caseDetails;
    }


}

/* EventCaseMap Table in my DynamoDB

data model is like this for EventCaseMap Table:
{
  "caseSystemId": "bb9cc488dbf67b40b3d57709af9619f8",
  "securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}
*/
@Repository
public class EventCaseMapRepositoryImpl implements EventCaseMapRepository {

    @Autowired
    DynamoDBMapper dynamoDBMapper;

    @Override
    public EventCaseMap findBySysId(String sysId) {
        EventCaseMap eventCaseMap = new EventCaseMap();
        eventCaseMap.setCaseSystemId(sysId);
        return dynamoDBMapper.load(eventCaseMap, DynamoDBMapperConfig.ConsistentReads.CONSISTENT.config());
    }

}


/*
data model is like this for Event Table:

{
  "caseManagementDetails": {
    "cases": [
      {
        "caseId": "SIR0123456",
        "caseStatus": "OPEN",
      },
      {
        "caseId": "SIR0654321",
        "caseStatus": "OPEN",
      },
      {
        many other cases(about two hundreds).....
      }
    ]

    },
    "state": "OPEN",
    "securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}

*/


@Repository
public class EventRepositoryImpl implements EventRepository {

    @Autowired
    DynamoDBMapper dynamoDBMapper;

    @Override
    public Event findByEventId(String eventId) {
        Event event = new Event();
        event.setSecurityEventManagerId(eventId);
        return dynamoDBMapper.load(event, DynamoDBMapperConfig.ConsistentReads.CONSISTENT.config());
    }

        @Override
    public boolean updateEvent(Event event) {
        dynamoDBMapper.save(event, DynamoDBMapperConfig.SaveBehavior.UPDATE_SKIP_NULL_ATTRIBUTES.config());
        return false;
    }

}

我已经尝试在 Intellij 的“RUN”和“DEBUG”模型中推送消息并一一消费消息。一切正常，所有案例都可以更新。所以我想知道 DynamoDB 中是否存在任何不一致问题，但我已经在代码中使用了强一致性。

有人知道我的代码发生了什么吗？

有输入、输出、预期输出:

输入:许多像这样的 json 文件:

{
  "number": "SIR0123456",
  "state": "Resolved",
  "sys_id": "bb9cc488dbf67b40b3d57709af9619f8",
  "MessageAttributes": {
    "TransactionGuid": {
      "Type": "String",
      "Value": "093ddb36-626b-4ecc-8943-62e30ffa2e26"
    }
  }
}
{
  "number": "SIR0654321",
  "state": "Resolved",
  "sys_id": "bb9cc488dbf67b40b3d57709af9619f7",
  "MessageAttributes": {
    "TransactionGuid": {
      "Type": "String",
      "Value": "093ddb36-626b-4ecc-8943-62e30ffa2e26"
    }
  }
}

事件表的输出:

{
  "caseManagementDetails": {
    "cases": [
      {
        "caseId": "SIR0123456",
        "caseStatus": "RESOLVED",
      },
      {
        "caseId": "SIR0654321",
        "caseStatus": "OPEN"
      },
      {
        many other cases(about two hundreds).....
      }
    ]

    },
    "state": "OPEN",
    "securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}

事件表的预期输出:

{
  "caseManagementDetails": {
    "cases": [
      {
        "caseId": "SIR0123456",
        "caseStatus": "RESOLVED",
      },
      {
        "caseId": "SIR0654321",
        "caseStatus": "RESOLVED"
      },
      {
        many other cases(about two hundreds).....
      }
    ]

    },
    "state": "OPEN",
    "securityEventManagerId": "756813a4-4e48-4abb-b37e-da00e931583b"
}

Answer 1

我认为问题在于DynamoDB持久化数据时，它以多线程的方式运行。因此，如果我们在短时间内消耗所有这些数据，可能会有一些线程没有完成。所以我们看到的结果只是最后一个线程的结果，而不是所有线程的结果。