public static ExtensibleLoadManagerImpl get(LoadManager loadManager) {
        if (!(loadManager instanceof ExtensibleLoadManagerWrapper loadManagerWrapper)) {
            throw new IllegalArgumentException("The load manager should be 'ExtensibleLoadManagerWrapper'.");
        }
        return loadManagerWrapper.get();
    }
@Test(timeOut = 30 * 1000)
    public void testRoleChange() throws Exception {
        makePrimaryAsLeader();

        var leader = primaryLoadManager;
        var follower = secondaryLoadManager;

        BrokerLoadData brokerLoadExpected = new BrokerLoadData();
        SystemResourceUsage usage = new SystemResourceUsage();
        var cpu = new ResourceUsage(1.0, 100.0);
        String key = "b1";
        usage.setCpu(cpu);
        brokerLoadExpected.update(usage, 0, 0, 0, 0, 0, 0, conf);
        String bundle = "public/default/0x00000000_0xffffffff";
        TopBundlesLoadData topBundlesExpected = new TopBundlesLoadData();
        topBundlesExpected.getTopBundlesLoadData().clear();
        topBundlesExpected.getTopBundlesLoadData().add(new TopBundlesLoadData.BundleLoadData(bundle, new NamespaceBundleStats()));

        follower.getBrokerLoadDataStore().pushAsync(key, brokerLoadExpected);
        follower.getTopBundlesLoadDataStore().pushAsync(bundle, topBundlesExpected);

        Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {

            assertNotNull(FieldUtils.readDeclaredField(leader.getTopBundlesLoadDataStore(), "tableView", true));
            assertNull(FieldUtils.readDeclaredField(follower.getTopBundlesLoadDataStore(), "tableView", true));

            for (String internalTopic : ExtensibleLoadManagerImpl.INTERNAL_TOPICS) {
                assertTrue(leader.pulsar.getBrokerService().getTopicReference(internalTopic)
                        .isPresent());
                assertTrue(follower.pulsar.getBrokerService().getTopicReference(internalTopic)
                        .isEmpty());

                assertTrue(leader.pulsar.getNamespaceService()
                        .isServiceUnitOwnedAsync(TopicName.get(internalTopic)).get());
                assertFalse(follower.pulsar.getNamespaceService()
                        .isServiceUnitOwnedAsync(TopicName.get(internalTopic)).get());
            }

            var actualBrokerLoadLeader = leader.getBrokerLoadDataStore().get(key);
            if (actualBrokerLoadLeader.isPresent()) {
                assertEquals(actualBrokerLoadLeader.get(), brokerLoadExpected);
            }

            var actualTopBundlesLeader = leader.getTopBundlesLoadDataStore().get(bundle);
            if (actualTopBundlesLeader.isPresent()) {
                assertEquals(actualTopBundlesLeader.get(), topBundlesExpected);
            }

            var actualBrokerLoadFollower = follower.getBrokerLoadDataStore().get(key);
            if (actualBrokerLoadFollower.isPresent()) {
                assertEquals(actualBrokerLoadFollower.get(), brokerLoadExpected);
            }
        });

        makeSecondaryAsLeader();

        var leader2 = secondaryLoadManager;
        var follower2 = primaryLoadManager;

        brokerLoadExpected.update(usage, 1, 0, 0, 0, 0, 0, conf);
        topBundlesExpected.getTopBundlesLoadData().get(0).stats().msgRateIn = 1;

        follower.getBrokerLoadDataStore().pushAsync(key, brokerLoadExpected);
        follower.getTopBundlesLoadDataStore().pushAsync(bundle, topBundlesExpected);

        Awaitility.await().atMost(30, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(() -> {
            assertNotNull(FieldUtils.readDeclaredField(leader2.getTopBundlesLoadDataStore(), "tableView", true));
            assertNull(FieldUtils.readDeclaredField(follower2.getTopBundlesLoadDataStore(), "tableView", true));

            for (String internalTopic : ExtensibleLoadManagerImpl.INTERNAL_TOPICS) {
                assertTrue(leader2.pulsar.getBrokerService().getTopicReference(internalTopic)
                        .isPresent());
                assertTrue(follower2.pulsar.getBrokerService().getTopicReference(internalTopic)
                        .isEmpty());

                assertTrue(leader2.pulsar.getNamespaceService()
                        .isServiceUnitOwnedAsync(TopicName.get(internalTopic)).get());
                assertFalse(follower2.pulsar.getNamespaceService()
                        .isServiceUnitOwnedAsync(TopicName.get(internalTopic)).get());
            }


            var actualBrokerLoadLeader = leader2.getBrokerLoadDataStore().get(key);
            assertEquals(actualBrokerLoadLeader.get(), brokerLoadExpected);

            var actualTopBundlesLeader = leader2.getTopBundlesLoadDataStore().get(bundle);
            assertEquals(actualTopBundlesLeader.get(), topBundlesExpected);

            var actualBrokerLoadFollower = follower2.getBrokerLoadDataStore().get(key);
            assertEquals(actualBrokerLoadFollower.get(), brokerLoadExpected);
        });
    }