public void close() throws IOException {
        try {
            closeAsync().get();
        } catch (ExecutionException e) {
            if (e.getCause() instanceof IOException) {
                throw (IOException) e.getCause();
            } else {
                throw new PulsarServerException(e.getCause());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
@Test
    public void testMetricsPersistentTopicLoadFails() throws Exception {
        final String namespace = "prop/" + UUID.randomUUID().toString().replaceAll("-", "");
        String topic = "persistent://" + namespace + "/topic1_" + UUID.randomUUID();
        admin.namespaces().createNamespace(namespace);
        admin.topics().createNonPartitionedTopic(topic);
        admin.topics().unload(topic);

        // Inject an error that makes the topic load fails.
        AtomicBoolean failMarker = new AtomicBoolean(true);
        mockZooKeeper.failConditional(KeeperException.Code.NODEEXISTS, (op, path) -> {
            if (failMarker.get() && op.equals(MockZooKeeper.Op.SET) &&
                    path.endsWith(TopicName.get(topic).getPersistenceNamingEncoding())) {
                return true;
            }
            return false;
        });

        // Do test
        CompletableFuture<Producer<byte[]>> producer = pulsarClient.newProducer().topic(topic).createAsync();
        JerseyClient httpClient = JerseyClientBuilder.createClient();
        Awaitility.await().until(() -> {
            String response = httpClient.target(pulsar.getWebServiceAddress()).path("/metrics/")
                    .request().get(String.class);
            Multimap<String, PrometheusMetricsClient.Metric> metricMap = PrometheusMetricsClient.parseMetrics(response);
            if (!metricMap.containsKey("pulsar_topic_load_failed_count")) {
                return false;
            }
            double topic_load_failed_count = 0;
            for (PrometheusMetricsClient.Metric metric : metricMap.get("pulsar_topic_load_failed_count")) {
                topic_load_failed_count += metric.value;
            }
            return topic_load_failed_count >= 1D;
        });

        // Remove the injection.
        failMarker.set(false);
        // cleanup.
        httpClient.close();
        producer.join().close();
        admin.topics().delete(topic);
        admin.namespaces().deleteNamespace(namespace);
    }