@Override
    public synchronized void execute() {
        boolean debugMode = conf.isLoadBalancerDebugModeEnabled() || log.isDebugEnabled();
        if (debugMode) {
            log.info("Load balancer enabled: {}, Shedding enabled: {}.",
                    conf.isLoadBalancerEnabled(), conf.isLoadBalancerSheddingEnabled());
        }
        if (!isLoadBalancerSheddingEnabled()) {
            if (debugMode) {
                log.info("The load balancer or load balancer shedding already disabled. Skipping.");
            }
            return;
        }
        // Remove bundles who have been unloaded for longer than the grace period from the recently unloaded map.
        final long timeout = System.currentTimeMillis()
                - TimeUnit.MINUTES.toMillis(conf.getLoadBalancerSheddingGracePeriodMinutes());
        recentlyUnloadedBundles.keySet().removeIf(e -> recentlyUnloadedBundles.get(e) < timeout);

        long asyncOpTimeoutMs = conf.getNamespaceBundleUnloadingTimeoutMs();
        synchronized (namespaceUnloadStrategy) {
            try {
                Boolean isChannelOwner = channel.isChannelOwnerAsync().get(asyncOpTimeoutMs, TimeUnit.MILLISECONDS);
                if (!isChannelOwner) {
                    if (debugMode) {
                        log.info("Current broker is not channel owner. Skipping.");
                    }
                    return;
                }
                List<String> availableBrokers = context.brokerRegistry().getAvailableBrokersAsync()
                        .get(asyncOpTimeoutMs, TimeUnit.MILLISECONDS);
                if (debugMode) {
                    log.info("Available brokers: {}", availableBrokers);
                }
                if (availableBrokers.size() <= 1) {
                    log.info("Only 1 broker available: no load shedding will be performed. Skipping.");
                    return;
                }
                final Set<UnloadDecision> decisions = namespaceUnloadStrategy
                        .findBundlesForUnloading(context, recentlyUnloadedBundles, recentlyUnloadedBrokers);
                if (debugMode) {
                    log.info("[{}] Unload decision result: {}",
                            namespaceUnloadStrategy.getClass().getSimpleName(), decisions);
                }
                if (decisions.isEmpty()) {
                    if (debugMode) {
                        log.info("[{}] Unload decision unloads is empty. Skipping.",
                                namespaceUnloadStrategy.getClass().getSimpleName());
                    }
                    return;
                }
                List<CompletableFuture<Void>> futures = new ArrayList<>();
                unloadBrokers.clear();
                decisions.forEach(decision -> {
                    if (decision.getLabel() == Success) {
                        Unload unload = decision.getUnload();
                        log.info("[{}] Unloading bundle: {}",
                                namespaceUnloadStrategy.getClass().getSimpleName(), unload);
                        futures.add(unloadManager.waitAsync(channel.publishUnloadEventAsync(unload),
                                        unload.serviceUnit(), decision, asyncOpTimeoutMs, TimeUnit.MILLISECONDS)
                                .thenAccept(__ -> {
                                    unloadBrokers.add(unload.sourceBroker());
                                    recentlyUnloadedBundles.put(unload.serviceUnit(), System.currentTimeMillis());
                                    recentlyUnloadedBrokers.put(unload.sourceBroker(), System.currentTimeMillis());
                                }));
                    }
                });
                FutureUtil.waitForAll(futures)
                        .whenComplete((__, ex) -> counter.updateUnloadBrokerCount(unloadBrokers.size()))
                        .get(asyncOpTimeoutMs, TimeUnit.MILLISECONDS);
            } catch (Exception ex) {
                log.error("[{}] Namespace unload has exception.",
                        namespaceUnloadStrategy.getClass().getSimpleName(), ex);
            } finally {
                if (counter.updatedAt() > counterLastUpdatedAt) {
                    unloadMetrics.set(counter.toMetrics(pulsar.getAdvertisedAddress()));
                    counterLastUpdatedAt = counter.updatedAt();
                }
            }
        }
    }
@Test(timeOut = 30 * 1000)
    public void testExecuteMoreThenOnceWhenFirstNotDone() throws InterruptedException {
        AtomicReference<List<Metrics>> reference = new AtomicReference<>();
        UnloadCounter counter = new UnloadCounter();
        LoadManagerContext context = setupContext();
        BrokerRegistry registry = context.brokerRegistry();
        ServiceUnitStateChannel channel = mock(ServiceUnitStateChannel.class);
        UnloadManager unloadManager = mock(UnloadManager.class);
        PulsarService pulsar = mock(PulsarService.class);
        NamespaceUnloadStrategy unloadStrategy = mock(NamespaceUnloadStrategy.class);
        doReturn(CompletableFuture.completedFuture(true)).when(channel).isChannelOwnerAsync();
        @Cleanup("shutdownNow")
        ExecutorService executor = Executors.newFixedThreadPool(1);
        doAnswer(__ -> CompletableFuture.supplyAsync(() -> {
                try {
                    // Delay 5 seconds to finish.
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                return Lists.newArrayList("broker-1", "broker-2");
            }, executor)).when(registry).getAvailableBrokersAsync();
        UnloadScheduler scheduler = new UnloadScheduler(pulsar, loadManagerExecutor, unloadManager, context,
                channel, unloadStrategy, counter, reference);
        @Cleanup("shutdownNow")
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        CountDownLatch latch = new CountDownLatch(5);
        for (int i = 0; i < 5; i++) {
            executorService.execute(() -> {
                scheduler.execute();
                latch.countDown();
            });
        }
        latch.await();

        verify(registry, times(5)).getAvailableBrokersAsync();
    }