public static ExtensibleLoadManagerImpl get(LoadManager loadManager) {
        if (!(loadManager instanceof ExtensibleLoadManagerWrapper loadManagerWrapper)) {
            throw new IllegalArgumentException("The load manager should be 'ExtensibleLoadManagerWrapper'.");
        }
        return loadManagerWrapper.get();
    }
@Test(enabled = false)
    public static void testOptimizeUnloadDisable(TopicDomain topicDomain,
                                                 String defaultTestNamespace,
                                                 PulsarAdmin admin,
                                                 String brokerServiceUrl,
                                                 PulsarService pulsar1,
                                                 PulsarService pulsar2) throws Exception {
        var id = String.format("test-optimize-unload-disable-%s-%s", topicDomain, UUID.randomUUID());
        var topic = String.format("%s://%s/%s", topicDomain, defaultTestNamespace, id);
        var topicName = TopicName.get(topic);

        pulsar1.getConfig().setLoadBalancerMultiPhaseBundleUnload(false);
        pulsar2.getConfig().setLoadBalancerMultiPhaseBundleUnload(false);

        @Cleanup
        var pulsarClient = pulsarClient(brokerServiceUrl, 0);

        @Cleanup
        var producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();

        @Cleanup
        var consumer = pulsarClient.newConsumer(Schema.STRING).subscriptionName(id).topic(topic).subscribe();

        Awaitility.await().until(() -> producer.isConnected() && consumer.isConnected());

        var lookup = spyLookupService(pulsarClient);

        final CountDownLatch cdl = new CountDownLatch(3);

        NamespaceBundle bundle = getBundleAsync(pulsar1, TopicName.get(topic)).get();
        var srcBrokerServiceUrl = admin.lookups().lookupTopic(topic);
        var dstBroker = srcBrokerServiceUrl.equals(pulsar1.getBrokerServiceUrl()) ? pulsar2 : pulsar1;

        CompletableFuture<Void> unloadNamespaceBundle = CompletableFuture.runAsync(() -> {
            try {
                cdl.await();
                admin.namespaces().unloadNamespaceBundle(defaultTestNamespace, bundle.getBundleRange(),
                        dstBroker.getBrokerId());
            } catch (InterruptedException | PulsarAdminException e) {
                fail();
            }
        });

        MutableInt sendCount = new MutableInt();
        Awaitility.await().atMost(20, TimeUnit.SECONDS).ignoreExceptions().until(() -> {
            var message = String.format("message-%d", sendCount.getValue());

            AtomicBoolean messageSent = new AtomicBoolean(false);
            while (true) {
                var recvFuture = consumer.receiveAsync().orTimeout(1000, TimeUnit.MILLISECONDS);
                if (!messageSent.get()) {
                    producer.sendAsync(message).thenAccept(messageId -> {
                        if (messageId != null) {
                            messageSent.set(true);
                        }
                    }).get(1000, TimeUnit.MILLISECONDS);
                }

                if (topicDomain == TopicDomain.non_persistent) {
                    // No need to wait for message receipt, we're only trying to stress the consumer lookup pathway.
                    break;
                }
                var msg = recvFuture.get();
                if (Objects.equals(msg.getValue(), message)) {
                    break;
                }
            }

            cdl.countDown();
            return sendCount.incrementAndGet() == 10;
        });

        assertTrue(producer.isConnected());
        assertTrue(consumer.isConnected());
        assertTrue(unloadNamespaceBundle.isDone());
        verify(lookup, times(2)).getBroker(topicName);
    }