[root@10 keep_rule_sender]# ps -ef|grep java
root      4896 18614  0 15:52 pts/0    00:00:00 grep --color=auto java
root     15550 15547  0 5月25 ?       00:24:08 ./jre/bin/java -jar -Dfile.encoding=UTF-8 -XX:+UseG1GC -Xms16384M -Xmx16384M -Xmn8192M -Xss2M AuthCenter.jar

[root@10 keep_rule_sender]# top Hp 15550
top - 15:55:19 up 115 days, 54 min,  1 user,  load average: 0.31, 0.36, 0.40
Threads: 146 total,   0 running, 146 sleeping,   0 stopped,   0 zombie
%Cpu(s):  3.2 us,  0.7 sy,  0.0 ni, 96.1 id,  0.0 wa,  0.0 hi,  0.1 si,  0.0 st
KiB Mem : 13157919+total, 62099732 free, 26188756 used, 43290704 buff/cache
KiB Swap:  4194300 total,  4194300 free,        0 used. 10426378+avail Mem
  PID USER      PR  NI    VIRT    RES    SHR S %CPU %MEM     TIME+ COMMAND                                                                                                                                      
15937 root      20   0 27.453g 9.252g  13756 S  0.3  7.4   0:16.47 java                                                                                                                                         
15947 root      20   0 27.453g 9.252g  13756 S  0.3  7.4   0:16.73 java

"nioEventLoopGroup-3-52" #78 prio=10 os_prio=0 tid=0x00007fe8b012c000 nid=0x3e41 runnable [0x00007fe7eb7fa000]
java.lang.Thread.State: RUNNABLE
  at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
  at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
  at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:79)
  at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
  - locked <0x00000003c0056660> (a io.netty.channel.nio.SelectedSelectionKeySet)
  - locked <0x00000003c0056608> (a java.util.Collections$UnmodifiableSet)
  - locked <0x00000003c0056618> (a sun.nio.ch.EPollSelectorImpl)
  at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
  at io.netty.channel.nio.NioEventLoop.select(NioEventLoop.java:639)
  at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:325)
  at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
  at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
  at java.lang.Thread.run(Thread.java:745)

private void startNewSnapshot() {
  cleanupTimedOutSnapshots();
  // 判断是否有新消息写入了
  if (topic.getLastDataMessagePublishedTimestamp() < lastCompletedSnapshotStartTime) {
    // There was no message written since the last snapshot, we can skip creating a new snapshot
    if (log.isDebugEnabled()) {
      log.debug("[{}] There is no new data in topic. Skipping snapshot creation.", topic.getName());
    }
    return;
  }

  MutableBoolean anyReplicatorDisconnected = new MutableBoolean();
  topic.getReplicators().forEach((cluster, replicator) -> {
    if (!replicator.isConnected()) {
      anyReplicatorDisconnected.setTrue();
    }
  });

    // 当所有集群都正常时才会发送通知。
  if (anyReplicatorDisconnected.isTrue()) {
    // Do not attempt to create snapshot when some of the clusters are not reachable
    if (log.isDebugEnabled()) {
      log.debug("[{}] Do not attempt to create snapshot when some of the clusters are not reachable.",
          topic.getName());
    }
    return;
  }

  if (log.isDebugEnabled()) {
    log.debug("[{}] Starting snapshot creation.", topic.getName());
  }

  pendingSnapshotsMetric.inc();
  ReplicatedSubscriptionsSnapshotBuilder builder = new ReplicatedSubscriptionsSnapshotBuilder(this,
      topic.getReplicators().keys(), topic.getBrokerService().pulsar().getConfiguration(), Clock.systemUTC());
  pendingSnapshots.put(builder.getSnapshotId(), builder);
    // 这里只是通过往该Topic写入一条消息，消息中包含通知ID。
  builder.start();
}

void start() {
  if (log.isDebugEnabled()) {
    log.debug("[{}] Starting new snapshot {} - Clusters: {}", controller.topic().getName(), snapshotId,
        missingClusters);
  }
  startTimeMillis = clock.millis();
    // 设置通知ID（快照ID）和本集群名称。
  controller.writeMarker(
      Markers.newReplicatedSubscriptionsSnapshotRequest(snapshotId, controller.localCluster()));
}

private void checkReplicatedSubscriptionMarker(Position position, MessageImpl<?> msg, ByteBuf payload) {
  if (!msg.getMessageBuilder().hasMarkerType()) {
    // No marker is defined
    return;
  }

  int markerType = msg.getMessageBuilder().getMarkerType();

    // 这里确保只处理消息发送方集群发送的消息，即只处理集群A发送的消息。
  if (!(msg.getMessageBuilder().hasReplicatedFrom()
      && remoteCluster.equals(msg.getMessageBuilder().getReplicatedFrom()))) {
    // Only consider markers that are coming from the same cluster that this
    // replicator instance is assigned to.
    // All the replicators will see all the markers, but we need to only process
    // it once.
    return;
  }

    // 需要处理的消息包含三类：通知、通知应答、消费位置更新。
  switch (markerType) {
  case MarkerType.REPLICATED_SUBSCRIPTION_SNAPSHOT_REQUEST_VALUE:
  case MarkerType.REPLICATED_SUBSCRIPTION_SNAPSHOT_RESPONSE_VALUE:
  case MarkerType.REPLICATED_SUBSCRIPTION_UPDATE_VALUE:
    topic.receivedReplicatedSubscriptionMarker(position, markerType, payload);
    break;

  default:
    // Do nothing
  }
}

