public void validatePipePluginUsageByTopicInternal(String pipePluginName)
      throws SubscriptionException {
    acquireReadLock();
    try {
      topicMetaKeeper
          .getAllTopicMeta()
          .forEach(
              meta -> {
                if (pipePluginName.equals(meta.getConfig().getAttribute().get("processor"))) {
                  final String exceptionMessage =
                      String.format(
                          "PipePlugin '%s' is already used by Topic '%s' as a processor.",
                          pipePluginName, meta.getTopicName());
                  LOGGER.warn(exceptionMessage);
                  throw new SubscriptionException(exceptionMessage);
                }
              });
    } finally {
      releaseReadLock();
    }
  }
Caller should ensure that the method is called in the lock {@link #acquireWriteLock}.
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java