[Pulsar] 消息从 Producer 到 Broker 的历程

@OverrideCompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message, Transaction txn) {    int partition = routerPolicy.choosePartition(message, topicMetadata);    checkArgument(partition >= 0 && partition < topicMetadata.numPartitions(),            "Illegal partition index chosen by the message routing policy: " + partition);
    ... ...
    return producers.get(partition).internalSendWithTxnAsync(message, txn);}

    // Batch will be compressed when closed    // If a message has a delayed delivery time, we'll always send it individually    if (!isBatchMessagingEnabled() || msgMetadata.hasDeliverAtTime()) {        compressedPayload = applyCompression(payload);        compressed = true;    }

// send in chunksint totalChunks = canAddToBatch(msg) ? 1        : Math.max(1, compressedPayload.readableBytes()) / ClientCnx.getMaxMessageSize()                + (Math.max(1, compressedPayload.readableBytes()) % ClientCnx.getMaxMessageSize() == 0 ? 0 : 1);

    try {        synchronized (this) {            int readStartIndex = 0;            long sequenceId;            if (!msgMetadata.hasSequenceId()) {                sequenceId = msgIdGeneratorUpdater.getAndIncrement(this);                msgMetadata.setSequenceId(sequenceId);            } else {                sequenceId = msgMetadata.getSequenceId();            }            String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null;            for (int chunkId = 0; chunkId < totalChunks; chunkId++) {                serializeAndSendMessage(msg, payload, sequenceId, uuid, chunkId, totalChunks,                        readStartIndex, ClientCnx.getMaxMessageSize(), compressedPayload, compressed,                        compressedPayload.readableBytes(), uncompressedSize, callback);                readStartIndex = ((chunkId + 1) * ClientCnx.getMaxMessageSize());            }        }    } catch (PulsarClientException e) {        e.setSequenceId(msg.getSequenceId());        completeCallbackAndReleaseSemaphore(uncompressedSize, callback, e);    } catch (Throwable t) {        completeCallbackAndReleaseSemaphore(uncompressedSize, callback, new PulsarClientException(t, msg.getSequenceId()));    }

@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);    }} 

ByteBufPair cmd = sendMessage(producerId, sequenceId, numMessages, msgMetadata, encryptedPayload);op = OpSendMsg.create(msg, cmd, sequenceId, callback);op.setNumMessagesInBatch(numMessages);op.setBatchSizeByte(encryptedPayload.readableBytes());if (totalChunks > 1) {    op.totalChunks = totalChunks;    op.chunkId = chunkId;}

message CommandSend {    required uint64 producer_id = 1;    required uint64 sequence_id = 2;    optional int32 num_messages = 3 [default = 1];    optional uint64 txnid_least_bits = 4 [default = 0];    optional uint64 txnid_most_bits = 5 [default = 0];
    /// Add highest sequence id to support batch message with external sequence id    optional uint64 highest_sequence_id = 6 [default = 0];    optional bool is_chunk     =7 [default = false];
    // Specify if the message being published is a Pulsar marker or not    optional bool marker = 8 [default = false];}

pendingMessages.add(op);ClientCnx cnx = cnx();// If we do have a connection, the message is sent immediately, otherwise we'll try again once a new// connection is establishedop.cmd.retain();cnx.ctx().channel().eventLoop().execute(WriteInEventLoopCallback.create(this, cnx, op));

cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());

if (producer.isNonPersistentTopic()) {  // avoid processing non-persist message if reached max concurrent-message limit  if (nonPersistentPendingMessages > maxNonPersistentPendingMessages) {    final long producerId = send.getProducerId();    final long sequenceId = send.getSequenceId();    final long highestSequenceId = send.getHighestSequenceId();    service.getTopicOrderedExecutor().executeOrdered(producer.getTopic().getName(), SafeRun.safeRun(() -> {      commandSender.sendSendReceiptResponse(producerId, sequenceId, highestSequenceId, -1, -1);    }));    producer.recordMessageDrop(send.getNumMessages());    return;  } else {    nonPersistentPendingMessages++;  }}

Long lastSequenceIdPushed = highestSequencedPushed.get(producerName);if (lastSequenceIdPushed != null && sequenceId <= lastSequenceIdPushed) {  if (log.isDebugEnabled()) {    log.debug("[{}] Message identified as duplicated producer={} seq-id={} -- highest-seq-id={}",              topic.getName(), producerName, sequenceId, lastSequenceIdPushed);  }
  // Also need to check sequence ids that has been persisted.  // If current message's seq id is smaller or equals to the  // lastSequenceIdPersisted than its definitely a dup  // If current message's seq id is between lastSequenceIdPersisted and  // lastSequenceIdPushed, then we cannot be sure whether the message is a dup or not  // we should return an error to the producer for the latter case so that it can retry at a future time  Long lastSequenceIdPersisted = highestSequencedPersisted.get(producerName);  if (lastSequenceIdPersisted != null && sequenceId <= lastSequenceIdPersisted) {    return MessageDupStatus.Dup;  } else {    return MessageDupStatus.Unknown;  }}highestSequencedPushed.put(producerName, highestSequenceId);

private boolean currentLedgerIsFull() {    boolean spaceQuotaReached = (currentLedgerEntries >= config.getMaxEntriesPerLedger()            || currentLedgerSize >= (config.getMaxSizePerLedgerMb() * MegaByte));
    long timeSinceLedgerCreationMs = clock.millis() - lastLedgerCreatedTimestamp;    boolean maxLedgerTimeReached = timeSinceLedgerCreationMs >= config.getMaximumRolloverTimeMs();
    if (spaceQuotaReached || maxLedgerTimeReached) {        if (config.getMinimumRolloverTimeMs() > 0) {
            boolean switchLedger = timeSinceLedgerCreationMs > config.getMinimumRolloverTimeMs();            return switchLedger;        } else {            return true;        }    } else {        return false;    }}

@Overridepublic void addComplete(Position pos, ByteBuf entryData, Object ctx) {    PublishContext publishContext = (PublishContext) ctx;    PositionImpl position = (PositionImpl) pos;
    // Message has been successfully persisted    messageDeduplication.recordMessagePersisted(publishContext, position);
    if (!publishContext.isMarkerMessage()) {        lastDataMessagePublishedTimestamp = Clock.systemUTC().millis();    }
    // in order to sync the max position when cursor read entries    transactionBuffer.syncMaxReadPositionForNormalPublish((PositionImpl) ledger.getLastConfirmedEntry());    publishContext.setMetadataFromEntryData(entryData);    publishContext.completed(null, position.getLedgerId(), position.getEntryId());    decrementPendingWriteOpsAndCheck();}