private void receivedSnapshotRequest(ReplicatedSubscriptionsSnapshotRequest request) {
  // if replicator producer is already closed, restart it to send snapshot response
  Replicator replicator = topic.getReplicators().get(request.getSourceCluster());
  if (!replicator.isConnected()) {
    topic.startReplProducers();
  }

  // Send response containing the current last written message id. The response
  // marker we're publishing locally and then replicating will have a higher
  // message id.
    // 得到集群上次写入的消息
  PositionImpl lastMsgId = (PositionImpl) topic.getLastPosition();
  if (log.isDebugEnabled()) {
    log.debug("[{}] Received snapshot request. Last msg id: {}", topic.getName(), lastMsgId);
  }

    // 把消息ID连同通知ID一起写入到Topic中。
  ByteBuf marker = Markers.newReplicatedSubscriptionsSnapshotResponse(
      request.getSnapshotId(),
      request.getSourceCluster(),
      localCluster,
      lastMsgId.getLedgerId(), lastMsgId.getEntryId());
  writeMarker(marker);
}

private void receivedSnapshotResponse(Position position, ReplicatedSubscriptionsSnapshotResponse response) {
  String snapshotId = response.getSnapshotId();
  ReplicatedSubscriptionsSnapshotBuilder builder = pendingSnapshots.get(snapshotId);
  if (builder == null) {
    if (log.isDebugEnabled()) {
      log.debug("[{}] Received late reply for timed-out snapshot {} from {}", topic.getName(), snapshotId,
          response.getCluster().getCluster());
    }
    return;
  }

  builder.receivedSnapshotResponse(position, response);
}

// 该Position是集群A和B返回的消息在集群A上的Position。
synchronized void receivedSnapshotResponse(Position position, ReplicatedSubscriptionsSnapshotResponse response) {
  if (log.isDebugEnabled()) {
    log.debug("[{}] Received response from {}", controller.topic().getName(),
        response.getCluster().getCluster());
  }
  String cluster = response.getCluster().getCluster();
    // Response中包含集群A/集群B中的LAC消息。
  responses.putIfAbsent(cluster, new MarkersMessageIdData().copyFrom(response.getCluster().getMessageId()));
  missingClusters.remove(cluster);

  if (log.isDebugEnabled()) {
    log.debug("[{}] Missing clusters {}", controller.topic().getName(), missingClusters);
  }

  if (!missingClusters.isEmpty()) {
    // We're still waiting for more responses to come back
    return;
  }

  // We have now received all responses
  // 如果超过两个集群需要两次通知，这里没看懂为啥...
  if (needTwoRounds && !firstRoundComplete) {
    // Mark that 1st round is done and start a 2nd round
    firstRoundComplete = true;
    missingClusters.addAll(remoteClusters);

    controller.writeMarker(
        Markers.newReplicatedSubscriptionsSnapshotRequest(snapshotId, controller.localCluster()));
    return;
  }

  if (log.isDebugEnabled()) {
    log.debug("[{}] Snapshot is complete {}", controller.topic().getName(), snapshotId);
  }
  // Snapshot is now complete, store it in the local topic
    // 这里是建立记录集群B/集群C的LAC消息和集群A收到集群B/集群C是保存LAC的消息的消息ID间的关联关系，而且是最后边一个消息。
    // 所以这里不能实现同一条消息在三个集群间的关联。
  PositionImpl p = (PositionImpl) position;
  controller.writeMarker(
      Markers.newReplicatedSubscriptionsSnapshot(snapshotId, controller.localCluster(),
          p.getLedgerId(), p.getEntryId(), responses));
  controller.snapshotCompleted(snapshotId);

  double latencyMillis = clock.millis() - startTimeMillis;
  snapshotMetric.observe(latencyMillis);
}

public int filterEntriesForConsumer(Optional<EntryWrapper[]> entryWrapper, int entryWrapperOffset,
     List<Entry> entries, EntryBatchSizes batchSizes, SendMessageInfo sendMessageInfo,
     EntryBatchIndexesAcks indexesAcks, ManagedCursor cursor, boolean isReplayRead) {
  int totalMessages = 0;
  long totalBytes = 0;
  int totalChunkedMessages = 0;
  int totalEntries = 0;
  for (int i = 0, entriesSize = entries.size(); i < entriesSize; i++) {
    Entry entry = entries.get(i);
    if (entry == null) {
      continue;
    }
    totalEntries++;
    ByteBuf metadataAndPayload = entry.getDataBuffer();
    int entryWrapperIndex = i + entryWrapperOffset;
    MessageMetadata msgMetadata = entryWrapper.isPresent() && entryWrapper.get()[entryWrapperIndex] != null
        ? entryWrapper.get()[entryWrapperIndex].getMetadata()
        : null;
    msgMetadata = msgMetadata == null
        ? Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1)
        : msgMetadata;
    if (!isReplayRead && msgMetadata != null && msgMetadata.hasTxnidMostBits()
        && msgMetadata.hasTxnidLeastBits()) {
      // 省略代码 ...
            
    } else if (msgMetadata == null || Markers.isServerOnlyMarker(msgMetadata)) {
            // 这里处理读到的快照消息。
      PositionImpl pos = (PositionImpl) entry.getPosition();
      // Message metadata was corrupted or the messages was a server-only marker

      if (Markers.isReplicatedSubscriptionSnapshotMarker(msgMetadata)) {
        processReplicatedSubscriptionSnapshot(pos, metadataAndPayload);
      }

      entries.set(i, null);
      entry.release();
      subscription.acknowledgeMessage(Collections.singletonList(pos), AckType.Individual,
          Collections.emptyMap());
      continue;
    } else if (msgMetadata.hasDeliverAtTime()
        && trackDelayedDelivery(entry.getLedgerId(), entry.getEntryId(), msgMetadata)) {
      // The message is marked for delayed delivery. Ignore for now.
      entries.set(i, null);
      entry.release();
      continue;
    }

    // 省略代码 ...
  }
  sendMessageInfo.setTotalMessages(totalMessages);
  sendMessageInfo.setTotalBytes(totalBytes);
  sendMessageInfo.setTotalChunkedMessages(totalChunkedMessages);
  return totalEntries;
}

