@Test
  public void testAccountUpdateConsumer() throws Exception {
    // pre-populate account metadata in ZK.
    writeAccountsToHelixPropertyStore(idToRefAccountMap.values(), false);
    accountService = mockHelixAccountServiceFactory.getAccountService();
    assertAccountsInAccountService(idToRefAccountMap.values(), NUM_REF_ACCOUNT, accountService);

    // add consumer
    int numOfConsumers = 10;
    List<Collection<Account>> updatedAccountsReceivedByConsumers = new ArrayList<>();
    List<Consumer<Collection<Account>>> accountUpdateConsumers = IntStream.range(0, numOfConsumers)
        .mapToObj(i -> (Consumer<Collection<Account>>) updatedAccountsReceivedByConsumers::add)
        .peek(accountService::addAccountUpdateConsumer)
        .collect(Collectors.toList());

    // listen to adding a new account
    Account newAccount = new AccountBuilder(refAccountId, refAccountName, refAccountStatus).build();
    Set<Account> accountsToUpdate = Collections.singleton(newAccount);
    updateAccountsAndAssertAccountExistence(accountsToUpdate, 1 + NUM_REF_ACCOUNT, true);
    assertAccountUpdateConsumers(accountsToUpdate, numOfConsumers, updatedAccountsReceivedByConsumers);

    // listen to modification of existing accounts. Only updated accounts will be received by consumers.
    updatedAccountsReceivedByConsumers.clear();
    accountsToUpdate = new HashSet<>();
    for (Account account : accountService.getAllAccounts()) {
      AccountBuilder accountBuilder = new AccountBuilder(account);
      accountBuilder.name(account.getName() + "-extra");
      accountsToUpdate.add(accountBuilder.build());
    }
    updateAccountsAndAssertAccountExistence(accountsToUpdate, 1 + NUM_REF_ACCOUNT, true);
    assertAccountUpdateConsumers(accountsToUpdate, numOfConsumers, updatedAccountsReceivedByConsumers);

    // removes the consumers so the consumers will not be informed.
    updatedAccountsReceivedByConsumers.clear();
    for (Consumer<Collection<Account>> accountUpdateConsumer : accountUpdateConsumers) {
      accountService.removeAccountUpdateConsumer(accountUpdateConsumer);
    }
    Account account = accountService.getAccountById(refAccountId);
    Account updatedAccount = new AccountBuilder(account).name(account.getName() + "-extra").build();
    accountsToUpdate = new HashSet<>(Collections.singleton(updatedAccount));
    updateAccountsAndAssertAccountExistence(accountsToUpdate, 1 + NUM_REF_ACCOUNT, true);
    assertAccountUpdateConsumers(Collections.emptySet(), 0, updatedAccountsReceivedByConsumers);
  }
Tests adding/removing {@link Consumer}.
@throws Exception
ambry-account/src/test/java/com/github/ambry/account/HelixAccountServiceTest.java