/*
 * 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.
 */

package org.apache.druid.server.coordinator;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Ordering;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Inject;
import it.unimi.dsi.fastutil.objects.Object2IntMap;
import it.unimi.dsi.fastutil.objects.Object2IntMaps;
import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
import it.unimi.dsi.fastutil.objects.Object2LongMap;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.utils.ZKPaths;
import org.apache.druid.client.DataSourcesSnapshot;
import org.apache.druid.client.DruidDataSource;
import org.apache.druid.client.DruidServer;
import org.apache.druid.client.ImmutableDruidDataSource;
import org.apache.druid.client.ImmutableDruidServer;
import org.apache.druid.client.ServerInventoryView;
import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.indexing.IndexingServiceClient;
import org.apache.druid.common.config.JacksonConfigManager;
import org.apache.druid.curator.discovery.ServiceAnnouncer;
import org.apache.druid.discovery.DruidLeaderSelector;
import org.apache.druid.guice.ManageLifecycle;
import org.apache.druid.guice.annotations.CoordinatorIndexingServiceDuty;
import org.apache.druid.guice.annotations.Self;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
import org.apache.druid.java.util.common.guava.Comparators;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.metadata.MetadataRuleManager;
import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordinator.duty.BalanceSegments;
import org.apache.druid.server.coordinator.duty.CompactSegments;
import org.apache.druid.server.coordinator.duty.CoordinatorDuty;
import org.apache.druid.server.coordinator.duty.EmitClusterStatsAndMetrics;
import org.apache.druid.server.coordinator.duty.LogUsedSegments;
import org.apache.druid.server.coordinator.duty.MarkAsUnusedOvershadowedSegments;
import org.apache.druid.server.coordinator.duty.RunRules;
import org.apache.druid.server.coordinator.duty.UnloadUnusedSegments;
import org.apache.druid.server.coordinator.rules.LoadRule;
import org.apache.druid.server.coordinator.rules.Rule;
import org.apache.druid.server.initialization.ZkPathsConfig;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManager;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
import org.joda.time.DateTime;
import org.joda.time.Duration;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
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.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;

/**
 *
 */
@ManageLifecycle
public class DruidCoordinator
{
  /**
   * This comparator orders "freshest" segments first, i. e. segments with most recent intervals.
   *
   * It is used in historical nodes' {@link LoadQueuePeon}s to make historicals load more recent segment first.
   *
   * It is also used in {@link DruidCoordinatorRuntimeParams} for {@link
   * DruidCoordinatorRuntimeParams#getUsedSegments()} - a collection of segments to be considered during some
   * coordinator run for different {@link CoordinatorDuty}s. The order matters only for {@link
   * RunRules}, which tries to apply the rules while iterating the segments in the order imposed by
   * this comparator. In {@link LoadRule} the throttling limit may be hit (via {@link ReplicationThrottler}; see
   * {@link CoordinatorDynamicConfig#getReplicationThrottleLimit()}). So before we potentially hit this limit, we want
   * to schedule loading the more recent segments (among all of those that need to be loaded).
   *
   * In both {@link LoadQueuePeon}s and {@link RunRules}, we want to load more recent segments first
   * because presumably they are queried more often and contain are more important data for users, so if the Druid
   * cluster has availability problems and struggling to make all segments available immediately, at least we try to
   * make more "important" (more recent) segments available as soon as possible.
   */
  static final Comparator<DataSegment> SEGMENT_COMPARATOR_RECENT_FIRST = Ordering
      .from(Comparators.intervalsByEndThenStart())
      .onResultOf(DataSegment::getInterval)
      .compound(Ordering.<DataSegment>natural())
      .reverse();

  private static final EmittingLogger log = new EmittingLogger(DruidCoordinator.class);

  private final Object lock = new Object();
  private final DruidCoordinatorConfig config;
  private final ZkPathsConfig zkPaths;
  private final JacksonConfigManager configManager;
  private final SegmentsMetadataManager segmentsMetadataManager;
  private final ServerInventoryView serverInventoryView;
  private final MetadataRuleManager metadataRuleManager;
  private final CuratorFramework curator;
  private final ServiceEmitter emitter;
  private final IndexingServiceClient indexingServiceClient;
  private final ScheduledExecutorService exec;
  private final LoadQueueTaskMaster taskMaster;
  private final Map<String, LoadQueuePeon> loadManagementPeons;
  private final ServiceAnnouncer serviceAnnouncer;
  private final DruidNode self;
  private final Set<CoordinatorDuty> indexingServiceDuties;
  private final BalancerStrategyFactory factory;
  private final LookupCoordinatorManager lookupCoordinatorManager;
  private final DruidLeaderSelector coordLeaderSelector;