private void processReplicatedSubscriptionSnapshot(PositionImpl pos, ByteBuf headersAndPayload) {
  // Remove the protobuf headers
  Commands.skipMessageMetadata(headersAndPayload);

  try {
    ReplicatedSubscriptionsSnapshot snapshot = Markers.parseReplicatedSubscriptionsSnapshot(headersAndPayload);
    subscription.processReplicatedSubscriptionSnapshot(snapshot);
  } catch (Throwable t) {
    log.warn("Failed to process replicated subscription snapshot at {} -- {}", pos, t.getMessage(), t);
    return;
  }
}

@Override
public void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
    ReplicatedSubscriptionSnapshotCache snapshotCache = this.replicatedSubscriptionSnapshotCache;
    if (snapshotCache != null) {
        snapshotCache.addNewSnapshot(new ReplicatedSubscriptionsSnapshot().copyFrom(snapshot));
    }
}

public synchronized void addNewSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
    // 这里的本地消息ID，就是保存集群B/集群C LAC消息的消息ID，且是最新的一个。
    MarkersMessageIdData msgId = snapshot.getLocalMessageId();
    PositionImpl position = new PositionImpl(msgId.getLedgerId(), msgId.getEntryId());

    if (log.isDebugEnabled()) {
        log.debug("[{}] Added new replicated-subscription snapshot at {} -- {}", subscription, position,
                  snapshot.getSnapshotId());
    }

    snapshots.put(position, snapshot);

    // Prune the cache
    while (snapshots.size() > maxSnapshotToCache) {
        snapshots.pollFirstEntry();
    }
}

@Override
public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<String, Long> properties) {
  Position previousMarkDeletePosition = cursor.getMarkDeletedPosition();

  // 省略代码 ... 

    // 当可删除位置更新了（注意不能处理单独确认的消息）
  if (!cursor.getMarkDeletedPosition().equals(previousMarkDeletePosition)) {
    this.updateLastMarkDeleteAdvancedTimestamp();

    // Mark delete position advance
    ReplicatedSubscriptionSnapshotCache snapshotCache  = this.replicatedSubscriptionSnapshotCache;
    if (snapshotCache != null) {
      ReplicatedSubscriptionsSnapshot snapshot = snapshotCache
          .advancedMarkDeletePosition((PositionImpl) cursor.getMarkDeletedPosition());
      if (snapshot != null) {
        topic.getReplicatedSubscriptionController()
            .ifPresent(c -> c.localSubscriptionUpdated(subName, snapshot));
      }
    }
  }

  deleteTransactionMarker(properties);

  if (topic.getManagedLedger().isTerminated() && cursor.getNumberOfEntriesInBacklog(false) == 0) {
    // Notify all consumer that the end of topic was reached
    if (dispatcher != null) {
      dispatcher.getConsumers().forEach(Consumer::reachedEndOfTopic);
    }
  }
}

public void localSubscriptionUpdated(String subscriptionName, ReplicatedSubscriptionsSnapshot snapshot) {
    if (log.isDebugEnabled()) {
        log.debug("[{}][{}] Updating subscription to snapshot {}", topic, subscriptionName,
                  snapshot.getClustersList().stream()
                  .map(cmid -> String.format("%s -> %d:%d", cmid.getCluster(),
                                             cmid.getMessageId().getLedgerId(), cmid.getMessageId().getEntryId()))
                  .collect(Collectors.toList()));
    }

    Map<String, MarkersMessageIdData> clusterIds = new TreeMap<>();
    for (int i = 0, size = snapshot.getClustersCount(); i < size; i++) {
        ClusterMessageId cmid = snapshot.getClusterAt(i);
        clusterIds.put(cmid.getCluster(), cmid.getMessageId());
    }

    ByteBuf subscriptionUpdate = Markers.newReplicatedSubscriptionsUpdate(subscriptionName, clusterIds);
    writeMarker(subscriptionUpdate);
}

