@Override
  protected Release findLatestActiveRelease(String appId, String clusterName, String namespaceName,
                                            ApolloNotificationMessages clientMessages) {
    String messageKey = ReleaseMessageKeyGenerator.generate(appId, clusterName, namespaceName);
    String cacheKey = messageKey;

    if (bizConfig.isConfigServiceCacheKeyIgnoreCase()) {
      cacheKey = cacheKey.toLowerCase();
    }

    Tracer.logEvent(TRACER_EVENT_CACHE_GET, cacheKey);

    ConfigCacheEntry cacheEntry = configCache.getUnchecked(cacheKey);

    //cache is out-dated
    if (clientMessages != null && clientMessages.has(messageKey) &&
        clientMessages.get(messageKey) > cacheEntry.getNotificationId()) {
      //invalidate the cache and try to load from db again
      invalidate(cacheKey);
      cacheEntry = configCache.getUnchecked(cacheKey);
    }

    return cacheEntry.getRelease();
  }
@Test
  public void testFindLatestActiveReleaseWithDirtyRelease() throws Exception {
    long someNewNotificationId = someNotificationId + 1;
    ReleaseMessage anotherReleaseMessage = mock(ReleaseMessage.class);
    Release anotherRelease = mock(Release.class);

    when(releaseMessageService.findLatestReleaseMessageForMessages(Lists.newArrayList(someKey))).thenReturn
        (someReleaseMessage);
    when(releaseService.findLatestActiveRelease(someAppId, someClusterName, someNamespaceName)).thenReturn
        (someRelease);
    when(someReleaseMessage.getId()).thenReturn(someNotificationId);

    Release release = configServiceWithCache.findLatestActiveRelease(someAppId, someClusterName, someNamespaceName,
        someNotificationMessages);

    when(releaseMessageService.findLatestReleaseMessageForMessages(Lists.newArrayList(someKey))).thenReturn
        (anotherReleaseMessage);
    when(releaseService.findLatestActiveRelease(someAppId, someClusterName, someNamespaceName)).thenReturn
        (anotherRelease);
    when(anotherReleaseMessage.getId()).thenReturn(someNewNotificationId);

    Release stillOldRelease = configServiceWithCache.findLatestActiveRelease(someAppId, someClusterName,
        someNamespaceName, someNotificationMessages);

    someNotificationMessages.put(someKey, someNewNotificationId);

    Release shouldBeNewRelease = configServiceWithCache.findLatestActiveRelease(someAppId, someClusterName,
        someNamespaceName, someNotificationMessages);

    assertEquals(someRelease, release);
    assertEquals(someRelease, stillOldRelease);
    assertEquals(anotherRelease, shouldBeNewRelease);

    verify(releaseMessageService, times(2)).findLatestReleaseMessageForMessages(Lists.newArrayList(someKey));
    verify(releaseService, times(2)).findLatestActiveRelease(someAppId, someClusterName, someNamespaceName);
  }