  private final CompactSegments compactSegments;

  private volatile boolean started = false;
  private volatile SegmentReplicantLookup segmentReplicantLookup = null;

  private int cachedBalancerThreadNumber;
  private ListeningExecutorService balancerExec;

  @Inject
  public DruidCoordinator(
      DruidCoordinatorConfig config,
      ZkPathsConfig zkPaths,
      JacksonConfigManager configManager,
      SegmentsMetadataManager segmentsMetadataManager,
      ServerInventoryView serverInventoryView,
      MetadataRuleManager metadataRuleManager,
      CuratorFramework curator,
      ServiceEmitter emitter,
      ScheduledExecutorFactory scheduledExecutorFactory,
      IndexingServiceClient indexingServiceClient,
      LoadQueueTaskMaster taskMaster,
      ServiceAnnouncer serviceAnnouncer,
      @Self DruidNode self,
      @CoordinatorIndexingServiceDuty Set<CoordinatorDuty> indexingServiceDuties,
      BalancerStrategyFactory factory,
      LookupCoordinatorManager lookupCoordinatorManager,
      @Coordinator DruidLeaderSelector coordLeaderSelector,
      CompactSegments compactSegments
  )
  {
    this(
        config,
        zkPaths,
        configManager,
        segmentsMetadataManager,
        serverInventoryView,
        metadataRuleManager,
        curator,
        emitter,
        scheduledExecutorFactory,
        indexingServiceClient,
        taskMaster,
        serviceAnnouncer,
        self,
        new ConcurrentHashMap<>(),
        indexingServiceDuties,
        factory,
        lookupCoordinatorManager,
        coordLeaderSelector,
        compactSegments
    );
  }

  DruidCoordinator(
      DruidCoordinatorConfig config,
      ZkPathsConfig zkPaths,
      JacksonConfigManager configManager,
      SegmentsMetadataManager segmentsMetadataManager,
      ServerInventoryView serverInventoryView,
      MetadataRuleManager metadataRuleManager,
      CuratorFramework curator,
      ServiceEmitter emitter,
      ScheduledExecutorFactory scheduledExecutorFactory,
      IndexingServiceClient indexingServiceClient,
      LoadQueueTaskMaster taskMaster,
      ServiceAnnouncer serviceAnnouncer,
      DruidNode self,
      ConcurrentMap<String, LoadQueuePeon> loadQueuePeonMap,
      Set<CoordinatorDuty> indexingServiceDuties,
      BalancerStrategyFactory factory,
      LookupCoordinatorManager lookupCoordinatorManager,
      DruidLeaderSelector coordLeaderSelector,
      CompactSegments compactSegments
  )
  {
    this.config = config;
    this.zkPaths = zkPaths;
    this.configManager = configManager;

    this.segmentsMetadataManager = segmentsMetadataManager;
    this.serverInventoryView = serverInventoryView;
    this.metadataRuleManager = metadataRuleManager;
    this.curator = curator;
    this.emitter = emitter;
    this.indexingServiceClient = indexingServiceClient;
    this.taskMaster = taskMaster;
    this.serviceAnnouncer = serviceAnnouncer;
    this.self = self;
    this.indexingServiceDuties = indexingServiceDuties;

    this.exec = scheduledExecutorFactory.create(1, "Coordinator-Exec--%d");

    this.loadManagementPeons = loadQueuePeonMap;
    this.factory = factory;
    this.lookupCoordinatorManager = lookupCoordinatorManager;
    this.coordLeaderSelector = coordLeaderSelector;

    this.compactSegments = compactSegments;
  }

  public boolean isLeader()
  {
    return coordLeaderSelector.isLeader();
  }

  public Map<String, LoadQueuePeon> getLoadManagementPeons()
  {
    return loadManagementPeons;
  }

  /**
   * @return tier -> { dataSource -> underReplicationCount } map
   */
  public Map<String, Object2LongMap<String>> computeUnderReplicationCountsPerDataSourcePerTier()
  {
    final Iterable<DataSegment> dataSegments = segmentsMetadataManager.iterateAllUsedSegments();
    return computeUnderReplicationCountsPerDataSourcePerTierForSegments(dataSegments);
  }