private void receiveSubscriptionUpdated(ReplicatedSubscriptionsUpdate update) {
  MarkersMessageIdData updatedMessageId = null;
  for (int i = 0, size = update.getClustersCount(); i < size; i++) {
    ClusterMessageId cmid = update.getClusterAt(i);
    if (localCluster.equals(cmid.getCluster())) {
      updatedMessageId = cmid.getMessageId();
    }
  }

  if (updatedMessageId == null) {
    // No updates for this cluster, ignore
    return;
  }

    // 这里的消息ID是上一次的LAC。
  Position pos = new PositionImpl(updatedMessageId.getLedgerId(), updatedMessageId.getEntryId());

  if (log.isDebugEnabled()) {
    log.debug("[{}][{}] Received update for subscription to {}", topic, update.getSubscriptionName(), pos);
  }

    // 更新消费者读取进度为上一次的LAC。
  PersistentSubscription sub = topic.getSubscription(update.getSubscriptionName());
  if (sub != null) {
    sub.acknowledgeMessage(Collections.singletonList(pos), AckType.Cumulative, Collections.emptyMap());
  } else {
    // Subscription doesn't exist. We need to force the creation of the subscription in this cluster, because
    log.info("[{}][{}] Creating subscription at {}:{} after receiving update from replicated subcription",
        topic, update.getSubscriptionName(), updatedMessageId.getLedgerId(), pos);
    topic.createSubscription(update.getSubscriptionName(),
        InitialPosition.Latest, true /* replicateSubscriptionState */);
  }
}

[mac bin]$ ./pulsar-admin namespaces set-backlog-quota --limit 100M --limitTime 1111 --policy producer_request_hold sample/ns118:38:54.968 [AsyncHttpClient-7-1] WARN  org.apache.pulsar.client.admin.internal.BaseResource - [http://x.x.x.x:8080/admin/v2/namespaces/sample/ns1/backlogQuota] Failed to perform http post request: javax.ws.rs.InternalServerErrorException: HTTP 500 Internal Server ErrorHTTP 500 Internal Server Error
Reason: HTTP 500 Internal Server Error[mac bin]$

11:50:05.098 [pulsar-web-63-6] ERROR org.apache.pulsar.broker.admin.impl.NamespacesBase - [null] Failed to update backlog quota map for namespace sample/ns1org.apache.pulsar.metadata.api.MetadataStoreException: com.fasterxml.jackson.databind.JsonMappingException: (was java.lang.NullPointerException) (through reference chain: org.apache.pulsar.common.policies.data.Policies["backlog_quota_map"]->java.util.LinkedHashMap["destination_storage"]->com.sun.proxy.$Proxy117["limitSize"])  at org.apache.pulsar.broker.resources.BaseResources.set(BaseResources.java:94) ~[org.apache.pulsar-pulsar-broker-common-2.8.0.jar:2.8.0]  at org.apache.pulsar.broker.admin.impl.NamespacesBase.internalSetBacklogQuota(NamespacesBase.java:1387) ~[pulsar-broker-2.8.0.jar:2.8.0]  at org.apache.pulsar.broker.admin.v2.Namespaces.setBacklogQuota(Namespaces.java:724) ~[pulsar-broker-2.8.0.jar:2.8.0]  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_231]  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_231]  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_231]  at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_231]  at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:124) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:167) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$VoidOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:159) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:79) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:475) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:397) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:255) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]  at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]  at org.glassfish.jersey.internal.Errors.process(Errors.java:292) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]  at org.glassfish.jersey.internal.Errors.process(Errors.java:274) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]  at org.glassfish.jersey.internal.Errors.process(Errors.java:244) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]  at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265) ~[org.glassfish.jersey.core-jersey-common-2.34.jar:?]  at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:234) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:680) ~[org.glassfish.jersey.core-jersey-server-2.34.jar:?]  at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]  at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:346) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]  at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:366) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]  at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:319) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]  at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205) ~[org.glassfish.jersey.containers-jersey-container-servlet-core-2.34.jar:?]  at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:799) ~[org.eclipse.jetty-jetty-servlet-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.servlet.ServletHandler$ChainEnd.doFilter(ServletHandler.java:1626) ~[org.eclipse.jetty-jetty-servlet-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.apache.pulsar.broker.web.ResponseHandlerFilter.doFilter(ResponseHandlerFilter.java:65) ~[pulsar-broker-2.8.0.jar:2.8.0]  at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193) ~[org.eclipse.jetty-jetty-servlet-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601) ~[org.eclipse.jetty-jetty-servlet-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:548) ~[org.eclipse.jetty-jetty-servlet-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1435) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:501) ~[org.eclipse.jetty-jetty-servlet-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1350) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:234) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.StatisticsHandler.handle(StatisticsHandler.java:179) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.Server.handle(Server.java:516) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:388) ~[org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:633) [org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:380) [org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277) [org.eclipse.jetty-jetty-server-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311) [org.eclipse.jetty-jetty-io-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105) [org.eclipse.jetty-jetty-io-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104) [org.eclipse.jetty-jetty-io-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338) [org.eclipse.jetty-jetty-util-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315) [org.eclipse.jetty-jetty-util-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173) [org.eclipse.jetty-jetty-util-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131) [org.eclipse.jetty-jetty-util-9.4.42.v20210604.jar:9.4.42.v20210604]  at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:383) [org.eclipse.jetty-jetty-util-9.4.42.v20210604.jar:9.4.42.v20210604]  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_231]  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_231]  at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.63.Final.jar:4.1.63.Final]  at java.lang.Thread.run(Thread.java:748) [?:1.8.0_231]

System.setProperty(FastJsonAutoDiscoverable.FASTJSON_AUTO_DISCOVERABLE, "false");

