@Override
    public void start() throws PulsarServerException {
        try {
            // At this point, the ports will be updated with the real port number that the server was assigned
            Map<String, String> protocolData = pulsar.getProtocolDataToAdvertise();

            lastData = new LocalBrokerData(pulsar.getWebServiceAddress(), pulsar.getWebServiceAddressTls(),
                    pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls(), pulsar.getAdvertisedListeners());
            lastData.setProtocols(protocolData);
            // configure broker-topic mode
            lastData.setPersistentTopicsEnabled(pulsar.getConfiguration().isEnablePersistentTopics());
            lastData.setNonPersistentTopicsEnabled(pulsar.getConfiguration().isEnableNonPersistentTopics());

            localData = new LocalBrokerData(pulsar.getWebServiceAddress(), pulsar.getWebServiceAddressTls(),
                    pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls(), pulsar.getAdvertisedListeners());
            localData.setProtocols(protocolData);
            localData.setBrokerVersionString(pulsar.getBrokerVersion());
            // configure broker-topic mode
            localData.setPersistentTopicsEnabled(pulsar.getConfiguration().isEnablePersistentTopics());
            localData.setNonPersistentTopicsEnabled(pulsar.getConfiguration().isEnableNonPersistentTopics());
            localData.setLoadManagerClassName(conf.getLoadManagerClassName());

            String brokerId = pulsar.getBrokerId();
            brokerZnodePath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + brokerId;
            updateLocalBrokerData();

            brokerDataLock = brokersData.acquireLock(brokerZnodePath, localData).join();
            pulsarResources.getLoadBalanceResources()
                    .getBrokerTimeAverageDataResources()
                    .updateTimeAverageBrokerData(brokerId, new TimeAverageBrokerData())
                    .join();
            updateAll();
        } catch (Exception e) {
            log.error("Unable to acquire lock for broker: [{}]", brokerZnodePath, e);
            throw new PulsarServerException(e);
        }
    }
@Test
    public void testBrokerAffinity() throws Exception {
        // Start broker 3
        pulsar3.start();

        final String tenant = "test";
        final String cluster = "test";
        String namespace = tenant + "/" + cluster + "/" + "test";
        String topic = "persistent://" + namespace + "/my-topic1";
        admin1.clusters().createCluster(cluster, ClusterData.builder().serviceUrl(pulsar1.getWebServiceAddress()).build());
        admin1.tenants().createTenant(tenant,
                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet(cluster)));
        admin1.namespaces().createNamespace(namespace, 16);

        String topicLookup = admin1.lookups().lookupTopic(topic);
        String bundleRange = admin1.lookups().getBundleRange(topic);

        String brokerServiceUrl = pulsar1.getBrokerServiceUrl();
        String brokerId = pulsar1.getBrokerId();
        log.debug("initial broker service url - {}", topicLookup);
        Random rand=new Random();

        if (topicLookup.equals(brokerServiceUrl)) {
            int x = rand.nextInt(2);
            if (x == 0) {
                brokerId = pulsar2.getBrokerId();
                brokerServiceUrl = pulsar2.getBrokerServiceUrl();
            }
            else {
                brokerId = pulsar3.getBrokerId();
                brokerServiceUrl = pulsar3.getBrokerServiceUrl();
            }
        }
        log.debug("destination broker service url - {}, broker url - {}", brokerServiceUrl, brokerId);
        String leaderBrokerId = admin1.brokers().getLeaderBroker().getBrokerId();
        log.debug("leader lookup address - {}, broker1 lookup address - {}", leaderBrokerId,
                pulsar1.getBrokerId());
        // Make a call to broker which is not a leader
        if (!leaderBrokerId.equals(pulsar1.getBrokerId())) {
            admin1.namespaces().unloadNamespaceBundle(namespace, bundleRange, brokerId);
        }
        else {
            admin2.namespaces().unloadNamespaceBundle(namespace, bundleRange, brokerId);
        }

        sleep(2000);
        String topicLookupAfterUnload = admin1.lookups().lookupTopic(topic);
        log.debug("final broker service url - {}", topicLookupAfterUnload);
        Assert.assertEquals(brokerServiceUrl, topicLookupAfterUnload);
    }