  /**
   * segmentReplicantLookup use in this method could potentially be stale since it is only updated on coordinator runs.
   * However, this is ok as long as the {@param dataSegments} is refreshed/latest as this would at least still ensure
   * that the stale data in segmentReplicantLookup would be under counting replication levels,
   * rather than potentially falsely reporting that everything is available.
   *
   * @return tier -> { dataSource -> underReplicationCount } map
   */
  public Map<String, Object2LongMap<String>> computeUnderReplicationCountsPerDataSourcePerTierForSegments(
      Iterable<DataSegment> dataSegments
  )
  {
    final Map<String, Object2LongMap<String>> underReplicationCountsPerDataSourcePerTier = new HashMap<>();

    if (segmentReplicantLookup == null) {
      return underReplicationCountsPerDataSourcePerTier;
    }

    final DateTime now = DateTimes.nowUtc();

    for (final DataSegment segment : dataSegments) {
      final List<Rule> rules = metadataRuleManager.getRulesWithDefault(segment.getDataSource());

      for (final Rule rule : rules) {
        if (!rule.appliesTo(segment, now)) {
          // Rule did not match. Continue to the next Rule.
          continue;
        }
        if (!rule.canLoadSegments()) {
          // Rule matched but rule does not and cannot load segments.
          // Hence, there is no need to update underReplicationCountsPerDataSourcePerTier map
          break;
        }

        rule.updateUnderReplicated(underReplicationCountsPerDataSourcePerTier, segmentReplicantLookup, segment);

        // Only the first matching rule applies. This is because the Coordinator cycle through all used segments
        // and match each segment with the first rule that applies. Each segment may only match a single rule.
        break;
      }
    }

    return underReplicationCountsPerDataSourcePerTier;
  }

  public Object2IntMap<String> computeNumsUnavailableUsedSegmentsPerDataSource()
  {
    if (segmentReplicantLookup == null) {
      return Object2IntMaps.emptyMap();
    }

    final Object2IntOpenHashMap<String> numsUnavailableUsedSegmentsPerDataSource = new Object2IntOpenHashMap<>();

    final Iterable<DataSegment> dataSegments = segmentsMetadataManager.iterateAllUsedSegments();

    for (DataSegment segment : dataSegments) {
      if (segmentReplicantLookup.getLoadedReplicants(segment.getId()) == 0) {
        numsUnavailableUsedSegmentsPerDataSource.addTo(segment.getDataSource(), 1);
      } else {
        numsUnavailableUsedSegmentsPerDataSource.addTo(segment.getDataSource(), 0);
      }
    }

    return numsUnavailableUsedSegmentsPerDataSource;
  }

  public Map<String, Double> getLoadStatus()
  {
    final Map<String, Double> loadStatus = new HashMap<>();
    final Collection<ImmutableDruidDataSource> dataSources =
        segmentsMetadataManager.getImmutableDataSourcesWithAllUsedSegments();

    for (ImmutableDruidDataSource dataSource : dataSources) {
      final Set<DataSegment> segments = Sets.newHashSet(dataSource.getSegments());
      final int numPublishedSegments = segments.size();

      // remove loaded segments
      for (DruidServer druidServer : serverInventoryView.getInventory()) {
        final DruidDataSource loadedView = druidServer.getDataSource(dataSource.getName());
        if (loadedView != null) {
          // This does not use segments.removeAll(loadedView.getSegments()) for performance reasons.
          // Please see https://github.com/apache/druid/pull/5632 and LoadStatusBenchmark for more info.
          for (DataSegment serverSegment : loadedView.getSegments()) {
            segments.remove(serverSegment);
          }
        }
      }
      final int numUnavailableSegments = segments.size();
      loadStatus.put(
          dataSource.getName(),
          100 * ((double) (numPublishedSegments - numUnavailableSegments) / (double) numPublishedSegments)
      );
    }

    return loadStatus;
  }

  @Nullable
  public Long getTotalSizeOfSegmentsAwaitingCompaction(String dataSource)
  {
    return compactSegments.getTotalSizeOfSegmentsAwaitingCompaction(dataSource);
  }

