[Pulsar] 一个消息的生命历程（一）

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