11:50:05.095 [pulsar-web-63-6] INFO  org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl - vvv method readModifyUpdate, path: /admin/policies/sample/ns1, class: class org.apache.pulsar.common.policies.data.Policies, value: Policies(auth_policies=AuthPoliciesImpl(namespaceAuthentication={}, topicAuthentication={}, subscriptionAuthentication={}), replication_clusters=[standalone], bundles=BundlesDataImpl(boundaries=[0x00000000, 0x40000000, 0x80000000, 0xc0000000, 0xffffffff], numBundles=4), backlog_quota_map={destination_storage={}}, clusterDispatchRate={}, topicDispatchRate={}, subscriptionDispatchRate={}, replicatorDispatchRate={}, clusterSubscribeRate={}, persistence=null, deduplicationEnabled=null, autoTopicCreationOverride=null, autoSubscriptionCreationOverride=null, publishMaxMessageRate={}, latency_stats_sample_rate={}, message_ttl_in_seconds=null, subscription_expiration_time_minutes=0, retention_policies=null, deleted=false, encryption_required=false, delayed_delivery_policies=null, inactive_topic_policies=null, subscription_auth_mode=None, max_producers_per_topic=null, max_consumers_per_topic=null, max_consumers_per_subscription=null, max_unacked_messages_per_consumer=null, max_unacked_messages_per_subscription=null, max_subscriptions_per_topic=null, compaction_threshold=null, offload_threshold=-1, offload_deletion_lag_ms=null, max_topics_per_namespace=null, schema_auto_update_compatibility_strategy=Full, schema_compatibility_strategy=UNDEFINED, is_allow_auto_update_schema=true, schema_validation_enforced=false, offload_policies=null, deduplicationSnapshotIntervalSeconds=null, subscription_types_enabled=[], properties={}, resource_group_name=null), hash: -2056889852, expectedVersion: 0, p: 1890262240

10:49:44.719 [metadata-store-32-1] INFO  org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl - vvv method readValueFromStore, path /admin/policies/sample/ns1, class class org.apache.pulsar.common.policies.data.Policies, value Policies(auth_policies=AuthPoliciesImpl(namespaceAuthentication={}, topicAuthentication={}, subscriptionAuthentication={}), replication_clusters=[standalone], bundles=BundlesDataImpl(boundaries=[0x00000000, 0x40000000, 0x80000000, 0xc0000000, 0xffffffff], numBundles=4), backlog_quota_map={destination_storage=BacklogQuotaImpl(limitSize=104857600, limitTime=1111, policy=producer_request_hold)}, clusterDispatchRate={}, topicDispatchRate={}, subscriptionDispatchRate={}, replicatorDispatchRate={}, clusterSubscribeRate={}, persistence=null, deduplicationEnabled=null, autoTopicCreationOverride=null, autoSubscriptionCreationOverride=null, publishMaxMessageRate={}, latency_stats_sample_rate={}, message_ttl_in_seconds=null, subscription_expiration_time_minutes=0, retention_policies=null, deleted=false, encryption_required=false, delayed_delivery_policies=null, inactive_topic_policies=null, subscription_auth_mode=None, max_producers_per_topic=null, max_consumers_per_topic=null, max_consumers_per_subscription=null, max_unacked_messages_per_consumer=null, max_unacked_messages_per_subscription=null, max_subscriptions_per_topic=null, compaction_threshold=null, offload_threshold=-1, offload_deletion_lag_ms=null, max_topics_per_namespace=null, schema_auto_update_compatibility_strategy=Full, schema_compatibility_strategy=UNDEFINED, is_allow_auto_update_schema=true, schema_validation_enforced=false, offload_policies=null, deduplicationSnapshotIntervalSeconds=null, subscription_types_enabled=[], properties={}, resource_group_name=null), hash 1614905576, version: 1, p: 1843660571

11:50:05.094 [pulsar-web-63-6] INFO  org.apache.pulsar.broker.admin.v2.Namespaces - vvv method setBacklogQuota, backlogQuota {}, class com.sun.proxy.$Proxy117 {}

