package org.apache.helix.spectator;

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import javax.management.JMException;

import com.google.common.collect.ImmutableMap;
import org.apache.helix.HelixConstants;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixException;
import org.apache.helix.HelixManager;
import org.apache.helix.NotificationContext;
import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyType;
import org.apache.helix.api.listeners.ConfigChangeListener;
import org.apache.helix.api.listeners.CurrentStateChangeListener;
import org.apache.helix.api.listeners.CustomizedViewChangeListener;
import org.apache.helix.api.listeners.CustomizedViewRootChangeListener;
import org.apache.helix.api.listeners.ExternalViewChangeListener;
import org.apache.helix.api.listeners.InstanceConfigChangeListener;
import org.apache.helix.api.listeners.LiveInstanceChangeListener;
import org.apache.helix.api.listeners.PreFetch;
import org.apache.helix.api.listeners.RoutingTableChangeListener;
import org.apache.helix.common.ClusterEventProcessor;
import org.apache.helix.common.caches.CurrentStateSnapshot;
import org.apache.helix.controller.stages.AttributeName;
import org.apache.helix.controller.stages.ClusterEvent;
import org.apache.helix.controller.stages.ClusterEventType;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.CustomizedView;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.monitoring.mbeans.RoutingTableProviderMonitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RoutingTableProvider
    implements ExternalViewChangeListener, InstanceConfigChangeListener, ConfigChangeListener,
               LiveInstanceChangeListener, CurrentStateChangeListener, CustomizedViewChangeListener,
               CustomizedViewRootChangeListener {
  private static final Logger logger = LoggerFactory.getLogger(RoutingTableProvider.class);
  private static final long DEFAULT_PERIODIC_REFRESH_INTERVAL = 300000L; // 5 minutes
  private final Map<String, AtomicReference<RoutingTable>> _routingTableRefMap;
  private final HelixManager _helixManager;
  private final RouterUpdater _routerUpdater;
  private final Map<PropertyType, List<String>> _sourceDataTypeMap;
  private final Map<RoutingTableChangeListener, ListenerContext> _routingTableChangeListenerMap;
  private final Map<PropertyType, RoutingTableProviderMonitor> _monitorMap;

  // For periodic refresh
  private long _lastRefreshTimestamp;
  private boolean _isPeriodicRefreshEnabled = true; // Default is enabled
  private long _periodRefreshInterval;
  private ScheduledThreadPoolExecutor _periodicRefreshExecutor;
  // For computing intensive reporting logic
  private ExecutorService _reportExecutor;
  private Future _reportingTask = null;

  protected static final  String DEFAULT_PROPERTY_TYPE = "HELIX_DEFAULT_PROPERTY";
  protected static final  String DEFAULT_STATE_TYPE = "HELIX_DEFAULT";


  public RoutingTableProvider() {
    this(null);
  }

  public RoutingTableProvider(HelixManager helixManager) throws HelixException {
    this(helixManager, ImmutableMap.of(PropertyType.EXTERNALVIEW, Collections.emptyList()), true,
        DEFAULT_PERIODIC_REFRESH_INTERVAL);
  }

  public RoutingTableProvider(HelixManager helixManager, PropertyType sourceDataType)
      throws HelixException {
    this(helixManager, ImmutableMap.of(sourceDataType, Collections.emptyList()), true,
        DEFAULT_PERIODIC_REFRESH_INTERVAL);
  }

  public RoutingTableProvider(HelixManager helixManager, Map<PropertyType, List<String>> sourceDataTypeMap) {
    this(helixManager, sourceDataTypeMap, true, DEFAULT_PERIODIC_REFRESH_INTERVAL);
  }

  /**
   * Initialize an instance of RoutingTableProvider
   * @param helixManager
   * @param sourceDataType
   * @param isPeriodicRefreshEnabled true if periodic refresh is enabled, false otherwise
   * @param periodRefreshInterval only effective if isPeriodRefreshEnabled is true
   * @throws HelixException
   */
  public RoutingTableProvider(HelixManager helixManager, PropertyType sourceDataType,
      boolean isPeriodicRefreshEnabled, long periodRefreshInterval) throws HelixException {
    this(helixManager, ImmutableMap.of(sourceDataType, Collections.emptyList()),
        isPeriodicRefreshEnabled, periodRefreshInterval);
  }

  /**
   * Initialize an instance of RoutingTableProvider
   * @param helixManager
   * @param sourceDataTypeMap
   * @param isPeriodicRefreshEnabled true if periodic refresh is enabled, false otherwise
   * @param periodRefreshInterval only effective if isPeriodRefreshEnabled is true
   * @throws HelixException
   */
  public RoutingTableProvider(HelixManager helixManager,
      Map<PropertyType, List<String>> sourceDataTypeMap, boolean isPeriodicRefreshEnabled,
      long periodRefreshInterval) throws HelixException {

    validateSourceDataTypeMap(sourceDataTypeMap);

    _routingTableRefMap = new HashMap<>();
    _helixManager = helixManager;
    _sourceDataTypeMap = sourceDataTypeMap;
    _routingTableChangeListenerMap = new ConcurrentHashMap<>();
    String clusterName = _helixManager != null ? _helixManager.getClusterName() : null;

    // Initialize the tables
    for (PropertyType propertyType : _sourceDataTypeMap.keySet()) {
      if (_sourceDataTypeMap.get(propertyType).size() == 0) {
        if (propertyType.equals(PropertyType.CUSTOMIZEDVIEW)) {
          throw new HelixException("CustomizedView has been used without any aggregation type!");
        }
        String key = generateReferenceKey(propertyType.name(),  DEFAULT_STATE_TYPE);
        if (_routingTableRefMap.get(key) == null) {
          _routingTableRefMap.put(key, new AtomicReference<>(new RoutingTable(propertyType)));
        }
      } else {
        if (!propertyType.equals(PropertyType.CUSTOMIZEDVIEW)) {
          throw new HelixException(
              String.format("Type %s has been used in addition to the propertyType %s !",
                  sourceDataTypeMap.get(propertyType), propertyType.name()));
        }
        for (String customizedStateType : _sourceDataTypeMap.get(propertyType)) {
          String key = generateReferenceKey(propertyType.name(),  customizedStateType);
          if (_routingTableRefMap.get(key) == null) {
            _routingTableRefMap.put(key, new AtomicReference<>(
                new CustomizedViewRoutingTable(propertyType, customizedStateType)));
          }
        }
      }
    }

    // Start Monitoring
    _monitorMap = new HashMap<>();

    for (PropertyType propertyType : _sourceDataTypeMap.keySet()) {
      _monitorMap.put(propertyType, new RoutingTableProviderMonitor(propertyType, clusterName));
      try {
        _monitorMap.get(propertyType).register();
      } catch (JMException e) {
        logger.error("Failed to register RoutingTableProvider monitor MBean.", e);
      }
    }
    _reportExecutor = Executors.newSingleThreadExecutor();

    // Start Updaters
    _routerUpdater = new RouterUpdater(clusterName, sourceDataTypeMap);
    _routerUpdater.start();

    // Add listeners
    addListeners();

    // For periodic refresh
    if (isPeriodicRefreshEnabled && _helixManager != null) {
      _lastRefreshTimestamp = System.currentTimeMillis(); // Initialize timestamp with current time
      _periodRefreshInterval = periodRefreshInterval;
      // Construct a periodic refresh context
      final NotificationContext periodicRefreshContext = new NotificationContext(_helixManager);
      periodicRefreshContext.setType(NotificationContext.Type.PERIODIC_REFRESH);
      // Create a thread that runs at specified interval
      _periodicRefreshExecutor = new ScheduledThreadPoolExecutor(1);
      _periodicRefreshExecutor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
          // If enough time has elapsed since last refresh, queue a refresh event
          if (_lastRefreshTimestamp + _periodRefreshInterval < System.currentTimeMillis()) {
            // changeType is irrelevant for NotificationContext.Type.PERIODIC_REFRESH
            _routerUpdater.queueEvent(periodicRefreshContext, ClusterEventType.PeriodicalRebalance,
                null);
          }
        }
      }, _periodRefreshInterval, _periodRefreshInterval, TimeUnit.MILLISECONDS);
    } else {
      _isPeriodicRefreshEnabled = false;
    }
  }

  /**
   * A method that adds the ChangeListeners to HelixManager
   */
  private void addListeners() {
    if (_helixManager != null) {
      for (PropertyType propertyType : _sourceDataTypeMap.keySet()) {
        switch (propertyType) {
        case EXTERNALVIEW:
          try {
            _helixManager.addExternalViewChangeListener(this);
          } catch (Exception e) {
            shutdown();
            throw new HelixException("Failed to attach ExternalView Listener to HelixManager!", e);
          }
          break;
        case CUSTOMIZEDVIEW:
          // Add CustomizedView root change listener
          try {
            _helixManager.addCustomizedViewRootChangeListener(this);
          } catch (Exception e) {
            shutdown();
            throw new HelixException(
                "Failed to attach CustomizedView Root Listener to HelixManager!", e);
          }
          // Add individual listeners for each customizedStateType
          List<String> customizedStateTypes = _sourceDataTypeMap.get(propertyType);
          for (String customizedStateType : customizedStateTypes) {
            try {
              _helixManager.addCustomizedViewChangeListener(this, customizedStateType);
            } catch (Exception e) {
              shutdown();
              throw new HelixException(String.format(
                  "Failed to attach CustomizedView Listener to HelixManager for type %s!",
                  customizedStateType), e);
            }
          }
          break;
        case TARGETEXTERNALVIEW:
          // Check whether target external has been enabled or not
          if (!_helixManager.getHelixDataAccessor().getBaseDataAccessor().exists(
              _helixManager.getHelixDataAccessor().keyBuilder().targetExternalViews().getPath(),
              0)) {
            shutdown();
            throw new HelixException("Target External View is not enabled!");
          }

          try {
            _helixManager.addTargetExternalViewChangeListener(this);
          } catch (Exception e) {
            shutdown();
            throw new HelixException(
                "Failed to attach TargetExternalView Listener to HelixManager!", e);
          }
          break;
        case CURRENTSTATES:
          // CurrentState change listeners will be added later in LiveInstanceChange call.
          break;
        default:
          throw new HelixException(String.format("Unsupported source data type: %s", propertyType));
        }
      }
      try {
        _helixManager.addInstanceConfigChangeListener(this);
        _helixManager.addLiveInstanceChangeListener(this);
      } catch (Exception e) {
        shutdown();
        throw new HelixException(
            "Failed to attach InstanceConfig and LiveInstance Change listeners to HelixManager!",
            e);
      }
    }
  }

  /**
   * Check and validate the input of the sourceDataTypeMap parameter
   * @param sourceDataTypeMap
   */
  private void validateSourceDataTypeMap(Map<PropertyType, List<String>> sourceDataTypeMap) {
    if (sourceDataTypeMap == null) {
      throw new IllegalArgumentException(
          "The sourceDataTypeMap of Routing Table Provider should not be null");
    }
    for (PropertyType propertyType : sourceDataTypeMap.keySet()) {
      if (propertyType.equals(PropertyType.CUSTOMIZEDVIEW)
          && sourceDataTypeMap.get(propertyType).size() == 0) {
        logger.error("CustomizedView has been used without any aggregation type!");
        throw new HelixException("CustomizedView has been used without any aggregation type!");
      }
      if (!propertyType.equals(PropertyType.CUSTOMIZEDVIEW)
          && sourceDataTypeMap.get(propertyType).size() != 0) {
        logger
            .error("Type has been used in addition to the propertyType {} !", propertyType.name());
        throw new HelixException(String
            .format("Type %s has been used in addition to the propertyType %s !",
                sourceDataTypeMap.get(propertyType), propertyType.name()));
      }
    }
  }

  /**
   * Shutdown current RoutingTableProvider. Once it is shutdown, it should never be reused.
   */
  public void shutdown() {
    if (_periodicRefreshExecutor != null) {
      _periodicRefreshExecutor.purge();
      _periodicRefreshExecutor.shutdown();
    }
    _routerUpdater.shutdown();

    for (PropertyType propertyType : _monitorMap.keySet()) {
      _monitorMap.get(propertyType).unregister();
    }

    if (_helixManager != null) {
      PropertyKey.Builder keyBuilder = _helixManager.getHelixDataAccessor().keyBuilder();
      // Clean up all the listeners.
      _helixManager.removeListener(keyBuilder.liveInstances(), this);
      _helixManager.removeListener(keyBuilder.instanceConfigs(), this);
      for (PropertyType propertyType : _sourceDataTypeMap.keySet()) {
        switch (propertyType) {
        case EXTERNALVIEW:
          _helixManager.removeListener(keyBuilder.externalViews(), this);
          break;
        case CUSTOMIZEDVIEW:
          List<String> customizedStateTypes = _sourceDataTypeMap.get(propertyType);
          // Remove listener on each individual customizedStateType
          for (String customizedStateType : customizedStateTypes) {
            _helixManager.removeListener(keyBuilder.customizedView(customizedStateType), this);
          }
          break;
        case TARGETEXTERNALVIEW:
          _helixManager.removeListener(keyBuilder.targetExternalViews(), this);
          break;
        case CURRENTSTATES:
          NotificationContext context = new NotificationContext(_helixManager);
          context.setType(NotificationContext.Type.FINALIZE);
          updateCurrentStatesListeners(Collections.emptyList(), context);
          break;
        default:
          break;
        }
      }
    }
  }

  /**
   * Get an snapshot of current RoutingTable information. The snapshot is immutable, it reflects the
   * routing table information at the time this method is called.
   * @return snapshot of current routing table.
   */
  public RoutingTableSnapshot getRoutingTableSnapshot() {
    return new RoutingTableSnapshot(getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE));
  }

  /**
   * Get an snapshot of current RoutingTable information for specific PropertyType.
   * The snapshot is immutable, it reflects the routing table information at the time this method is
   * called.
   * @return snapshot of current routing table.
   */
  public RoutingTableSnapshot getRoutingTableSnapshot(PropertyType propertyType) {
    return new RoutingTableSnapshot(getRoutingTableRef(propertyType.name(), DEFAULT_STATE_TYPE));
  }

  /**
   * Get an snapshot of all of the available RoutingTable information. The snapshot is immutable, it
   * reflects the routing table information at the time this method is called.
   * @return snapshot associated with specific propertyType and type.
   */
  public RoutingTableSnapshot getRoutingTableSnapshot(PropertyType propertyType, String stateType) {
    return new RoutingTableSnapshot(getRoutingTableRef(propertyType.name(), stateType));
  }

  /**
   * Get an snapshot of all of the available RoutingTable information. The snapshot is immutable, it
   * reflects the routing table information at the time this method is called.
   * @return all of the available snapshots of current routing table.
   */
  public Map<String, Map<String, RoutingTableSnapshot>> getRoutingTableSnapshots() {
    Map<String, Map<String, RoutingTableSnapshot>> snapshots = new HashMap<>();
    for (String key : _routingTableRefMap.keySet()) {
      RoutingTable routingTable = _routingTableRefMap.get(key).get();
      String propertyTypeName = routingTable.getPropertyType().name();
      String customizedStateType = routingTable.getStateType();
      if (!snapshots.containsKey(propertyTypeName)) {
        snapshots.put(propertyTypeName, new HashMap<>());
      }
      snapshots.get(propertyTypeName).put(customizedStateType,
          new RoutingTableSnapshot(routingTable));
    }
    return snapshots;
  }


  /**
   * Add RoutingTableChangeListener with user defined context
   * @param routingTableChangeListener
   * @param context user defined context
   */
  public void addRoutingTableChangeListener(
      final RoutingTableChangeListener routingTableChangeListener, Object context) {
    addRoutingTableChangeListener(routingTableChangeListener, context, false);
  }

  /**
   * Add RoutingTableChangeListener with user defined context
   * @param routingTableChangeListener
   * @param context user defined context
   * @param isTriggerCallback whether to trigger the initial callback during adding listener
   */
  public void addRoutingTableChangeListener(
      final RoutingTableChangeListener routingTableChangeListener, Object context,
      boolean isTriggerCallback) {
    _routingTableChangeListenerMap.put(routingTableChangeListener, new ListenerContext(context));
    logger.info("Attach RoutingTableProviderChangeListener {}.",
        routingTableChangeListener.getClass().getName());
    if (isTriggerCallback) {
      logger.info("Force triggering a callback for the new listener in routing table provider");
      final NotificationContext periodicRefreshContext = new NotificationContext(_helixManager);
      periodicRefreshContext.setType(NotificationContext.Type.PERIODIC_REFRESH);
      _routerUpdater.queueEvent(periodicRefreshContext, ClusterEventType.PeriodicalRebalance, null);
      }
  }

  /**
   * Remove RoutingTableChangeListener
   * @param routingTableChangeListener
   */
  public Object removeRoutingTableChangeListener(
      final RoutingTableChangeListener routingTableChangeListener) {
    logger.info("Detach RoutingTableProviderChangeListener {}",
        routingTableChangeListener.getClass().getName());
    return _routingTableChangeListenerMap.remove(routingTableChangeListener);
  }

  /**
   * returns the instances for {resource,partition} pair that are in a specific {state}.
   * This method will be deprecated, please use the
   * {@link #getInstancesForResource(String, String, String)} getInstancesForResource} method.
   * @param resourceName
   * @param partitionName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public List<InstanceConfig> getInstances(String resourceName, String partitionName,
      String state) {
    return getInstancesForResource(resourceName, partitionName, state);
  }

  /**
   * returns the instances for {resource,partition} pair that are in a specific {state}
   * @param resourceName
   * @param partitionName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public List<InstanceConfig> getInstancesForResource(String resourceName, String partitionName,
      String state) {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE)
        .getInstancesForResource(resourceName, partitionName, state);
  }

  /**
   * returns the instances for {resource group,partition} pair in all resources belongs to the given
   * resource group that are in a specific {state}.
   * The return results aggregate all partition states from all the resources in the given resource
   * group.
   * @param resourceGroupName
   * @param partitionName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public List<InstanceConfig> getInstancesForResourceGroup(String resourceGroupName,
      String partitionName, String state) {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE)
        .getInstancesForResourceGroup(resourceGroupName, partitionName, state);
  }

  /**
   * returns the instances for {resource group,partition} pair contains any of the given tags
   * that are in a specific {state}.
   * Find all resources belongs to the given resource group that have any of the given resource tags
   * and return the aggregated partition states from all these resources.
   * @param resourceGroupName
   * @param partitionName
   * @param state
   * @param resourceTags
   * @return empty list if there is no instance in a given state
   */
  public List<InstanceConfig> getInstancesForResourceGroup(String resourceGroupName,
      String partitionName, String state, List<String> resourceTags) {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE)
        .getInstancesForResourceGroup(resourceGroupName, partitionName, state, resourceTags);
  }

  /**
   * For specific routing table associated with {propertyType, stateType}
   * returns all instances for {resource} that are in a specific {state}
   * This method will be deprecated, please use the
   * {@link #getInstancesForResource(String, String) getInstancesForResource} method.
   * @param resourceName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public Set<InstanceConfig> getInstances(String resourceName, String state) {
    return getInstancesForResource(resourceName, state);
  }

  /**
   * returns all instances for {resource} that are in a specific {state}.
   * @param resourceName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public Set<InstanceConfig> getInstancesForResource(String resourceName, String state) {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE)
        .getInstancesForResource(resourceName, state);
  }

  /**
   * returns all instances for all resources in {resource group} that are in a specific {state}
   * @param resourceGroupName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public Set<InstanceConfig> getInstancesForResourceGroup(String resourceGroupName, String state) {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE)
        .getInstancesForResourceGroup(resourceGroupName, state);
  }

  /**
   * returns all instances for resources contains any given tags in {resource group} that are in a
   * specific {state}
   * @param resourceGroupName
   * @param state
   * @return empty list if there is no instance in a given state
   */
  public Set<InstanceConfig> getInstancesForResourceGroup(String resourceGroupName, String state,
      List<String> resourceTags) {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE)
        .getInstancesForResourceGroup(resourceGroupName, state, resourceTags);
  }

  /**
   * Return all liveInstances in the cluster now.
   * @return
   */
  public Collection<LiveInstance> getLiveInstances() {
    // Since line instances will be the same across all _routingTableRefMap, here one of the keys
    // will be used without considering PropertyType
    // TODO each table will keep a separate instance list.This can be improve by only keeping one
    // copy of the data
    return _routingTableRefMap.values().iterator().next().get().getLiveInstances();
  }

  /**
   * Return all instance's config in this cluster.
   * @return
   */
  public Collection<InstanceConfig> getInstanceConfigs() {
    // Since line instances will be the same across all _routingTableRefMap, here one of the keys
    // will be used without considering PropertyType
    // TODO each table will keep a separate instance list.This can be improve by only keeping one copy of the data
    return _routingTableRefMap.values().iterator().next().get().getInstanceConfigs();
  }

  /**
   * Return names of all resources (shown in ExternalView or CustomizedView) in this cluster.
   */
  public Collection<String> getResources() {
    return getRoutingTableRef(DEFAULT_PROPERTY_TYPE, DEFAULT_STATE_TYPE).getResources();
  }

  /**
   * Provide the key associated with specific PropertyType and StateType for _routingTableRefMap lookup.
   * @param propertyTypeName
   * @param stateType
   * @return
   */
  private RoutingTable getRoutingTableRef(String propertyTypeName, String stateType) {
    if (propertyTypeName.equals(DEFAULT_PROPERTY_TYPE)) {
      // Check whether there exist only one snapshot (_routingTableRefMap)
      if (_routingTableRefMap.keySet().size() == 1) {
        String key = _routingTableRefMap.keySet().iterator().next();
        if (!_routingTableRefMap.containsKey(key)) {
          throw new HelixException(
              String.format("Currently there is no snapshot available for PropertyType %s and stateType %s",
                  propertyTypeName, stateType));
        }
        return _routingTableRefMap.get(key).get();
      } else {
        throw new HelixException("There is none or more than one RoutingTableSnapshot");
      }
    }

    if (stateType.equals(DEFAULT_STATE_TYPE)) {
      if (propertyTypeName.equals(PropertyType.CUSTOMIZEDVIEW.name())) {
        throw new HelixException("Specific type needs to be used for CUSTOMIZEDVIEW PropertyType");
      }
    }

    String key = generateReferenceKey(propertyTypeName,  stateType);
    if (!_routingTableRefMap.containsKey(key)) {
      throw new HelixException(
          String.format("Currently there is no snapshot available for PropertyType %s and stateType %s",
              propertyTypeName, stateType));
    }
    return _routingTableRefMap.get(key).get();
  }

  private String generateReferenceKey(String propertyType, String stateType) {
    return propertyType + "_" + stateType;
  }

  @Override
  @PreFetch(enabled = false)
  public void onExternalViewChange(List<ExternalView> externalViewList,
      NotificationContext changeContext) {
    HelixConstants.ChangeType changeType = changeContext.getChangeType();
    if (changeType != null && !_sourceDataTypeMap.containsKey(changeType.getPropertyType())) {
      logger.warn(
          "onExternalViewChange called with mismatched change types. Source data types does not contain changed data type: {}",
          changeType);
      return;
    }
    // Refresh with full list of external view.
    if (externalViewList != null && externalViewList.size() > 0) {
      // keep this here for back-compatibility, application can call onExternalViewChange directly
      // with externalview list supplied.
      String keyReference = generateReferenceKey(PropertyType.EXTERNALVIEW.name(),  DEFAULT_STATE_TYPE);
      HelixDataAccessor accessor = changeContext.getManager().getHelixDataAccessor();
      PropertyKey.Builder keyBuilder = accessor.keyBuilder();
      List<InstanceConfig> configList = accessor.getChildValues(keyBuilder.instanceConfigs(), true);
      List<LiveInstance> liveInstances = accessor.getChildValues(keyBuilder.liveInstances(), true);
      refreshExternalView(externalViewList, configList, liveInstances, keyReference);
    } else {
      ClusterEventType eventType;
      if (_sourceDataTypeMap.containsKey(PropertyType.EXTERNALVIEW)) {
        eventType = ClusterEventType.ExternalViewChange;
      } else if (_sourceDataTypeMap.containsKey(PropertyType.TARGETEXTERNALVIEW)) {
        eventType = ClusterEventType.TargetExternalViewChange;
      } else {
        logger.warn(
            "onExternalViewChange called with mismatched change types. Source data types does not contain changed data type: {}",
            changeType);
        return;
      }
      _routerUpdater.queueEvent(changeContext, eventType, changeType);
    }
  }

  @Override
  @PreFetch(enabled = false)
  public void onInstanceConfigChange(List<InstanceConfig> configs,
      NotificationContext changeContext) {
    _routerUpdater.queueEvent(changeContext, ClusterEventType.InstanceConfigChange,
        HelixConstants.ChangeType.INSTANCE_CONFIG);
  }

  @Override
  @PreFetch(enabled = false)
  public void onConfigChange(List<InstanceConfig> configs, NotificationContext changeContext) {
    onInstanceConfigChange(configs, changeContext);
  }

  @Override
  @PreFetch(enabled = true)
  public void onLiveInstanceChange(List<LiveInstance> liveInstances,
      NotificationContext changeContext) {
    if (_sourceDataTypeMap.containsKey(PropertyType.CURRENTSTATES)) {
      // Go though the live instance list and update CurrentState listeners
      updateCurrentStatesListeners(liveInstances, changeContext);
    }
    _routerUpdater.queueEvent(changeContext, ClusterEventType.LiveInstanceChange,
        HelixConstants.ChangeType.LIVE_INSTANCE);
  }

  @Override
  @PreFetch(enabled = false)
  public void onStateChange(String instanceName, List<CurrentState> statesInfo,
      NotificationContext changeContext) {
    if (_sourceDataTypeMap.containsKey(PropertyType.CURRENTSTATES)) {
      _routerUpdater.queueEvent(changeContext, ClusterEventType.CurrentStateChange,
          HelixConstants.ChangeType.CURRENT_STATE);
    } else {
      logger.warn(
          "RoutingTableProvider does not use CurrentStates as source, ignore CurrentState changes!");
    }
  }

  @Override
  @PreFetch(enabled = false)
  public void onCustomizedViewChange(List<CustomizedView> customizedViewList,
      NotificationContext changeContext) {
    if (_sourceDataTypeMap.containsKey(PropertyType.CUSTOMIZEDVIEW)) {
      _routerUpdater.queueEvent(changeContext, ClusterEventType.CustomizedViewChange,
          HelixConstants.ChangeType.CUSTOMIZED_VIEW);
    } else {
      logger.warn(
          "RoutingTableProvider does not use CurrentStates as source, ignore CurrentState changes!");
    }
  }

  @Override
  @PreFetch(enabled = false)
  public void onCustomizedViewRootChange(List<String> customizedViewTypes,
      NotificationContext changeContext) {
    logger.info(
        "Registering the CustomizedView listeners again due to the CustomizedView root change.");
    List<String> userRequestedTypes =
        _sourceDataTypeMap.getOrDefault(PropertyType.CUSTOMIZEDVIEW, Collections.emptyList());
    for (String customizedStateType : userRequestedTypes) {
      try {
        _helixManager.addCustomizedViewChangeListener(this, customizedStateType);
      } catch (Exception e) {
        shutdown();
        throw new HelixException(
            String.format("Failed to attach CustomizedView Listener to HelixManager for type %s!",
                customizedStateType),
            e);
      }
    }
  }

  final AtomicReference<Map<String, LiveInstance>> _lastSeenSessions = new AtomicReference<>();

  /**
   * Go through all live instances in the cluster, add CurrentStateChange listener to
   * them if they are newly added, and remove CurrentStateChange listener if instance is offline.
   */
  private void updateCurrentStatesListeners(List<LiveInstance> liveInstances,
      NotificationContext changeContext) {
    HelixManager manager = changeContext.getManager();
    PropertyKey.Builder keyBuilder = new PropertyKey.Builder(manager.getClusterName());

    if (changeContext.getType() == NotificationContext.Type.FINALIZE) {
      // on finalize, should remove all current-state listeners
      logger.info("remove current-state listeners. lastSeenSessions: {}", _lastSeenSessions.get());
      liveInstances = Collections.emptyList();
    }

    Map<String, LiveInstance> curSessions = liveInstances.stream().collect(
        Collectors.toConcurrentMap(LiveInstance::getEphemeralOwner, liveInstance -> liveInstance));

    // Add listeners to new live instances
    for (String session : curSessions.keySet()) {
      String instanceName = curSessions.get(session).getInstanceName();
      try {
        // Add current-state listeners for new sessions
        // It is possible this loop attempt to re-add listeners. But the HelixManager will skip the
        // operation if a listener has been already registered.
        manager.addCurrentStateChangeListener(this, instanceName, session);
        logger.info("{} added current-state listener for instance: {}, session: {}, listener: {}",
            manager.getInstanceName(), instanceName, session, this);
      } catch (Exception e) {
        logger.error("Fail to add current state listener for instance: {} with session: {}",
            instanceName, session, e);
      }
    }

    // The getAndSet ensures that all the records in _lastSeenSessions map are active listeners.
    // And the following logic ensures that all the exchanged last seen listeners are cleaned up.
    Map<String, LiveInstance> lastSessions = _lastSeenSessions.getAndSet(curSessions);
    if (lastSessions == null) {
      lastSessions = Collections.emptyMap();
    }

    // remove current-state listener for expired session
    for (String session : lastSessions.keySet()) {
      if (!curSessions.containsKey(session)) {
        String instanceName = lastSessions.get(session).getInstanceName();
        try {
          manager.removeListener(keyBuilder.currentStates(instanceName, session), this);
          logger.info("remove current-state listener for instance: {}, session: {}", instanceName,
              session);
        } catch (Exception e) {
          logger.error("Fail to remove current state listener for instance: {} with session: {}",
              instanceName, session, e);
        }
      }
    }
  }

  private void reset() {
    logger.info("Resetting the routing table.");
    RoutingTable newRoutingTable;
    for (String key: _routingTableRefMap.keySet()) {
      PropertyType propertyType = _routingTableRefMap.get(key).get().getPropertyType();
      if (propertyType == PropertyType.CUSTOMIZEDVIEW) {
        String stateType = _routingTableRefMap.get(key).get().getStateType();
        newRoutingTable = new CustomizedViewRoutingTable(propertyType, stateType);
      } else {
        newRoutingTable = new RoutingTable(propertyType);
      }
      _routingTableRefMap.get(key).set(newRoutingTable);
    }
  }

  protected void refreshExternalView(Collection<ExternalView> externalViews,
      Collection<InstanceConfig> instanceConfigs, Collection<LiveInstance> liveInstances,
      String referenceKey) {
    long startTime = System.currentTimeMillis();
    PropertyType propertyType = _routingTableRefMap.get(referenceKey).get().getPropertyType();
    RoutingTable newRoutingTable =
        new RoutingTable(externalViews, instanceConfigs, liveInstances, propertyType);
    resetRoutingTableAndNotify(startTime, newRoutingTable, referenceKey);
  }

  protected void refreshCustomizedView(Collection<CustomizedView> customizedViews,
      Collection<InstanceConfig> instanceConfigs, Collection<LiveInstance> liveInstances,
      String referenceKey) {
    long startTime = System.currentTimeMillis();
    PropertyType propertyType = _routingTableRefMap.get(referenceKey).get().getPropertyType();
    String customizedStateType = _routingTableRefMap.get(referenceKey).get().getStateType();
    RoutingTable newRoutingTable = new CustomizedViewRoutingTable(customizedViews, instanceConfigs,
        liveInstances, propertyType, customizedStateType);
    resetRoutingTableAndNotify(startTime, newRoutingTable, referenceKey);
  }

  protected void refreshCurrentState(Map<String, Map<String, Map<String, CurrentState>>> currentStateMap,
      Collection<InstanceConfig> instanceConfigs, Collection<LiveInstance> liveInstances,
      String referenceKey) {
    long startTime = System.currentTimeMillis();
    RoutingTable newRoutingTable =
        new RoutingTable(currentStateMap, instanceConfigs, liveInstances);
    resetRoutingTableAndNotify(startTime, newRoutingTable, referenceKey);
  }

  private void resetRoutingTableAndNotify(long startTime, RoutingTable newRoutingTable, String referenceKey) {
    _routingTableRefMap.get(referenceKey).set(newRoutingTable);
    String clusterName = _helixManager != null ? _helixManager.getClusterName() : null;
    logger.info("Refreshed the RoutingTable for cluster {}, took {} ms.", clusterName,
        (System.currentTimeMillis() - startTime));

    // TODO: move the callback user code logic to separate thread upon routing table statePropagation latency
    // integration test result. If the latency is more than 2 secs, we need to change this part.
    notifyRoutingTableChange(clusterName, referenceKey);

    // Update timestamp for last refresh
    if (_isPeriodicRefreshEnabled) {
      _lastRefreshTimestamp = System.currentTimeMillis();
    }
  }

  private void notifyRoutingTableChange(String clusterName, String referenceKey) {
    // This call back is called in the main event queue of RoutingTableProvider. We add log to
    // record time spent
    // here. Potentially, we should call this callback in a separate thread if this is a bottleneck.
    long startTime = System.currentTimeMillis();
    for (Map.Entry<RoutingTableChangeListener, ListenerContext> entry : _routingTableChangeListenerMap
        .entrySet()) {
      entry.getKey().onRoutingTableChange(
          new RoutingTableSnapshot(_routingTableRefMap.get(referenceKey).get()),
          entry.getValue().getContext());
    }
    logger.info("RoutingTableProvider user callback time for cluster {}, took {} ms.", clusterName,
        (System.currentTimeMillis() - startTime));
  }

  private class RouterUpdater extends ClusterEventProcessor {
    private final RoutingDataCache _dataCache;
    private final Map<PropertyType, List<String>> _sourceDataTypeMap;

    public RouterUpdater(String clusterName, Map<PropertyType, List<String>> sourceDataTypeMap) {
      super(clusterName, "Helix-RouterUpdater-event_process");
      _sourceDataTypeMap = sourceDataTypeMap;
      _dataCache = new RoutingDataCache(clusterName, _sourceDataTypeMap);
    }

    @Override
    protected void handleEvent(ClusterEvent event) {
      NotificationContext changeContext = event.getAttribute(AttributeName.changeContext.name());
      HelixConstants.ChangeType changeType = changeContext.getChangeType();

      // Set cluster event id for later processing methods and it also helps debug.
      _dataCache.setClusterEventId(event.getEventId());

      if (changeContext == null || changeContext.getType() != NotificationContext.Type.CALLBACK) {
        _dataCache.requireFullRefresh();
      } else {
        _dataCache.notifyDataChange(changeType, changeContext.getPathChanged());
      }

      // session has expired clean up the routing table
      if (changeContext.getType() == NotificationContext.Type.FINALIZE) {
        reset();
      } else {
        // refresh routing table.
        HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
        if (manager == null) {
          logger.error(String.format("HelixManager is null for router update event: %s", event));
          throw new HelixException("HelixManager is null for router update event.");
        }
        if (!manager.isConnected()) {
          logger.error(String.format("HelixManager is not connected for router update event: %s", event));
          throw new HelixException("HelixManager is not connected for router update event.");
        }

        long startTime = System.currentTimeMillis();

        _dataCache.refresh(manager.getHelixDataAccessor());
        for (PropertyType propertyType : _sourceDataTypeMap.keySet()) {
          switch (propertyType) {
          case EXTERNALVIEW: {
            String keyReference = generateReferenceKey(propertyType.name(), DEFAULT_STATE_TYPE);
            refreshExternalView(_dataCache.getExternalViews().values(),
                _dataCache.getRoutableInstanceConfigMap().values(),
                _dataCache.getRoutableLiveInstances().values(),
                keyReference);
          }
            break;
          case TARGETEXTERNALVIEW: {
            String keyReference = generateReferenceKey(propertyType.name(), DEFAULT_STATE_TYPE);
            refreshExternalView(_dataCache.getTargetExternalViews().values(),
                _dataCache.getRoutableInstanceConfigMap().values(),
                _dataCache.getRoutableLiveInstances().values(),
                keyReference);
          }
              break;
            case CUSTOMIZEDVIEW:
              for (String customizedStateType : _sourceDataTypeMap.getOrDefault(PropertyType.CUSTOMIZEDVIEW, Collections.emptyList())) {
                String keyReference = generateReferenceKey(propertyType.name(),  customizedStateType);
                refreshCustomizedView(_dataCache.getCustomizedView(customizedStateType).values(),
                    _dataCache.getRoutableInstanceConfigMap().values(),
                    _dataCache.getRoutableLiveInstances().values(), keyReference);
              }
              break;
            case CURRENTSTATES: {
              String keyReference = generateReferenceKey(propertyType.name(),  DEFAULT_STATE_TYPE);;
              refreshCurrentState(_dataCache.getCurrentStatesMap(),
                  _dataCache.getRoutableInstanceConfigMap().values(),
                  _dataCache.getRoutableLiveInstances().values(), keyReference);
              recordPropagationLatency(System.currentTimeMillis(), _dataCache.getCurrentStateSnapshot());
            }
              break;
            default:
              logger.warn("Unsupported source data type: {}, stop refreshing the routing table!", propertyType);
          }

          _monitorMap.get(propertyType).increaseDataRefreshCounters(startTime);
        }
      }
    }

    /**
     * Report current state to routing table propagation latency
     * This method is not threadsafe. Take care of _reportingTask atomicity if use in multi-threads.
     */
    private void recordPropagationLatency(final long currentTime,
        final CurrentStateSnapshot currentStateSnapshot) {
      // Note that due to the extra mem footprint introduced by currentStateSnapshot ref, we
      // restrict running report task count to be 1.
      // Any parallel tasks will be skipped. So the reporting metric data is sampled.
      if (_reportingTask == null || _reportingTask.isDone()) {
        _reportingTask = _reportExecutor.submit(new Callable<Object>() {
          @Override
          public Object call() {
            // getNewCurrentStateEndTimes() needs to iterate all current states. Make it async to
            // avoid performance impact.
            Map<PropertyKey, Map<String, Long>> currentStateEndTimeMap =
                currentStateSnapshot.getNewCurrentStateEndTimes();
            for (PropertyKey key : currentStateEndTimeMap.keySet()) {
              Map<String, Long> partitionStateEndTimes = currentStateEndTimeMap.get(key);
              for (String partition : partitionStateEndTimes.keySet()) {
                long endTime = partitionStateEndTimes.get(partition);
                if (currentTime >= endTime) {
                  for (PropertyType propertyType : _sourceDataTypeMap.keySet()) {
                    _monitorMap.get(propertyType).recordStatePropagationLatency(currentTime - endTime);
                    logger.debug(
                        "CurrentState updated in the routing table. Node Key {}, Partition {}, end time {}, Propagation latency {}",
                        key.toString(), partition, endTime, currentTime - endTime);
                  }
                } else {
                  // Verbose log in case currentTime < endTime. This could be the case that Router
                  // clock is slower than the participant clock.
                  logger.trace(
                      "CurrentState updated in the routing table. Node Key {}, Partition {}, end time {}, Propagation latency {}",
                      key.toString(), partition, endTime, currentTime - endTime);
                }
              }
            }
            return null;
          }
        });
      }
    }


    public void queueEvent(NotificationContext context, ClusterEventType eventType,
        HelixConstants.ChangeType changeType) {
      ClusterEvent event = new ClusterEvent(_clusterName, eventType);

      // Null check for manager in the following line is done in handleEvent()
      event.addAttribute(AttributeName.helixmanager.name(), context.getManager());
      event.addAttribute(AttributeName.changeContext.name(), context);
      queueEvent(event);
      // TODO: Split the monitor into. One is general router callback tracking. The other one is for
      // each type of tracking.
      // TODO: We may need to add more complexity to the customized view monitor for each state
      // type.
      for (PropertyType propertyType : _monitorMap.keySet()) {
        _monitorMap.get(propertyType).increaseCallbackCounters(_eventQueue.size());
      }
    }
  }

  protected class ListenerContext {
    private Object _context;

    public ListenerContext(Object context) {
      _context = context;
    }

    public Object getContext() {
      return _context;
    }
  }
}
