@GetMapping
  public DeferredResult<ResponseEntity<ApolloConfigNotification>> pollNotification(
      @RequestParam(value = "appId") String appId,
      @RequestParam(value = "cluster") String cluster,
      @RequestParam(value = "namespace", defaultValue = ConfigConsts.NAMESPACE_APPLICATION) String namespace,
      @RequestParam(value = "dataCenter", required = false) String dataCenter,
      @RequestParam(value = "notificationId", defaultValue = "-1") long notificationId,
      @RequestParam(value = "ip", required = false) String clientIp) {
    //strip out .properties suffix
    namespace = namespaceUtil.filterNamespaceName(namespace);

    Set<String> watchedKeys = watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespace, dataCenter);

    DeferredResult<ResponseEntity<ApolloConfigNotification>> deferredResult =
        new DeferredResult<>(TIMEOUT, NOT_MODIFIED_RESPONSE);

    //check whether client is out-dated
    ReleaseMessage latest = releaseMessageService.findLatestReleaseMessageForMessages(watchedKeys);

    /**
     * Manually close the entity manager.
     * Since for async request, Spring won't do so until the request is finished,
     * which is unacceptable since we are doing long polling - means the db connection would be hold
     * for a very long time
     */
    entityManagerUtil.closeEntityManager();

    if (latest != null && latest.getId() != notificationId) {
      deferredResult.setResult(new ResponseEntity<>(
          new ApolloConfigNotification(namespace, latest.getId()), HttpStatus.OK));
    } else {
      //register all keys
      for (String key : watchedKeys) {
        this.deferredResults.put(key, deferredResult);
      }

      deferredResult
          .onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));

      deferredResult.onCompletion(() -> {
        //unregister all keys
        for (String key : watchedKeys) {
          deferredResults.remove(key, deferredResult);
        }
        logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys");
      });

      logWatchedKeys(watchedKeys, "Apollo.LongPoll.RegisteredKeys");
      logger.debug("Listening {} from appId: {}, cluster: {}, namespace: {}, datacenter: {}",
          watchedKeys, appId, cluster, namespace, dataCenter);
    }

    return deferredResult;
  }
@Test
  public void testPollNotificationWithDefaultNamespace() throws Exception {
    String someWatchKey = "someKey";
    String anotherWatchKey = "anotherKey";

    Set<String> watchKeys = Sets.newHashSet(someWatchKey, anotherWatchKey);

    when(watchKeysUtil
        .assembleAllWatchKeys(someAppId, someCluster, defaultNamespace,
            someDataCenter)).thenReturn(
        watchKeys);

    DeferredResult<ResponseEntity<ApolloConfigNotification>>
        deferredResult = controller
        .pollNotification(someAppId, someCluster, defaultNamespace, someDataCenter,
            someNotificationId, someClientIp);

    assertEquals(watchKeys.size(), deferredResults.size());

    for (String watchKey : watchKeys) {
      assertTrue(deferredResults.get(watchKey).contains(deferredResult));
    }
  }