12:14:01.371 [pulsar-web-63-12] INFO  org.apache.pulsar.broker.admin.v2.Namespaces - vvv method setBacklogQuota, backlogQuota VVData(limitSize=104857600, limitTime=1111, policy=producer_request_hold

package org.glassfish.jersey.internal.spi;
import javax.ws.rs.core.FeatureContext;
/** * A service provider contract for JAX-RS and Jersey components that need to be automatically discovered and registered in * {@link javax.ws.rs.core.Configuration runtime configurations}. * <p/> * A component implementing this contract becomes auto-discoverable by adding a new entry with fully qualified name of its * implementation class name to a {@code org.glassfish.jersey.internal.spi.AutoDiscoverable} file in the {@code * META-INF/services} directory. * <p/> * Almost all Jersey {@code AutoDiscoverable} implementations have * {@link #DEFAULT_PRIORITY} {@link javax.annotation.Priority priority} set. * * @author Michal Gajdos */public interface AutoDiscoverable {
    /**     * Default common priority of Jersey build-in auto-discoverables.     * Use lower number on your {@code AutoDiscoverable} implementation to run it before Jersey auto-discoverables     * and vice versa.     * 优先级，在fastjson中设置的优先级是 (DEFAULT_PRIORITY - 1)，所以会优先加载。     */    public static final int DEFAULT_PRIORITY = 2000;
    /**     * A call-back method called when an auto-discoverable component is to be configured in a given runtime configuration scope.     * <p>     * Note that as with {@link javax.ws.rs.core.Feature JAX-RS features}, before registering new JAX-RS components in a     * given configurable context, an auto-discoverable component should verify that newly registered components are not     * already registered in the configurable context.     * </p>     *     * @param context configurable context in which the auto-discoverable should be configured.     */    public void configure(FeatureContext context);}

// AbstractMetadataStore 构造方法
protected AbstractMetadataStore() {
        this.executor = Executors
                .newSingleThreadExecutor(new DefaultThreadFactory("metadata-store"));
        registerListener(this);
 // other code   
}

// 执行查询或者回调函数
/**
  * Run the task in the executor thread and fail the future if the executor is shutting down
  */
protected void execute(Runnable task, CompletableFuture<?> future) {
    try {
        executor.execute(task);
    } catch (Throwable t) {
        future.completeExceptionally(t);
    }
}

@Override
public CompletableFuture<Stat> storePut(String path, byte[] value, Optional<Long> optExpectedVersion,
    EnumSet<CreateOption> options) {
  boolean hasVersion = optExpectedVersion.isPresent();
  int expectedVersion = optExpectedVersion.orElse(-1L).intValue();

  CompletableFuture<Stat> future = new CompletableFuture<>();

  try {
    if (hasVersion && expectedVersion == -1) {
      CreateMode createMode = getCreateMode(options);
      ZkUtils.asyncCreateFullPathOptimistic(zkc, path, value, ZooDefs.Ids.OPEN_ACL_UNSAFE,
          createMode, (rc, path1, ctx, name) -> {
            execute(() -> {
              Code code = Code.get(rc);
              if (code == Code.OK) {
                                // 运行到此处时出现阻塞。
                future.complete(new Stat(name, 0, 0, 0, createMode.isEphemeral(), true));
              } else if (code == Code.NODEEXISTS) {
                // We're emulating a request to create node, so the version is invalid
                future.completeExceptionally(getException(Code.BADVERSION, path));
              } else {
                future.completeExceptionally(getException(code, path));
              }
            }, future);
          }, null);
    } else {
      // other code...
    }
  } catch (Throwable t) {
    future.completeExceptionally(new MetadataStoreException(t));
  }

  return future;
}

"metadata-store-7-1" #84 prio=5 os_prio=0 tid=0x00007f1560004800 nid=0x1ca1 waiting on condition [0x00007f1506a2c000]
   java.lang.Thread.State: WAITING (parking)
  at sun.misc.Unsafe.park(Native Method)
  - parking to wait for  <0x00000000fd0c3d18> (a java.util.concurrent.CompletableFuture$Signaller)
  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
  at java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1693)
  at java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3323)
  at java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1729)
  at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
  at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.recoverCursorFromZK(ManagedLedgerImpl.java:1072)
  at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.asyncOpenCursor(ManagedLedgerImpl.java:1019)
    // 阻塞原因
  - locked <0x00000000f5841ce8> (a org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl)
  at org.apache.pulsar.broker.service.persistent.PersistentTopic.getDurableSubscription(PersistentTopic.java:857)
  at org.apache.pulsar.broker.service.persistent.PersistentTopic.subscribe(PersistentTopic.java:747)
  at org.apache.pulsar.broker.service.ServerCnx.lambda$null$12(ServerCnx.java:1050)
  at org.apache.pulsar.broker.service.ServerCnx$$Lambda$277/1382611823.apply(Unknown Source)
  at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
  at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
  at org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage$$Lambda$295/640790180.accept(Unknown Source)
  at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656)
  at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:632)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
    // 阻塞代码
  at org.apache.pulsar.metadata.impl.ZKMetadataStore.lambda$null$15(ZKMetadataStore.java:234)
  at org.apache.pulsar.metadata.impl.ZKMetadataStore$$Lambda$127/153703270.run(Unknown Source)
  at org.apache.pulsar.metadata.impl.AbstractMetadataStore.lambda$execute$8(AbstractMetadataStore.java:261)
  at org.apache.pulsar.metadata.impl.AbstractMetadataStore$$Lambda$128/1782705605.run(Unknown Source)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
  at java.lang.Thread.run(Thread.java:748)

"main-EventThread" #82 daemon prio=5 os_prio=0 tid=0x00007f15edd8e800 nid=0x1ca0 waiting on condition [0x00007f150692c000]
   java.lang.Thread.State: WAITING (parking)
  at sun.misc.Unsafe.park(Native Method)
  - parking to wait for  <0x00000000f5840660> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
  at java.util.concurrent.locks.AbstractQue

uedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
	at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
	at org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:559)

"main-SendThread(172.20.140.23:2181)" #81 daemon prio=5 os_prio=0 tid=0x00007f15edd8a800 nid=0x1c9f runnable [0x00007f150672a000]
   java.lang.Thread.State: RUNNABLE
	at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
	at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
	at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
	at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
	- locked <0x00000000f571b3e8> (a sun.nio.ch.Util$3)
	- locked <0x00000000f571b3f8> (a java.util.Collections$UnmodifiableSet)
	- locked <0x00000000f571b3a0> (a sun.nio.ch.EPollSelectorImpl)
	at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
	at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:332)
	at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1290)

