@Override
    public Set<ServiceInstance> getServiceInstance(String serviceName) {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // first check the cache and return if service instance exists
        if (XdsDataCache.isContainsRequestObserver(serviceName)) {
            return XdsDataCache.getServiceInstance(serviceName);
        }

        // locking ensures that a service only creates one stream
        LOCK.lock();
        try {
            // check the cache again after locking and return if service instance exists
            if (XdsDataCache.isContainsRequestObserver(serviceName)) {
                return XdsDataCache.getServiceInstance(serviceName);
            }
            edsHandler.subscribe(serviceName, countDownLatch);
        } finally {
            LOCK.unlock();
        }
        try {
            countDownLatch.await(TIMEOUT, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            LOGGER.log(Level.WARNING, "Occur InterruptedException when wait server send message.", e);
        }
        return XdsDataCache.getServiceInstance(serviceName);
    }
@Test
    public void getServiceInstance() {
        // clear data
        XdsDataCache.removeServiceInstance(serviceName);
        XdsDataCache.removeRequestObserver(serviceName);

        // no service instance in cache
        Set<ServiceInstance> result = xdsServiceDiscovery.getServiceInstance(serviceName);
        Assert.assertNotNull(XdsDataCache.getRequestObserver(serviceName));

        // service instance in cache
        Set<ServiceInstance> instances = new HashSet<>();
        instances.add(new XdsServiceInstance());
        XdsDataCache.updateServiceInstance(serviceName, instances);
        result = xdsServiceDiscovery.getServiceInstance(serviceName);
        Assert.assertEquals(1, result.size());
    }