@Override
    public void execute() {
        boolean debugMode = ExtensibleLoadManagerImpl.debug(conf, log);
        if (debugMode) {
            log.info("Load balancer enabled: {}, Split enabled: {}.",
                    conf.isLoadBalancerEnabled(), conf.isLoadBalancerAutoBundleSplitEnabled());
        }

        if (!isLoadBalancerAutoBundleSplitEnabled()) {
            if (debugMode) {
                log.info("The load balancer or load balancer split already disabled. Skipping.");
            }
            return;
        }

        synchronized (bundleSplitStrategy) {
            final Set<SplitDecision> decisions = bundleSplitStrategy.findBundlesToSplit(context, pulsar);
            if (debugMode) {
                log.info("Split Decisions:", decisions);
            }
            if (!decisions.isEmpty()) {
                // currently following the unloading timeout
                var asyncOpTimeoutMs = conf.getNamespaceBundleUnloadingTimeoutMs();
                List<CompletableFuture<Void>> futures = new ArrayList<>();
                for (SplitDecision decision : decisions) {
                    if (decision.getLabel() == Success) {
                        var split = decision.getSplit();
                        futures.add(
                                splitManager.waitAsync(
                                        serviceUnitStateChannel.publishSplitEventAsync(split),
                                        split.serviceUnit(),
                                        decision,
                                        asyncOpTimeoutMs, TimeUnit.MILLISECONDS)
                        );
                    }
                }
                try {
                    FutureUtil.waitForAll(futures)
                            .get(asyncOpTimeoutMs, TimeUnit.MILLISECONDS);
                } catch (Throwable e) {
                    log.error("Failed to wait for split events to persist.", e);
                }
            } else {
                if (debugMode) {
                    log.info("BundleSplitStrategy returned no bundles to split.");
                }
            }
        }

        if (counter.updatedAt() > counterLastUpdatedAt) {
            splitMetrics.set(counter.toMetrics(pulsar.getAdvertisedAddress()));
            counterLastUpdatedAt = counter.updatedAt();
        }
    }
@Test(timeOut = 30 * 1000)
    public void testDisableLoadBalancer() {

        config.setLoadBalancerEnabled(false);
        SplitScheduler scheduler = new SplitScheduler(pulsar, channel, null, null, null, context, strategy);

        scheduler.execute();

        verify(strategy, times(0)).findBundlesToSplit(any(), any());

        config.setLoadBalancerEnabled(true);
        config.setLoadBalancerAutoBundleSplitEnabled(false);
        scheduler.execute();

        verify(strategy, times(0)).findBundlesToSplit(any(), any());
    }