private boolean recoverCursorFromZK(String cursorName, final OpenCursorCallback callback, final Object ctx) {
  CompletableFuture<Boolean> existsFuture = new CompletableFuture<>();
  store.asyncGetCursorInfo(name, cursorName, new MetaStoreCallback<MLDataFormats.ManagedCursorInfo>() {
    @Override
    public void operationComplete(MLDataFormats.ManagedCursorInfo info, Stat stat) {
      existsFuture.complete(true);
    }

    @Override
    public void operationFailed(MetaStoreException e) {
      log.info("vvv method {} open failed", "recoverCursorFromZK");
      existsFuture.complete(false);
    }
  });
    
  try {
        // 阻塞的原因是因为这条语句
    if (!existsFuture.get()) {
      return false;
    }
  } catch (Exception e) {
    log.error("vvv method recoverCursorFromZK error", e);
    return false;
  }
    
    // other code
  
  return true;
}

public void testFuture() throws Exception {
  ExecutorService executor = Executors.newSingleThreadExecutor();

  CompletableFuture<Boolean> f1 = new CompletableFuture<>();

  new Thread(() -> {
    CompletableFuture<Boolean> f2 = new CompletableFuture<>();
    executor.execute(() -> {
      sleep(1000);

      System.out.println("f2 start");
      f2.complete(true);
      System.out.println("f2 done");
    });

    executor.execute(() -> {
      sleep(1000);

      System.out.println("f1 start");
      f1.complete(true);
      System.out.println("f1 done");
    });
  }).start();

  System.out.println("init done");
  f1.get();
  System.out.println("run done");
}

private void sleep(int i) {
  try {
    Thread.sleep(1000);
  } catch (Exception e) {
    e.printStackTrace();
  }
}

cursorsLedgerId=-1, markDeleteLedgerId=603, markDeleteEntryId=15, lastActive=1629692232349
upper ledgerId=603, upper entryId=18, lower ledgerId=603, lower entryId=16
upper ledgerId=37740, upper entryId=2, lower ledgerId=37740, lower entryId=1
upper ledgerId=37740, upper entryId=4, lower ledgerId=37740, lower entryId=3

lowLedgerId=603, lowEntryId=16, upperLedgerId=603, upperEntryId=18
lowLedgerId=37740, lowEntryId=1, upperLedgerId=37740, upperEntryId=2
lowLedgerId=37740, lowEntryId=3, upperLedgerId=37740, upperEntryId=4
lowLedgerId=37740, lowEntryId=5, upperLedgerId=37740, upperEntryId=6

private boolean shouldPersistUnackRangesToLedger() {
    return cursorLedger != null
        && !isCursorLedgerReadOnly
        && config.getMaxUnackedRangesToPersist() > 0
        && individualDeletedMessages.size() > config.getMaxUnackedRangesToPersistInZk();
}

return topics.computeIfAbsent(topic, (topicName) -> {
        return this.loadOrCreatePersistentTopic(topicName, createIfMissing);
});

# Max number of concurrent topic loading request broker allows to control number of zk-operations
# 限制获取topic信息时，zk并发操作数量
maxConcurrentTopicLoadRequest=5000

// BrokerService -> createPersistentTopic
// Once we have the configuration, we can proceed with the async open operation
managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(), managedLedgerConfig, new OpenLedgerCallback())...