  @Nullable
  public AutoCompactionSnapshot getAutoCompactionSnapshotForDataSource(String dataSource)
  {
    return compactSegments.getAutoCompactionSnapshot(dataSource);
  }

  public Map<String, AutoCompactionSnapshot> getAutoCompactionSnapshot()
  {
    return compactSegments.getAutoCompactionSnapshot();
  }

  public CoordinatorDynamicConfig getDynamicConfigs()
  {
    return CoordinatorDynamicConfig.current(configManager);
  }

  public CoordinatorCompactionConfig getCompactionConfig()
  {
    return CoordinatorCompactionConfig.current(configManager);
  }

  public void markSegmentAsUnused(DataSegment segment)
  {
    log.debug("Marking segment[%s] as unused", segment.getId());
    segmentsMetadataManager.markSegmentAsUnused(segment.getId().toString());
  }

  public String getCurrentLeader()
  {
    return coordLeaderSelector.getCurrentLeader();
  }

  public void moveSegment(
      DruidCoordinatorRuntimeParams params,
      ImmutableDruidServer fromServer,
      ImmutableDruidServer toServer,
      DataSegment segment,
      final LoadPeonCallback callback
  )
  {
    if (segment == null) {
      log.makeAlert(new IAE("Can not move null DataSegment"), "Exception moving null segment").emit();
      if (callback != null) {
        callback.execute();
      }
      throw new ISE("Cannot move null DataSegment");
    }
    SegmentId segmentId = segment.getId();
    try {
      if (fromServer.getMetadata().equals(toServer.getMetadata())) {
        throw new IAE("Cannot move [%s] to and from the same server [%s]", segmentId, fromServer.getName());
      }

      ImmutableDruidDataSource dataSource = params.getDataSourcesSnapshot().getDataSource(segment.getDataSource());
      if (dataSource == null) {
        throw new IAE("Unable to find dataSource for segment [%s] in metadata", segmentId);
      }

      // get segment information from SegmentsMetadataManager instead of getting it from fromServer's.
      // This is useful when SegmentsMetadataManager and fromServer DataSegment's are different for same
      // identifier (say loadSpec differs because of deep storage migration).
      final DataSegment segmentToLoad = dataSource.getSegment(segment.getId());
      if (segmentToLoad == null) {
        throw new IAE("No segment metadata found for segment Id [%s]", segment.getId());
      }
      final LoadQueuePeon loadPeon = loadManagementPeons.get(toServer.getName());
      if (loadPeon == null) {
        throw new IAE("LoadQueuePeon hasn't been created yet for path [%s]", toServer.getName());
      }

      final LoadQueuePeon dropPeon = loadManagementPeons.get(fromServer.getName());
      if (dropPeon == null) {
        throw new IAE("LoadQueuePeon hasn't been created yet for path [%s]", fromServer.getName());
      }

      final ServerHolder toHolder = new ServerHolder(toServer, loadPeon);
      if (toHolder.getAvailableSize() < segmentToLoad.getSize()) {
        throw new IAE(
            "Not enough capacity on server [%s] for segment [%s]. Required: %,d, available: %,d.",
            toServer.getName(),
            segmentToLoad,
            segmentToLoad.getSize(),
            toHolder.getAvailableSize()
        );
      }

      final String toLoadQueueSegPath =
          ZKPaths.makePath(zkPaths.getLoadQueuePath(), toServer.getName(), segmentId.toString());

      final LoadPeonCallback loadPeonCallback = () -> {
        dropPeon.unmarkSegmentToDrop(segmentToLoad);
        if (callback != null) {
          callback.execute();
        }
      };

      // mark segment to drop before it is actually loaded on server
      // to be able to account this information in DruidBalancerStrategy immediately
      dropPeon.markSegmentToDrop(segmentToLoad);
      try {
        loadPeon.loadSegment(
            segmentToLoad,
            () -> {
              try {
                if (serverInventoryView.isSegmentLoadedByServer(toServer.getName(), segment) &&
                    curator.checkExists().forPath(toLoadQueueSegPath) == null &&
                    !dropPeon.getSegmentsToDrop().contains(segment)) {
                  dropPeon.dropSegment(segment, loadPeonCallback);
                } else {
                  loadPeonCallback.execute();
                }
              }
              catch (Exception e) {
                throw new RuntimeException(e);
              }
            }
        );
      }
      catch (Exception e) {
        dropPeon.unmarkSegmentToDrop(segmentToLoad);
        throw new RuntimeException(e);
      }
    }
    catch (Exception e) {
      log.makeAlert(e, "Exception moving segment %s", segmentId).emit();
      if (callback != null) {
        callback.execute();
      }
    }
  }

