@Test
  public void testAccountRefresh() throws Exception {
    MySqlAccountService producerAccountService = mySqlAccountService;
    MySqlAccountStore producerAccountStore = mySqlAccountStore;
    // Create second account service with scheduled polling disabled
    mySqlConfigProps.setProperty(UPDATER_POLLING_INTERVAL_SECONDS, "0");
    accountServiceConfig = new MySqlAccountServiceConfig(new VerifiableProperties(mySqlConfigProps));
    MySqlAccountStore consumerAccountStore =
        spy(new MySqlAccountStoreFactory(new VerifiableProperties(mySqlConfigProps),
            new MetricRegistry()).getMySqlAccountStore());
    MySqlAccountStoreFactory mockMySqlAccountStoreFactory = mock(MySqlAccountStoreFactory.class);
    when(mockMySqlAccountStoreFactory.getMySqlAccountStore()).thenReturn(consumerAccountStore);
    AccountServiceMetricsWrapper accountServiceMetricsWrapper = new AccountServiceMetricsWrapper(new MetricRegistry());
    MySqlAccountService consumerAccountService =
        new MySqlAccountService(accountServiceMetricsWrapper, accountServiceConfig, mockMySqlAccountStoreFactory,
            mockNotifier);

    // Add account with 3 containers
    short accountId = 101;
    String accountName = "a1";
    // Number of calls expected in producer account store
    List<Container> containers = new ArrayList<>();
    containers.add(new ContainerBuilder((short) 1, "c1", ContainerStatus.ACTIVE, DESCRIPTION, accountId).build());
    containers.add(new ContainerBuilder((short) 2, "c2", ContainerStatus.ACTIVE, DESCRIPTION, accountId).build());
    containers.add(new ContainerBuilder((short) 3, "c3", ContainerStatus.ACTIVE, DESCRIPTION, accountId).build());
    Account a1 =
        new AccountBuilder(accountId, accountName, Account.AccountStatus.ACTIVE).containers(containers).build();
    producerAccountService.updateAccounts(Collections.singletonList(a1));
    Account finalA = a1;
    verify(producerAccountStore).updateAccounts(argThat(accountInfos -> {
      AccountUpdateInfo accountInfo = accountInfos.get(0);
      return accountInfo.getAccount().equals(finalA) && accountInfo.getAddedContainers().equals(containers)
          && accountInfo.isAdded() && !accountInfo.isUpdated() && accountInfo.getUpdatedContainers().isEmpty();
    }));
    // Note: because consumer is notified of changes, its cache should already be in sync with DB
    //this use the abstract accountInfoMapRef
    long lmt = consumerAccountService.getCachedAccountService().accountInfoMapRef.get().getLastModifiedTime();
    assertEquals("Account mismatch", a1, consumerAccountService.getAccountByName(accountName));

    // Update account only
    String newAccountName = "a1-updated";
    a1 = new AccountBuilder(a1).name(newAccountName).build();
    producerAccountService.updateAccounts(Collections.singletonList(a1));
    Account finalA1 = a1;
    verify(producerAccountStore).updateAccounts(argThat(accountInfos -> {
      AccountUpdateInfo accountInfo = accountInfos.get(0);
      return accountInfo.getAccount().equals(finalA1) && accountInfo.getAddedContainers().isEmpty()
          && !accountInfo.isAdded() && accountInfo.isUpdated() && accountInfo.getUpdatedContainers().isEmpty();
    }));
    verify(consumerAccountStore).getNewAccounts(eq(lmt));
    verify(consumerAccountStore).getNewContainers(eq(lmt));
    assertEquals("Account mismatch", a1, consumerAccountService.getAccountByName(newAccountName));
    assertNull("Expected no account with old name", consumerAccountService.getAccountByName(accountName));
    accountName = newAccountName;

    // Update container only
    Container c1Mod = new ContainerBuilder(containers.get(0)).setStatus(ContainerStatus.DELETE_IN_PROGRESS).build();
    containers.set(0, c1Mod);
    a1 = new AccountBuilder(a1).containers(containers).build();
    Collection<Container> result =
        producerAccountService.updateContainers(accountName, Collections.singletonList(c1Mod));
    // Account should not have been touched
    verify(producerAccountStore).updateAccounts(
        argThat(accountsInfo -> accountsInfo.get(0).getUpdatedContainers().size() == 1));
    assertEquals("Expected one result", 1, result.size());
    assertEquals("Container mismatch", c1Mod, result.iterator().next());
    verify(consumerAccountStore).getNewAccounts(eq(lmt));
    verify(consumerAccountStore).getNewContainers(eq(lmt));
    assertEquals("Container mismatch", c1Mod, consumerAccountService.getContainerByName(accountName, "c1"));
    assertEquals("Account mismatch", a1, consumerAccountService.getAccountByName(accountName));
    lmt = consumerAccountService.getCachedAccountService().accountInfoMapRef.get().getLastModifiedTime();

    // Add container only
    Container cNew = makeNewContainer("c4", accountId, ContainerStatus.ACTIVE);
    result = producerAccountService.updateContainers(accountName, Collections.singletonList(cNew));
    verify(producerAccountStore).updateAccounts(
        argThat(accountsInfo -> accountsInfo.get(0).getAddedContainers().size() == 1));
    assertEquals("Expected one result", 1, result.size());
    cNew = result.iterator().next();
    containers.add(cNew);
    a1 = new AccountBuilder(a1).containers(containers).build();
    verify(consumerAccountStore).getNewAccounts(eq(lmt));
    verify(consumerAccountStore).getNewContainers(eq(lmt));
    assertEquals("Container mismatch", cNew, consumerAccountService.getContainerByName(accountName, "c4"));
    assertEquals("Account mismatch", a1, consumerAccountService.getAccountByName(accountName));

    // For this section, consumer must get out of date so need to unsubscribe its notifier
    mockNotifier.unsubscribeAll();
    // TODO:
    // Add container in AS1, call AS2.getContainer() to force fetch
    // Add C1 in AS1, add C1 in AS2 (should succeed and return existing id)
    Container cNewProd = makeNewContainer("c5", accountId, ContainerStatus.ACTIVE);
    result = producerAccountService.updateContainers(accountName, Collections.singletonList(cNewProd));
    short newId = result.iterator().next().getId();
    // Add the same container to second AS with stale cache
    // Expect it to fail first time (conflict), refresh cache and succeed on retry
    result = consumerAccountService.updateContainers(accountName, Collections.singletonList(cNewProd));
    assertEquals(newId, result.iterator().next().getId());
    assertEquals(1, accountServiceMetricsWrapper.getAccountServiceMetrics().updateAccountErrorCount.getCount());
    assertEquals(1, accountServiceMetricsWrapper.getAccountServiceMetrics().conflictRetryCount.getCount());

    // Add C1 in AS1, add C2 in AS2
    cNewProd = makeNewContainer("c6", accountId, ContainerStatus.ACTIVE);
    result = producerAccountService.updateContainers(accountName, Collections.singletonList(cNewProd));
    newId = result.iterator().next().getId();
    Container cNewCons = makeNewContainer("c7", accountId, ContainerStatus.ACTIVE);
    result = consumerAccountService.updateContainers(accountName, Collections.singletonList(cNewCons));
    assertNotSame(newId, result.iterator().next().getId());
    assertEquals(2, accountServiceMetricsWrapper.getAccountServiceMetrics().updateAccountErrorCount.getCount());
    assertEquals(2, accountServiceMetricsWrapper.getAccountServiceMetrics().conflictRetryCount.getCount());

    // Check gauge values
    assertTrue("Sync time not updated",
        accountServiceMetricsWrapper.getAccountServiceMetrics().timeInSecondsSinceLastSyncGauge.getValue() < 10);
    assertEquals("Unexpected container count", 7,
        accountServiceMetricsWrapper.getAccountServiceMetrics().containerCountGauge.getValue().intValue());
  }
Producer-consumer test for multiple account services.
ambry-account/src/integration-test/java/com/github/ambry/account/MySqlAccountServiceIntegrationTest.java