// ManagedLedgerImpl -> synchronized void initialize(final ManagedLedgerInitializeLedgerCallback callback, final Object ctx)
store.getManagedLedgerInfo(name, config.isCreateIfMissing(), new MetaStoreCallback<ManagedLedgerInfo>() {

// ManagedLedgerImpl -> private void initializeCursors(final ManagedLedgerInitializeLedgerCallback callback)store.getCursors(name, new MetaStoreCallback<List<String>>() {...

message ManagedCursorInfo {
    // If the ledger id is -1, then the mark-delete position is
    // the one from the (ledgerId, entryId) snapshot below
    required int64 cursorsLedgerId = 1;
    // Last snapshot of the mark-delete position
    optional int64 markDeleteLedgerId = 2;
    optional int64 markDeleteEntryId = 3;
    repeated MessageRange individualDeletedMessages = 4;
    // Additional custom properties associated with
    // the current cursor position
    repeated LongProperty properties = 5;
    optional int64 lastActive = 6;
    // Store which index in the batch message has been deleted
    repeated BatchedEntryDeletionIndexInfo batchedEntryDeletionIndexInfo = 7;
}

message PositionInfo {
    required int64 ledgerId = 1;
    required int64 entryId = 2;
    repeated MessageRange individualDeletedMessages = 3;
    // Additional custom properties associated with
    // the current cursor position
    repeated LongProperty properties = 4;
    // Store which index in the batch message has been deleted
    repeated BatchedEntryDeletionIndexInfo batchedEntryDeletionIndexInfo = 5;
}

# 权限验证
14:47:56.559 [pulsar-io-4-1] INFO  auth.server.VVPulsarAuthorizationProvider - vv_auth_v2 allowTopicOperationAsync, topicName persistent://tenant_c/ns1/topic_cursor, role vv-role
# 执行lookup阶段
14:47:56.561 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - vvv_try_to_lookup,topicName: persistent://tenant_c/ns1/topic_cursor, requestId: 2
# 权限验证
14:47:56.563 [pulsar-io-4-1] INFO  auth.server.VVPulsarAuthorizationProvider - vv_auth_v2 allowTopicOperationAsync, topicName persistent://tenant_c/ns1/topic_cursor, role vv-role
# 权限验证
14:47:56.644 [pulsar-io-4-1] INFO  auth.server.VVPulsarAuthorizationProvider - vv_auth_v2 allowTopicOperationAsync, topicName persistent://tenant_c/ns1/topic_cursor, role vv-role
# consumer发送subscribe消息（注意的是创建ledger在subscribe之后，）
14:47:56.644 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - vvv_subscribe, consumer_002 Subscribing on topic persistent://tenant_c/ns1/topic_cursor
14:47:56.646 [pulsar-io-4-1] INFO  org.apache.pulsar.broker.service.ServerCnx - [/172.21.32.64:59676] Subscribing on topic persistent://tenant_c/ns1/topic_cursor / consumer_002
# 这里由于broker刚重启，没有加载ledger信息，所以在consumer连接后才创建ledger对象
14:47:56.640 [pulsar-ordered-OrderedExecutor-7-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl - vv_cc asyncOpen, name=tenant_c/ns1/persistent/topic_cursor
# 创建ledger
14:47:56.667 [pulsar-ordered-OrderedExecutor-7-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - Opening managed ledger tenant_c/ns1/persistent/topic_cursor
# 从zk中获取ledger信息
14:47:56.668 [pulsar-ordered-OrderedExecutor-7-0] INFO  org.apache.bookkeeper.mledger.impl.MetaStoreImpl - vvv_debug /managed-ledgers/tenant_c/ns1/persistent/topic_cursor
14:47:56.722 [BookKeeperClientWorker-OrderedExecutor-0-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] Creating ledger, metadata: {component=[109, 97, 110, 97, 103, 101, 100, 45, 108, 101, 100, 103, 101, 114], pulsar/managed-ledger=[116, 101, 110, 97, 110, 116, 95, 99, 47, 110, 115, 49, 47, 112, 101, 114, 115, 105, 115, 116, 101, 110, 116, 47, 116, 111, 112, 105, 99, 95, 99, 117, 114, 115, 111, 114], application=[112, 117, 108, 115, 97, 114]} - metadata ops timeout : 60 seconds
# ledger创建成功
14:47:56.761 [BookKeeperClientWorker-OrderedExecutor-0-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] Created ledger 603
# 开始加载cursor信息
14:47:56.778 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] Loading cursor consumer_002
14:47:56.792 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [tenant_c/ns1/persistent/topic_cursor] Recovering from bookkeeper ledger cursor: consumer_002
# 从zk中获取之前保存的cursor信息，主要包含cursorLedgerId、markDeletedLedgerId和markDeletedEntryId
14:47:56.792 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.MetaStoreImpl - vv_cc asyncGetCursorInfo, ledgerName=tenant_c/ns1/persistent/topic_cursor, cursorName=consumer_002, path=/managed-ledgers/tenant_c/ns1/persistent/topic_cursor/consumer_002
# 开始恢复cursor
14:47:56.798 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [tenant_c/ns1/persistent/topic_cursor] Cursor consumer_002 recovered to position 56:9
14:47:56.800 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] Recovery for cursor consumer_002 completed. pos=56:9 -- todo=0
# cursor的恢复是在创建ledger的过程中进行的，cursor恢复后则ledger创建过程结束。
14:47:56.801 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl - [tenant_c/ns1/persistent/topic_cursor] Successfully initialize managed ledger
14:47:56.808 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.pulsar.broker.service.AbstractTopic - Disabling publish throttling for persistent://tenant_c/ns1/topic_cursor
14:47:56.828 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://tenant_c/ns1/topic_cursor] There are no replicated subscriptions on the topic
14:47:56.837 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.pulsar.broker.service.BrokerService - Created topic persistent://tenant_c/ns1/topic_cursor - dedup is disabled
14:47:56.843 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - Reset cursor:ManagedCursorImpl{ledger=tenant_c/ns1/persistent/topic_cursor, name=consumer_002, ackPos=56:9, readPos=56:10} to 603:-1 since ledger consumed completely
14:47:56.847 [BookKeeperClientWorker-OrderedExecutor-0-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] Ledger 56 contains the current last confirmed entry 56:9, and it is going to be deleted
14:47:56.858 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] End TrimConsumedLedgers. ledgers=1 totalSize=0
14:47:56.858 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [tenant_c/ns1/persistent/topic_cursor] Removing ledger 56 - size: 673
14:47:57.251 [ForkJoinPool.commonPool-worker-5] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - vv_cc asyncOpenCursor, cursorName=consumer_002
14:47:57.262 [ForkJoinPool.commonPool-worker-5] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [tenant_c/ns1/persistent/topic_cursor-consumer_002] Rewind from 56:10 to 56:10
14:47:57.263 [ForkJoinPool.commonPool-worker-5] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://tenant_c/ns1/topic_cursor] There are no replicated subscriptions on the topic
14:47:57.263 [ForkJoinPool.commonPool-worker-5] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://tenant_c/ns1/topic_cursor][consumer_002] Created new subscription for 0
14:47:57.264 [ForkJoinPool.commonPool-worker-5] INFO  org.apache.pulsar.broker.service.ServerCnx - [/172.21.32.64:59676] Created subscription on topic persistent://tenant_c/ns1/topic_cursor / consumer_002