  @VisibleForTesting
  public int getCachedBalancerThreadNumber()
  {
    return cachedBalancerThreadNumber;
  }
  
  @VisibleForTesting
  public ListeningExecutorService getBalancerExec()
  {
    return balancerExec;
  }

  @LifecycleStart
  public void start()
  {
    synchronized (lock) {
      if (started) {
        return;
      }
      started = true;

      coordLeaderSelector.registerListener(
          new DruidLeaderSelector.Listener()
          {
            @Override
            public void becomeLeader()
            {
              DruidCoordinator.this.becomeLeader();
            }

            @Override
            public void stopBeingLeader()
            {
              DruidCoordinator.this.stopBeingLeader();
            }
          }
      );
    }
  }

  @LifecycleStop
  public void stop()
  {
    synchronized (lock) {
      if (!started) {
        return;
      }

      coordLeaderSelector.unregisterListener();

      started = false;

      exec.shutdownNow();

      if (balancerExec != null) {
        balancerExec.shutdownNow();
      }
    }
  }

  public void runCompactSegmentsDuty()
  {
    final int startingLeaderCounter = coordLeaderSelector.localTerm();
    DutiesRunnable compactSegmentsDuty = new DutiesRunnable(makeCompactSegmentsDuty(), startingLeaderCounter);
    compactSegmentsDuty.run();
  }

  private void becomeLeader()
  {
    synchronized (lock) {
      if (!started) {
        return;
      }

      log.info(
          "I am the leader of the coordinators, all must bow! Starting coordination in [%s].",
          config.getCoordinatorStartDelay()
      );

      segmentsMetadataManager.startPollingDatabasePeriodically();
      metadataRuleManager.start();
      lookupCoordinatorManager.start();
      serviceAnnouncer.announce(self);
      final int startingLeaderCounter = coordLeaderSelector.localTerm();

      final List<Pair<? extends DutiesRunnable, Duration>> dutiesRunnables = new ArrayList<>();
      dutiesRunnables.add(
          Pair.of(
              new DutiesRunnable(makeHistoricalManagementDuties(), startingLeaderCounter),
              config.getCoordinatorPeriod()
          )
      );
      if (indexingServiceClient != null) {
        dutiesRunnables.add(
            Pair.of(
                new DutiesRunnable(makeIndexingServiceDuties(), startingLeaderCounter),
                config.getCoordinatorIndexingPeriod()
            )
        );
      }

      for (final Pair<? extends DutiesRunnable, Duration> dutiesRunnable : dutiesRunnables) {
        // CompactSegmentsDuty can takes a non trival amount of time to complete.
        // Hence, we schedule at fixed rate to make sure the other tasks still run at approximately every
        // config.getCoordinatorIndexingPeriod() period. Note that cautious should be taken
        // if setting config.getCoordinatorIndexingPeriod() lower than the default value.
        ScheduledExecutors.scheduleAtFixedRate(
            exec,
            config.getCoordinatorStartDelay(),
            dutiesRunnable.rhs,
            new Callable<ScheduledExecutors.Signal>()
            {
              private final DutiesRunnable theRunnable = dutiesRunnable.lhs;

              @Override
              public ScheduledExecutors.Signal call()
              {
                if (coordLeaderSelector.isLeader() && startingLeaderCounter == coordLeaderSelector.localTerm()) {
                  theRunnable.run();
                }
                if (coordLeaderSelector.isLeader()
                    && startingLeaderCounter == coordLeaderSelector.localTerm()) { // (We might no longer be leader)
                  return ScheduledExecutors.Signal.REPEAT;
                } else {
                  return ScheduledExecutors.Signal.STOP;
                }
              }
            }
        );
      }
    }
  }

  private void stopBeingLeader()
  {
    synchronized (lock) {

      log.info("I am no longer the leader...");

      for (String server : loadManagementPeons.keySet()) {
        LoadQueuePeon peon = loadManagementPeons.remove(server);
        peon.stop();
      }
      loadManagementPeons.clear();

      serviceAnnouncer.unannounce(self);
      lookupCoordinatorManager.stop();
      metadataRuleManager.stop();
      segmentsMetadataManager.stopPollingDatabasePeriodically();

      if (balancerExec != null) {
        balancerExec.shutdownNow();
        balancerExec = null;
      }
    }
  }

