[Pulsar] 消息生命历程（四）——Broker 的消息处理

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