public void start() throws Exception {
        this.producerNameGenerator = new DistributedIdGenerator(pulsar.getCoordinationService(),
                PRODUCER_NAME_GENERATOR_PATH, pulsar.getConfiguration().getClusterName());

        ServiceConfiguration serviceConfig = pulsar.getConfiguration();
        List<BindAddress> bindAddresses = BindAddressValidator.validateBindAddresses(serviceConfig,
                Arrays.asList("pulsar", "pulsar+ssl"));
        String internalListenerName = serviceConfig.getInternalListenerName();

        // create a channel for each bind address
        if (bindAddresses.size() == 0) {
            throw new IllegalArgumentException("At least one broker bind address must be configured");
        }
        for (BindAddress a : bindAddresses) {
            InetSocketAddress addr = new InetSocketAddress(a.getAddress().getHost(), a.getAddress().getPort());
            boolean isTls = "pulsar+ssl".equals(a.getAddress().getScheme());
            PulsarChannelInitializer.PulsarChannelOptions opts = PulsarChannelInitializer.PulsarChannelOptions.builder()
                            .enableTLS(isTls)
                            .listenerName(a.getListenerName()).build();

            ServerBootstrap b = defaultServerBootstrap.clone();
            b.childHandler(
                    pulsarChannelInitFactory.newPulsarChannelInitializer(pulsar, opts));
            try {
                Channel ch = b.bind(addr).sync().channel();
                listenChannels.add(ch);

                // identify the primary channel. Note that the legacy bindings appear first and have no listener.
                if (StringUtils.isBlank(a.getListenerName())
                        || StringUtils.equalsIgnoreCase(a.getListenerName(), internalListenerName)) {
                    if (this.listenChannel == null && !isTls) {
                        this.listenChannel = ch;
                    }
                    if (this.listenChannelTls == null && isTls) {
                        this.listenChannelTls = ch;
                    }
                }

                log.info("Started Pulsar Broker service on {}, TLS: {}, listener: {}",
                        ch.localAddress(),
                        isTls ? SslContext.defaultServerProvider().toString() : "(none)",
                        StringUtils.defaultString(a.getListenerName(), "(none)"));
            } catch (Exception e) {
                throw new IOException("Failed to bind Pulsar broker on " + addr, e);
            }
        }

        // start other housekeeping functions
        this.startStatsUpdater(
                serviceConfig.getStatsUpdateInitialDelayInSecs(),
                serviceConfig.getStatsUpdateFrequencyInSecs());
        this.startInactivityMonitor();
        this.startMessageExpiryMonitor();
        this.startCompactionMonitor();
        this.startConsumedLedgersMonitor();
        this.startBacklogQuotaChecker();
        this.updateBrokerPublisherThrottlingMaxRate();
        this.updateBrokerDispatchThrottlingMaxRate();
        this.startCheckReplicationPolicies();
        this.startDeduplicationSnapshotMonitor();
        this.startClearInvalidateTopicNameCacheTask();
    }
@Test
    public void shouldNotPreventCreatingTopicWhenNonexistingTopicIsCached() throws Exception {
        // run multiple iterations to increase the chance of reproducing a race condition in the topic cache
        for (int i = 0; i < 100; i++) {
            final String topicName = "persistent://prop/ns-abc/topic-caching-test-topic" + i;
            CountDownLatch latch = new CountDownLatch(1);
            Thread getStatsThread = new Thread(() -> {
                try {
                    latch.countDown();
                    // create race condition with a short delay
                    // the bug might not reproduce in all environments, this works at least on i7-10750H CPU
                    Thread.sleep(1);
                    admin.topics().getStats(topicName);
                    fail("The topic should not exist yet.");
                } catch (PulsarAdminException.NotFoundException e) {
                    // expected exception
                } catch (PulsarAdminException | InterruptedException e) {
                    log.error("Exception in {}", Thread.currentThread().getName(), e);
                }
            }, "getStatsThread#" + i);
            getStatsThread.start();
            latch.await();
            @Cleanup
            Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
            assertNotNull(producer);
            getStatsThread.join();
        }
    }