public void bindPrefetchingQueue(final SubscriptionConnectorSubtask subtask) {
    final String consumerGroupId = subtask.getConsumerGroupId();
    consumerGroupIdToSubscriptionBroker
        .compute(
            consumerGroupId,
            (id, broker) -> {
              if (Objects.isNull(broker)) {
                LOGGER.info(
                    "Subscription: broker bound to consumer group [{}] does not exist, create new for binding prefetching queue",
                    consumerGroupId);
                // TODO: consider more robust metadata semantics
                return new SubscriptionBroker(consumerGroupId);
              }
              return broker;
            })
        .bindPrefetchingQueue(subtask.getTopicName(), subtask.getInputPendingQueue());
    prefetchingQueueCount.invalidate();
  }
@return {@code true} if drop broker success, {@code false} otherwise
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java