@VisibleForTesting
    CompletableFuture<Void> checkPersistencePolicies() {
        TopicName topicName = TopicName.get(topic);
        CompletableFuture<Void> future = new CompletableFuture<>();
        brokerService.getManagedLedgerConfig(topicName).thenAccept(config -> {
            // update managed-ledger config and managed-cursor.markDeleteRate
            this.ledger.setConfig(config);
            future.complete(null);
        }).exceptionally(ex -> {
            log.warn("[{}] Failed to update persistence-policies {}", topic, ex.getMessage());
            future.completeExceptionally(ex);
            return null;
        });
        return future;
    }
@Test
    public void testCheckPersistencePolicies() throws Exception {
        final String myNamespace = "prop/ns";
        admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
        final String topic = "persistent://" + myNamespace + "/testConfig" + UUID.randomUUID();
        conf.setForceDeleteNamespaceAllowed(true);
        pulsarClient.newProducer().topic(topic).create().close();
        RetentionPolicies retentionPolicies = new RetentionPolicies(1, 1);
        PersistentTopic persistentTopic = spy((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get());
        TopicPoliciesService policiesService = spy(pulsar.getTopicPoliciesService());
        doReturn(policiesService).when(pulsar).getTopicPoliciesService();
        TopicPolicies policies = new TopicPolicies();
        policies.setRetentionPolicies(retentionPolicies);
        doReturn(CompletableFuture.completedFuture(Optional.of(policies))).when(policiesService).getTopicPoliciesAsync(TopicName.get(topic));
        persistentTopic.onUpdate(policies);
        verify(persistentTopic, times(1)).checkPersistencePolicies();
        Awaitility.await().untilAsserted(() -> {
            assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionSizeInMB(), 1L);
            assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionTimeMillis(), TimeUnit.MINUTES.toMillis(1));
        });
        // throw exception
        doReturn(CompletableFuture.failedFuture(new RuntimeException())).when(persistentTopic).checkPersistencePolicies();
        policies.setRetentionPolicies(new RetentionPolicies(2, 2));
        persistentTopic.onUpdate(policies);
        assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionSizeInMB(), 1L);
        assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionTimeMillis(), TimeUnit.MINUTES.toMillis(1));
    }