@Override
    public void shutdown() throws Throwable {
        StringBuilder strBuffer = new StringBuilder(512);
        if (!clientStatus.compareAndSet(2, 3)) {
            switch (clientStatus.get()) {
                case 3: {
                    logger.info(strBuffer.append("[SHUTDOWN_CONSUMER] ")
                            .append(this.consumerId)
                            .append(" is shutting down, do nothing...").toString());
                }
                    break;

                case 0: {
                    logger.info(strBuffer.append("[SHUTDOWN_CONSUMER] ")
                            .append(this.consumerId)
                            .append(" was already shutdown, do nothing...").toString());
                }
                    break;

                case 1:
                default: {
                    logger.info(strBuffer.append("[SHUTDOWN_CONSUMER] ")
                            .append(this.consumerId)
                            .append(" is starting, please wait a minute!").toString());
                }
            }
            return;
        }
        logger.info(strBuffer.append("[SHUTDOWN_CONSUMER] Shutting down consumer:")
                .append(this.consumerId).toString());
        strBuffer.delete(0, strBuffer.length());
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            //
        }
        //
        this.clientRmtDataCache.close();
        Map<BrokerInfo, List<PartitionSelectResult>> unRegisterInfoMap =
                clientRmtDataCache.getAllPartitionListWithStatus();
        unregisterPartitions(unRegisterInfoMap);
        this.sessionFactory.removeClient(this);
        if (this.heartService2Master != null) {
            try {
                this.heartService2Master.shutdownNow();
            } catch (Throwable ee) {
                //
            }
        }
        if (this.heartBeatThread2Broker != null) {
            try {
                this.heartBeatThread2Broker.interrupt();
                heartBeatThread2Broker.join();
                this.heartBeatThread2Broker = null;
            } catch (Throwable ee) {
                //
            }
        }
        // print metric information
        clientStatsInfo.selfPrintStatsInfo(true, true, strBuffer);
        logger.info(strBuffer
                .append("[SHUTDOWN_CONSUMER] Partitions unregistered,  consumer :")
                .append(this.consumerId).toString());
        strBuffer.delete(0, strBuffer.length());
        try {
            masterService.consumerCloseClientC2M(createMasterCloseRequest(),
                    AddressUtils.getLocalAddress(), consumerConfig.isTlsEnable());
        } catch (Throwable e) {
            strBuffer.delete(0, strBuffer.length());
            logger.warn(strBuffer
                    .append("[SHUTDOWN_CONSUMER] call closeRequest failure, error is ")
                    .append(e.getMessage()).toString());
            strBuffer.delete(0, strBuffer.length());
        }
        logger.info(strBuffer.append("[SHUTDOWN_CONSUMER] Client closed, consumer : ")
                .append(this.consumerId).toString());
        this.clientStatus.set(0);
    }
Initial a client-balance consumer object
@param messageSessionFactory   the session factory
@param consumerConfig          the consumer configure
@throws TubeClientException    the exception while creating object.
inlong-tubemq/tubemq-client/src/main/java/org/apache/inlong/tubemq/client/consumer/SimpleClientBalanceConsumer.java