@Override
    public boolean getMessage(ConsumeResult result) throws TubeClientException {
        if (result == null) {
            throw new TubeClientException("Illegal parameter: parameter result is null!");
        }
        if (isShutdown()) {
            result.setFailResult(TErrCodeConstants.CLIENT_SHUTDOWN,
                    "The client is not started or closed!");
            return result.isSuccess();
        }
        PartitionSelectResult selectResult = null;
        long startTime = System.currentTimeMillis();
        while (true) {
            if (isShutdown()) {
                result.setFailResult(TErrCodeConstants.CLIENT_SHUTDOWN,
                        "The client has been shutdown!");
                return result.isSuccess();
            }
            selectResult = clientRmtDataCache.getCurrPartsStatus();
            if (selectResult.isSuccess()) {
                break;
            }
            if ((consumerConfig.getPullConsumeReadyWaitPeriodMs() >= 0L)
                    && ((System.currentTimeMillis() - startTime) >= consumerConfig.getPullConsumeReadyWaitPeriodMs())) {
                result.setFailResult(selectResult.getErrCode(), selectResult.getErrMsg());
                return result.isSuccess();
            }
            if (consumerConfig.getPullConsumeReadyChkSliceMs() > 0L) {
                ThreadUtils.sleep(consumerConfig.getPullConsumeReadyChkSliceMs());
            }
        }
        StringBuilder sBuilder = new StringBuilder(512);
        // Check the data cache first
        selectResult = clientRmtDataCache.pullSelect();
        if (!selectResult.isSuccess()) {
            result.setFailResult(selectResult.getErrCode(), selectResult.getErrMsg());
            return result.isSuccess();
        }
        FetchContext taskContext = fetchMessage(selectResult, sBuilder);
        result.setProcessResult(taskContext);
        return result.isSuccess();
    }
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