@Override
    public void stop(String why) {
        if (!shutdown.compareAndSet(false, true)) {
            return;
        }
        logger.info(why + ".Stopping Tube server...");
        try {
            TubeBroker.this.webServer.stop();
            logger.info("Tube WebService stopped.......");
            masterService.brokerCloseClientB2M(createMasterCloseRequest(),
                    tubeConfig.getHostName(), false);
            logger.info("Tube Closing to Master.....");
        } catch (Throwable e) {
            logger.warn("CloseBroker throw exception : ", e);
        }
        ThreadUtils.sleep(2000);
        logger.info("Tube Client StoreService stopping.....");
        TubeBroker.this.brokerServiceServer.stop();
        logger.info("Tube Client StoreService stopped.....");
        this.offsetRecordService.close();
        logger.info("Tube offset record service stopped.....");
        TubeBroker.this.storeManager.close();
        logger.info("Tube message store stopped.....");
        TubeBroker.this.offsetManager.close(-1);
        logger.info("Tube offset store stopped.....");
        scheduledExecutorService.shutdownNow();
        if (!shutdownHooked.get()) {
            Runtime.getRuntime().removeShutdownHook(TubeBroker.this.shutdownHook);
        }
        try {
            TubeBroker.this.rpcServiceFactory.destroy();
            TubeBroker.this.clientFactory.shutdown();
        } catch (Throwable e2) {
            logger.error("Stop rpcService failure", e2);
        }
        logger.info("Stop tube server successfully.");
        LogManager.shutdown();
        ThreadUtils.sleep(2000);
    }
Start broker service.

@throws Exception  the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/TubeBroker.java