@Override
    public boolean confirmConsume(String confirmContext, boolean isConsumed,
            ConfirmResult result) throws TubeClientException {
        if (result == null) {
            throw new TubeClientException("Illegal parameter: parameter result is null!");
        }
        StringBuilder sBuilder = new StringBuilder(512);
        if (isShutdown()) {
            result.setFailResult(TErrCodeConstants.CLIENT_SHUTDOWN,
                    "The client is not started or closed!");
            return result.isSuccess();
        }
        long currOffset = TBaseConstants.META_VALUE_UNDEFINED;
        long maxOffset = TBaseConstants.META_VALUE_UNDEFINED;
        // Verify if the confirmContext is valid
        if (TStringUtils.isBlank(confirmContext)) {
            result.setFailResult(TErrCodeConstants.BAD_REQUEST,
                    "ConfirmContext is null!");
            return result.isSuccess();
        }
        String[] strConfirmContextItems =
                confirmContext.split(TokenConstants.ATTR_SEP);
        if (strConfirmContextItems.length != 4) {
            result.setFailResult(TErrCodeConstants.BAD_REQUEST,
                    "ConfirmContext format error: value must be aaaa:bbbb:cccc:ddddd !");
            return result.isSuccess();
        }
        for (String itemStr : strConfirmContextItems) {
            if (TStringUtils.isBlank(itemStr)) {
                result.setFailResult(TErrCodeConstants.BAD_REQUEST,
                        sBuilder.append("ConfirmContext's format error: item (")
                                .append(itemStr).append(") is null !").toString());
                sBuilder.delete(0, sBuilder.length());
                return result.isSuccess();
            }
        }
        String keyId = sBuilder.append(strConfirmContextItems[0].trim())
                .append(TokenConstants.ATTR_SEP).append(strConfirmContextItems[1].trim())
                .append(TokenConstants.ATTR_SEP).append(strConfirmContextItems[2].trim()).toString();
        sBuilder.delete(0, sBuilder.length());
        String topicName = strConfirmContextItems[1].trim();
        long timeStamp = Long.parseLong(strConfirmContextItems[3]);
        long midTime = System.currentTimeMillis();
        // book statistics information
        clientStatsInfo.bookReturnDuration(keyId, midTime - timeStamp);
        if (!clientRmtDataCache.isPartitionInUse(keyId, timeStamp)) {
            result.setFailResult(TErrCodeConstants.BAD_REQUEST,
                    "The confirmContext's value invalid!");
            return result.isSuccess();
        }
        Partition curPartition =
                clientRmtDataCache.getPartitionByKey(keyId);
        if (curPartition == null) {
            result.setFailResult(TErrCodeConstants.NOT_FOUND,
                    sBuilder.append("Not found the partition by confirmContext:")
                            .append(confirmContext).toString());
            sBuilder.delete(0, sBuilder.length());
            return result.isSuccess();
        }
        if (this.consumerConfig.isPullConfirmInLocal()) {
            clientRmtDataCache.succRspRelease(keyId, topicName,
                    timeStamp, isConsumed, isFilterConsume(topicName), currOffset, maxOffset);
            result.setSuccResult(topicName, curPartition, currOffset, maxOffset);
            return result.isSuccess();
        } else {
            try {
                ClientBroker.CommitOffsetResponseB2C response =
                        getBrokerService(curPartition.getBroker())
                                .consumerCommitC2B(createBrokerCommitRequest(curPartition, isConsumed),
                                        AddressUtils.getLocalAddress(), getConsumerConfig().isTlsEnable());
                if (response == null) {
                    result.setFailResult(TErrCodeConstants.CONNECT_RETURN_NULL,
                            sBuilder.append("Confirm ").append(confirmContext)
                                    .append("'s offset failed, response is null!").toString());
                    sBuilder.delete(0, sBuilder.length());
                    return result.isSuccess();
                } else {
                    if (response.hasCurrOffset() && response.getCurrOffset() >= 0) {
                        currOffset = response.getCurrOffset();
                    }
                    if (response.hasMaxOffset() && response.getMaxOffset() >= 0) {
                        maxOffset = response.getMaxOffset();
                    }
                    result.setProcessResult(response.getSuccess(),
                            response.getErrCode(), response.getErrMsg(),
                            topicName, curPartition, currOffset, maxOffset);
                    return result.isSuccess();
                }
            } catch (Throwable e) {
                sBuilder.delete(0, sBuilder.length());
                result.setFailResult(TErrCodeConstants.BAD_REQUEST,
                        sBuilder.append("Confirm ").append(confirmContext)
                                .append("'s offset failed, exception is ")
                                .append(e.toString()).toString());
                sBuilder.delete(0, sBuilder.length());
                return result.isSuccess();
            } finally {
                clientRmtDataCache.succRspRelease(keyId, topicName, timeStamp,
                        isConsumed, isFilterConsume(topicName), currOffset, maxOffset);
                clientStatsInfo.bookConfirmDuration(keyId,
                        System.currentTimeMillis() - midTime);
            }
        }
    }
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