[Pulsar] 一个消息的生命历程（二）——Batch 和消息重复处理

@Overridepublic boolean haveEnoughSpace(MessageImpl<?> msg) {    int messageSize = msg.getDataBuffer().readableBytes();    return (        (maxBytesInBatch <= 0 && (messageSize + currentBatchSizeBytes) <= ClientCnx.getMaxMessageSize())        || (maxBytesInBatch > 0 && (messageSize + currentBatchSizeBytes) <= maxBytesInBatch)    ) && (maxNumMessagesInBatch <= 0 || numMessagesInBatch < maxNumMessagesInBatch);}

    if (canAddToBatch(msg) && totalChunks <= 1) {        if (canAddToCurrentBatch(msg)) {            // should trigger complete the batch message, new message will add to a new batch and new batch            // sequence id use the new message, so that broker can handle the message duplication            if (sequenceId <= lastSequenceIdPushed) {                isLastSequenceIdPotentialDuplicated = true;                if (sequenceId <= lastSequenceIdPublished) {                    log.warn("Message with sequence id {} is definitely a duplicate", sequenceId);                } else {                    log.info("Message with sequence id {} might be a duplicate but cannot be determined at this time.",                        sequenceId);                }                doBatchSendAndAdd(msg, callback, payload);            } else {                // Should flush the last potential duplicated since can't combine potential duplicated messages                // and non-duplicated messages into a batch.                if (isLastSequenceIdPotentialDuplicated) {                    doBatchSendAndAdd(msg, callback, payload);                } else {                    // handle boundary cases where message being added would exceed                    // batch size and/or max message size                    boolean isBatchFull = batchMessageContainer.add(msg, callback);                    lastSendFuture = callback.getFuture();                    payload.release();                    if (isBatchFull) {                        batchMessageAndSend();                    }                }                isLastSequenceIdPotentialDuplicated = false;            }        } else {            doBatchSendAndAdd(msg, callback, payload);        }    }