public CompletableFuture<NamespaceEphemeralData> tryAcquiringOwnership(NamespaceBundle namespaceBundle) {
        log.info("Try acquiring ownership for bundle: {} - {}.", namespaceBundle, brokerRegistry.getBrokerId());
        final String bundle = namespaceBundle.toString();
        return assign(Optional.empty(), namespaceBundle, LookupOptions.builder().readOnly(false).build())
                .thenApply(brokerLookupData -> {
                    if (brokerLookupData.isEmpty()) {
                        String errorMsg = String.format(
                                "Failed to get the broker lookup data for bundle:%s", bundle);
                        log.error(errorMsg);
                        throw new IllegalStateException(errorMsg);
                    }
                    return brokerLookupData.get().toNamespaceEphemeralData();
                });
    }
@Test(timeOut = 30 * 1000)
    public void testTryAcquiringOwnership()
            throws PulsarAdminException, ExecutionException, InterruptedException {
        final String namespace = "public/testTryAcquiringOwnership";
        admin.namespaces().createNamespace(namespace, 1);
        String topic = "persistent://" + namespace + "/test";
        NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get();
        NamespaceEphemeralData namespaceEphemeralData = primaryLoadManager.tryAcquiringOwnership(bundle).get();
        assertTrue(Set.of(pulsar1.getBrokerServiceUrl(), pulsar2.getBrokerServiceUrl())
                .contains(namespaceEphemeralData.getNativeUrl()));
        Awaitility.await()
                .atMost(5, TimeUnit.SECONDS)
                .untilAsserted(() -> {
                    admin.namespaces().deleteNamespace(namespace, true);
                });
    }