private void syncInternal(final AbstractSubscriptionConsumer consumer) {
    if (hasNoAvailableProviders()) {
      try {
        openProviders(consumer);
      } catch (final Exception e) {
        LOGGER.warn("Failed to open providers for consumer {} because of {}", consumer, e, e);
        return;
      }
    }

    final Map<Integer, TEndPoint> allEndPoints;
    try {
      allEndPoints = consumer.fetchAllEndPointsWithRedirection();
    } catch (final Exception e) {
      LOGGER.warn("Failed to fetch all endpoints for consumer {} because of {}", consumer, e, e);
      return;
    }

    // add new providers or handshake existing providers
    for (final Map.Entry<Integer, TEndPoint> entry : allEndPoints.entrySet()) {
      final AbstractSubscriptionProvider provider = getProvider(entry.getKey());
      if (Objects.isNull(provider)) {
        // new provider
        final TEndPoint endPoint = entry.getValue();
        final AbstractSubscriptionProvider newProvider;
        try {
          newProvider = consumer.constructProviderAndHandshake(endPoint);
        } catch (final Exception e) {
          LOGGER.warn(
              "{} failed to create connection with {} because of {}", consumer, endPoint, e, e);
          continue;
        }
        addProvider(entry.getKey(), newProvider);
      } else {
        // existing provider
        try {
          consumer.subscribedTopics = provider.heartbeat().getTopics();
          provider.setAvailable();
        } catch (final Exception e) {
          LOGGER.warn(
              "{} failed to sending heartbeat to subscription provider {} because of {}, set subscription provider unavailable",
              consumer,
              provider,
              e,
              e);
          provider.setUnavailable();
        }
        // close and remove unavailable provider (reset the connection as much as possible)
        if (!provider.isAvailable()) {
          try {
            closeAndRemoveProvider(entry.getKey());
          } catch (final Exception e) {
            LOGGER.warn(
                "Exception occurred when {} closing and removing subscription provider {} because of {}",
                consumer,
                provider,
                e,
                e);
          }
        }
      }
    }

    // close and remove stale providers
    for (final AbstractSubscriptionProvider provider : getAllProviders()) {
      final int dataNodeId = provider.getDataNodeId();
      if (!allEndPoints.containsKey(dataNodeId)) {
        try {
          closeAndRemoveProvider(dataNodeId);
        } catch (final Exception e) {
          LOGGER.warn(
              "Exception occurred when {} closing and removing subscription provider {} because of {}",
              consumer,
              provider,
              e,
              e);
        }
      }
    }
  }
Caller should ensure that the method is called in the lock {@link #acquireReadLock()}.
iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java