[Pulsar] Broker 消息分发

for (int i = 0; i < entries.size(); i++) {  Entry entry = entries.get(i);  if (entry == null) {    // Entry was filtered out    continue;  }
  int batchSize = batchSizes.getBatchSize(i);
  if (batchSize > 1 && !cnx.isBatchMessageCompatibleVersion()) {    log.warn("[{}-{}] Consumer doesn't support batch messages -  consumerId {}, msg id {}-{}",             topicName, subscription,             consumerId, entry.getLedgerId(), entry.getEntryId());    ctx.close();    entry.release();    continue;  }
  ByteBuf metadataAndPayload = entry.getDataBuffer();  // increment ref-count of data and release at the end of process:  // so, we can get chance to call entry.release  metadataAndPayload.retain();
  int redeliveryCount = 0;  PositionImpl position = PositionImpl.get(entry.getLedgerId(), entry.getEntryId());  if (redeliveryTracker.contains(position)) {    redeliveryCount = redeliveryTracker.incrementAndGetRedeliveryCount(position);  }
  ctx.write(    cnx.newMessageAndIntercept(consumerId, entry.getLedgerId(), entry.getEntryId(), partitionIdx,                               redeliveryCount, metadataAndPayload,                               batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i), topicName),    ctx.voidPromise());  entry.release();}

                // Schedule a new read batch operation only after the previous batch has been written to the socket.                topic.getBrokerService().getTopicOrderedExecutor().executeOrdered(topicName,                    SafeRun.safeRun(() -> {                        synchronized (PersistentDispatcherSingleActiveConsumer.this) {                            Consumer newConsumer = getActiveConsumer();                            if (newConsumer != null && !havePendingRead) {                                readMoreEntries(newConsumer);                            } else {                                log.debug(                                        "[{}-{}] Ignoring write future complete."                                                + " consumerAvailable={} havePendingRead={}",                                        name, newConsumer, newConsumer != null, havePendingRead);                            }                        }                    }));