@Override
    public ZNRecord update(ZNRecord znRecord) {
      /* There are several steps to finish an update
       * 1. Fetch the list from the ZNRecord
       * 2. Fetch the AccountMetadata from the blob id if the list exist in the ZNRecord
       * 3. Construct a new AccountMetadata
       * 4. Save it as a blob in the ambry server
       * 5. Remove oldest version if number of version exceeds the maximum value.
       * 6. Add the new blob id to the list.
       */

      // Start step 1:
      ZNRecord recordToUpdate;
      if (znRecord == null) {
        logger.info(
            "ZNRecord does not exist on path={} in HelixPropertyStore when updating accounts. Creating a new ZNRecord.",
            ACCOUNT_METADATA_BLOB_IDS_PATH);
        recordToUpdate = new ZNRecord(ZN_RECORD_ID);
      } else {
        recordToUpdate = znRecord;
      }
      // This is the version number for the new blob id.
      int newVersion = 1;
      List<BlobIDAndVersion> blobIDAndVersions = new ArrayList<>();
      List<String> blobIDAndVersionsJson = recordToUpdate.getListField(ACCOUNT_METADATA_BLOB_IDS_LIST_KEY);
      blobIDAndVersionsJson =
          blobIDAndVersionsJson == null ? new LinkedList<>() : new LinkedList<>(blobIDAndVersionsJson);
      Map<String, String> accountMap = new HashMap<>();

      if (blobIDAndVersionsJson.size() != 0) {
        try {
          // Parse the json string list and get the BlobIDAndVersion with the latest version number.
          blobIDAndVersionsJson.forEach(
              accountBlobIDInJson -> blobIDAndVersions.add(BlobIDAndVersion.fromJson(accountBlobIDInJson)));
          Collections.sort(blobIDAndVersions, Comparator.comparing(BlobIDAndVersion::getVersion));
          BlobIDAndVersion blobIDAndVersion = blobIDAndVersions.get(blobIDAndVersions.size() - 1);
          newVersion = blobIDAndVersion.version + 1;

          // Start Step 2:
          // If this is not for backfill, then just read account metadata from blob, otherwise, initialize it with
          // an empty map and fill it up with the accountMap passed to constructor.
          accountMap = (!forBackFill) ? readAccountMetadataFromBlobID(blobIDAndVersion.blobID) : new HashMap<>();
          if (accountMap == null) {
            logAndThrowIllegalStateException("Fail to read account metadata from blob id " + blobIDAndVersion.blobID,
                null);
          }
        } catch (JSONException e) {
          accountServiceMetrics.remoteDataCorruptionErrorCount.inc();
          logAndThrowIllegalStateException(
              "Exception occurred when parsing the blob id list from " + blobIDAndVersionsJson, e);
        } catch (Exception e) {
          logAndThrowIllegalStateException(
              "Unexpected exception occurred when parsing the blob id list from " + blobIDAndVersionsJson, e);
        }
      }

      if (!forBackFill) {
        // Start step 3:
        AccountInfoMap localAccountInfoMap = null;
        try {
          localAccountInfoMap = new AccountInfoMap(accountServiceMetrics, accountMap);
        } catch (JSONException e) {
          accountServiceMetrics.remoteDataCorruptionErrorCount.inc();
          logAndThrowIllegalStateException(
              "Exception occurred when building AccountInfoMap from accountMap " + accountMap, e);
        }

        // If there is any conflict with the existing record, fail the update. Exception thrown in this updater will
        // be caught by Helix and helixStore#update will return false.
        if (localAccountInfoMap.hasConflictingAccount(this.accounts, config.ignoreVersionMismatch)) {
          // Throw exception, so that helixStore can capture and terminate the update operation
          logAndThrowIllegalStateException(
              "Updating accounts failed because one account to update conflicts with existing accounts", null);
        }
      }

      for (Account account : this.accounts) {
        try {
          String accountJsonStr = objectMapper.writeValueAsString(
              new AccountBuilder(account).snapshotVersion(account.getSnapshotVersion() + 1).build());
          accountMap.put(String.valueOf(account.getId()), accountJsonStr);
        } catch (Exception e) {
          logAndThrowIllegalStateException(
              "Updating accounts failed because unexpected exception occurred when updating accountId="
                  + account.getId() + " accountName=" + account.getName(), e);
        }
      }

      // Start step 4:
      long startTimeMs = System.currentTimeMillis();
      try {
        this.newBlobID = writeAccountMapToRouter(accountMap, router.get());
        accountServiceMetrics.accountUpdateToAmbryTimeInMs.update(System.currentTimeMillis() - startTimeMs);
      } catch (Exception e) {
        accountServiceMetrics.accountUpdatesToAmbryServerErrorCount.inc();
        logAndThrowIllegalStateException(
            "Updating accounts failed because unexpected error occurred when uploading AccountMetadata to ambry", e);
      }

      // Start step 5:
      if (blobIDAndVersions.size() + 1 > totalNumberOfVersionToKeep) {
        Iterator<BlobIDAndVersion> iter = blobIDAndVersions.iterator();
        oldBlobIDsToDelete = new ArrayList<>();
        while (blobIDAndVersions.size() + 1 > totalNumberOfVersionToKeep) {
          BlobIDAndVersion blobIDAndVersion = iter.next();
          iter.remove();
          logger.info("Adding blob {} at version {} to delete", blobIDAndVersion.getBlobID(),
              blobIDAndVersion.getVersion());
          oldBlobIDsToDelete.add(blobIDAndVersion.getBlobID());
        }
        blobIDAndVersionsJson = blobIDAndVersions.stream().map(BlobIDAndVersion::toJson).collect(Collectors.toList());
      }

      // Start step 6:
      blobIDAndVersionsJson.add(new BlobIDAndVersion(this.newBlobID, newVersion).toJson());
      recordToUpdate.setListField(ACCOUNT_METADATA_BLOB_IDS_LIST_KEY, blobIDAndVersionsJson);
      return recordToUpdate;
    }
@param accounts The {@link Account}s to update.
ambry-account/src/main/java/com/github/ambry/account/RouterStore.java