  private List<CoordinatorDuty> makeHistoricalManagementDuties()
  {
    return ImmutableList.of(
        new LogUsedSegments(),
        new UpdateCoordinatorStateAndPrepareCluster(),
        new RunRules(DruidCoordinator.this),
        new UnloadUnusedSegments(),
        new MarkAsUnusedOvershadowedSegments(DruidCoordinator.this),
        new BalanceSegments(DruidCoordinator.this),
        new EmitClusterStatsAndMetrics(DruidCoordinator.this)
    );
  }

  private List<CoordinatorDuty> makeIndexingServiceDuties()
  {
    List<CoordinatorDuty> duties = new ArrayList<>();
    duties.add(new LogUsedSegments());
    duties.addAll(indexingServiceDuties);
    // CompactSegmentsDuty should be the last duty as it can take a long time to complete
    duties.addAll(makeCompactSegmentsDuty());

    log.debug(
        "Done making indexing service duties %s",
        duties.stream().map(duty -> duty.getClass().getName()).collect(Collectors.toList())
    );
    return ImmutableList.copyOf(duties);
  }

  private List<CoordinatorDuty> makeCompactSegmentsDuty()
  {
    return ImmutableList.of(compactSegments);
  }

  @VisibleForTesting
  protected class DutiesRunnable implements Runnable
  {
    private final long startTimeNanos = System.nanoTime();
    private final List<CoordinatorDuty> duties;
    private final int startingLeaderCounter;

    protected DutiesRunnable(List<CoordinatorDuty> duties, final int startingLeaderCounter)
    {
      this.duties = duties;
      this.startingLeaderCounter = startingLeaderCounter;
    }

    @VisibleForTesting
    protected void initBalancerExecutor()
    {
      final int currentNumber = getDynamicConfigs().getBalancerComputeThreads();
      final String threadNameFormat = "coordinator-cost-balancer-%s";
      // fist time initialization
      if (balancerExec == null) {
        balancerExec = MoreExecutors.listeningDecorator(Execs.multiThreaded(
            currentNumber,
            threadNameFormat
        ));
        cachedBalancerThreadNumber = currentNumber;
        return;
      }

      if (cachedBalancerThreadNumber != currentNumber) {
        log.info(
            "balancerComputeThreads has been changed from [%s] to [%s], recreating the thread pool.",
            cachedBalancerThreadNumber,
            currentNumber
        );
        balancerExec.shutdownNow();
        balancerExec = MoreExecutors.listeningDecorator(Execs.multiThreaded(
            currentNumber,
            threadNameFormat
        ));
        cachedBalancerThreadNumber = currentNumber;
      }
    }

    @Override
    public void run()
    {
      try {
        synchronized (lock) {
          if (!coordLeaderSelector.isLeader()) {
            log.info("LEGGO MY EGGO. [%s] is leader.", coordLeaderSelector.getCurrentLeader());
            stopBeingLeader();
            return;
          }
        }

        List<Boolean> allStarted = Arrays.asList(
            segmentsMetadataManager.isPollingDatabasePeriodically(),
            serverInventoryView.isStarted()
        );
        for (Boolean aBoolean : allStarted) {
          if (!aBoolean) {
            log.error("InventoryManagers not started[%s]", allStarted);
            stopBeingLeader();
            return;
          }
        }

        initBalancerExecutor();
        BalancerStrategy balancerStrategy = factory.createBalancerStrategy(balancerExec);

        // Do coordinator stuff.
        DataSourcesSnapshot dataSourcesSnapshot = segmentsMetadataManager.getSnapshotOfDataSourcesWithAllUsedSegments();

        DruidCoordinatorRuntimeParams params =
            DruidCoordinatorRuntimeParams
                .newBuilder()
                .withDatabaseRuleManager(metadataRuleManager)
                .withStartTimeNanos(startTimeNanos)
                .withSnapshotOfDataSourcesWithAllUsedSegments(dataSourcesSnapshot)
                .withDynamicConfigs(getDynamicConfigs())
                .withCompactionConfig(getCompactionConfig())
                .withEmitter(emitter)
                .withBalancerStrategy(balancerStrategy)
                .build();

        boolean coordinationPaused = getDynamicConfigs().getPauseCoordination();
        if (coordinationPaused
            && coordLeaderSelector.isLeader()
            && startingLeaderCounter == coordLeaderSelector.localTerm()) {

          log.debug(
              "Coordination is paused via dynamic configs! I will not be running Coordination Duties at this time"
          );
        }

        for (CoordinatorDuty duty : duties) {
          // Don't read state and run state in the same duty otherwise racy conditions may exist
          if (!coordinationPaused
              && coordLeaderSelector.isLeader()
              && startingLeaderCounter == coordLeaderSelector.localTerm()) {

            params = duty.run(params);

            if (params == null) {
              // This duty wanted to cancel the run. No log message, since the duty should have logged a reason.
              return;
            }
          }
        }
      }
      catch (Exception e) {
        log.makeAlert(e, "Caught exception, ignoring so that schedule keeps going.").emit();
      }
    }
  }

