docs: function loadBatchStateFromDynamoDB(batch, context) {
    const task = this;
  
    const startTime = Date.now();
  
    try {
      const streamConsumerId = batch.streamConsumerId || (batch.key && batch.key.streamConsumerId);
      const shardOrEventID = batch.shardOrEventID || (batch.key && batch.key.shardOrEventID);
  
      const batchDesc = `batch (${shardOrEventID}) with ${batch.describeContents()}`;
  
      // Resolve the stage-qualified table name
      const unqualifiedTableName = getBatchStateTableName(context);
      const tableName = stages.toStageQualifiedResourceName(unqualifiedTableName, context.stage, context);
  
      const getRequest = {
        TableName: tableName,
        Key: {streamConsumerId: streamConsumerId, shardOrEventID: shardOrEventID},
        ConsistentRead: true,
        ReturnConsumedCapacity: "NONE", //"TOTAL" or "INDEXES" or "NONE"
        ProjectionExpression: "#streamConsumerId, #shardOrEventID, #messageStates, #rejectedMessageStates, #unusableRecordStates",
        ExpressionAttributeNames: {
          "#streamConsumerId": "streamConsumerId",
          "#shardOrEventID": "shardOrEventID",
          "#messageStates": "messageStates",
          "#rejectedMessageStates": "rejectedMessageStates",
          "#unusableRecordStates": "unusableRecordStates"
        }
      };
  
      if (context.traceEnabled) context.trace("get request: " + JSON.stringify(getRequest));
  
      const dynamoDBDocClient = getDynamoDBDocClient(context);
  
      // Load the batch state for the entire batch of messages, rejected messages and unusable records from DynamoDB
      return dynamoDBDocClient.get(getRequest).promise()
        .then(result => {
          const hasPreviouslySavedItem = !!(result && result.Item);
          if (context.traceEnabled) {
            let loadedDesc = 'found no previously saved state';
            if (hasPreviouslySavedItem) {
              const item = result.Item;
              const m = item.messageStates ? item.messageStates.length : 0;
              const r = item.rejectedMessageStates ? item.rejectedMessageStates.length : 0;
              const u = item.unusableRecordStates ? item.unusableRecordStates.length : 0;
              loadedDesc = `found ${toCountString(m, 'msg state')}, ${toCountString(r, 'rejected msg state')} & ${toCountString(u, 'unusable rec state')}`;
            }
            context.debug('Loaded state of', batchDesc, 'from', tableName, '-', loadedDesc, took(startTime));
          }
  
          batch.previouslySaved = hasPreviouslySavedItem;
          if (hasPreviouslySavedItem) {
            updateBatchWithPriorState(batch, result.Item, context);
          }
          return result;
        })
        .catch(err => {
          batch.previouslySaved = undefined;
  
          if (awsErrors.isResourceNotFoundException(err)) {
            const errMsg = `FATAL - Cannot load state of batch, since missing DynamoDB table (${tableName}) - ${err.message}`;
            context.error(errMsg, '-', batchDesc, took(startTime), err);
            throw new FatalError(errMsg, err);
          }
          if (awsErrors.isRetryable(err)) {
            const errMsg = `TRANSIENT - Failed to load state of batch from ${tableName} - ${err.message}`;
            context.error(errMsg, '-', batchDesc, took(startTime), err);
            if (task) task.revertAttempts(true);
            throw new TransientError(errMsg, err);
          }
          context.error('Failed to load state of', batchDesc, 'from', tableName, took(startTime), err);
          throw err;
        });
  
    } catch (err) {
      batch.previouslySaved = undefined;
      context.error('Failed to load state of batch', took(startTime), err);
      return Promise.reject(err);
    }
  }