@VisibleForTesting
    protected void monitor() {
        try {
            initWaiter.get();

            // Monitor role
            // Periodically check the role in case ZK watcher fails.
            var isChannelOwner = serviceUnitStateChannel.isChannelOwner();
            if (isChannelOwner) {
                // System topic config might fail due to the race condition
                // with topic policy init(Topic policies cache have not init).
                if (!configuredSystemTopics) {
                    configuredSystemTopics = configureSystemTopics(pulsar);
                }
                if (role != Leader) {
                    log.warn("Current role:{} does not match with the channel ownership:{}. "
                            + "Playing the leader role.", role, isChannelOwner);
                    playLeader();
                }
            } else {
                if (role != Follower) {
                    log.warn("Current role:{} does not match with the channel ownership:{}. "
                            + "Playing the follower role.", role, isChannelOwner);
                    playFollower();
                }
            }
        } catch (Throwable e) {
            log.error("Failed to get the channel ownership.", e);
        }
    }
@Test(timeOut = 30 * 1000)
    public void compactionScheduleTest() {
        Awaitility.await()
                .pollInterval(200, TimeUnit.MILLISECONDS)
                .atMost(30, TimeUnit.SECONDS)
                .ignoreExceptions()
                .untilAsserted(() -> { // wait until true
                    primaryLoadManager.monitor();
                    secondaryLoadManager.monitor();
                    var threshold = admin.topicPolicies()
                            .getCompactionThreshold(ServiceUnitStateChannelImpl.TOPIC, false);
                    AssertJUnit.assertEquals(5 * 1024 * 1024, threshold == null ? 0 : threshold.longValue());
                });
    }