@Override
  public void handleMessage(ReleaseMessage message, String channel) {
    logger.info("message received - channel: {}, message: {}", channel, message);

    String content = message.getMessage();
    if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(content)) {
      return;
    }

    if (!watchedKeys2CacheKey.containsKey(content)) {
      return;
    }

    //create a new list to avoid ConcurrentModificationException
    List<String> cacheKeys = new ArrayList<>(watchedKeys2CacheKey.get(content));

    for (String cacheKey : cacheKeys) {
      logger.debug("invalidate cache key: {}", cacheKey);
      localCache.invalidate(cacheKey);
    }
  }
@Test
  public void testHandleMessage() throws Exception {
    String someWatchKey = "someWatchKey";
    String anotherWatchKey = "anotherWatchKey";
    String someCacheKey = "someCacheKey";
    String anotherCacheKey = "anotherCacheKey";
    String someValue = "someValue";

    ReleaseMessage someReleaseMessage = mock(ReleaseMessage.class);
    when(someReleaseMessage.getMessage()).thenReturn(someWatchKey);

    Cache<String, String> cache =
        (Cache<String, String>) ReflectionTestUtils.getField(configFileController, "localCache");
    cache.put(someCacheKey, someValue);
    cache.put(anotherCacheKey, someValue);

    watchedKeys2CacheKey.putAll(someWatchKey, Lists.newArrayList(someCacheKey, anotherCacheKey));
    watchedKeys2CacheKey.putAll(anotherWatchKey, Lists.newArrayList(someCacheKey, anotherCacheKey));

    cacheKey2WatchedKeys.putAll(someCacheKey, Lists.newArrayList(someWatchKey, anotherWatchKey));
    cacheKey2WatchedKeys.putAll(anotherCacheKey, Lists.newArrayList(someWatchKey, anotherWatchKey));

    configFileController.handleMessage(someReleaseMessage, Topics.APOLLO_RELEASE_TOPIC);

    assertTrue(watchedKeys2CacheKey.isEmpty());
    assertTrue(cacheKey2WatchedKeys.isEmpty());
  }