@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 testExecuteFailure() {
        AtomicReference<List<Metrics>> reference = new AtomicReference();
        SplitCounter counter = new SplitCounter();
        SplitManager manager = new SplitManager(counter);
        SplitScheduler scheduler = new SplitScheduler(pulsar, channel, manager, counter, reference, context, strategy);
        doReturn(CompletableFuture.failedFuture(new RuntimeException())).when(channel).publishSplitEventAsync(any());

        scheduler.execute();


        var counterExpected = new SplitCounter();
        counterExpected.update(SplitDecision.Label.Failure, SplitDecision.Reason.Unknown);
        counterExpected.update(SplitDecision.Label.Failure, SplitDecision.Reason.Unknown);
        verify(channel, times(1)).publishSplitEventAsync(eq(decision1.getSplit()));
        verify(channel, times(1)).publishSplitEventAsync(eq(decision2.getSplit()));

        assertEquals(reference.get().toString(), counterExpected.toMetrics(pulsar.getAdvertisedAddress()).toString());
    }