  /**
   * Updates the enclosing {@link DruidCoordinator}'s state and prepares an immutable view of the cluster state (which
   * consists of {@link DruidCluster} and {@link SegmentReplicantLookup}) and feeds it into {@link
   * DruidCoordinatorRuntimeParams} for use in subsequent {@link CoordinatorDuty}s (see the order in {@link
   * #makeHistoricalManagementDuties()}).
   */
  private class UpdateCoordinatorStateAndPrepareCluster implements CoordinatorDuty
  {
    @Nullable
    @Override
    public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
    {
      List<ImmutableDruidServer> currentServers = prepareCurrentServers();

      startPeonsForNewServers(currentServers);

      final DruidCluster cluster = prepareCluster(params, currentServers);
      segmentReplicantLookup = SegmentReplicantLookup.make(cluster);

      stopPeonsForDisappearedServers(currentServers);

      return params.buildFromExisting()
                   .withDruidCluster(cluster)
                   .withLoadManagementPeons(loadManagementPeons)
                   .withSegmentReplicantLookup(segmentReplicantLookup)
                   .build();
    }

    List<ImmutableDruidServer> prepareCurrentServers()
    {
      List<ImmutableDruidServer> currentServers = serverInventoryView
          .getInventory()
          .stream()
          .filter(DruidServer::isSegmentReplicationOrBroadcastTarget)
          .map(DruidServer::toImmutableDruidServer)
          .collect(Collectors.toList());

      if (log.isDebugEnabled()) {
        // Display info about all segment-replicatable (historical and bridge) servers
        log.debug("Servers");
        for (ImmutableDruidServer druidServer : currentServers) {
          log.debug("  %s", druidServer);
          log.debug("    -- DataSources");
          for (ImmutableDruidDataSource druidDataSource : druidServer.getDataSources()) {
            log.debug("    %s", druidDataSource);
          }
        }
      }
      return currentServers;
    }

    void startPeonsForNewServers(List<ImmutableDruidServer> currentServers)
    {
      for (ImmutableDruidServer server : currentServers) {
        loadManagementPeons.computeIfAbsent(server.getName(), serverName -> {
          LoadQueuePeon loadQueuePeon = taskMaster.giveMePeon(server);
          loadQueuePeon.start();
          log.debug("Created LoadQueuePeon for server[%s].", server.getName());
          return loadQueuePeon;
        });
      }
    }

    DruidCluster prepareCluster(DruidCoordinatorRuntimeParams params, List<ImmutableDruidServer> currentServers)
    {
      Set<String> decommissioningServers = params.getCoordinatorDynamicConfig().getDecommissioningNodes();
      final DruidCluster cluster = new DruidCluster();
      for (ImmutableDruidServer server : currentServers) {
        cluster.add(
            new ServerHolder(
                server,
                loadManagementPeons.get(server.getName()),
                decommissioningServers.contains(server.getHost())
            )
        );
      }
      return cluster;
    }

    void stopPeonsForDisappearedServers(List<ImmutableDruidServer> servers)
    {
      final Set<String> disappeared = Sets.newHashSet(loadManagementPeons.keySet());
      for (ImmutableDruidServer server : servers) {
        disappeared.remove(server.getName());
      }
      for (String name : disappeared) {
        log.debug("Removing listener for server[%s] which is no longer there.", name);
        LoadQueuePeon peon = loadManagementPeons.remove(name);
        peon.stop();
      }
    }
  }
}