/*
 * 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.lookup.cache;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.jaxrs.smile.SmileMediaTypes;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import com.google.common.net.HostAndPort;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableScheduledFuture;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Inject;
import org.apache.druid.audit.AuditInfo;
import org.apache.druid.common.config.JacksonConfigManager;
import org.apache.druid.concurrent.LifecycleLock;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.guice.annotations.EscalatedGlobal;
import org.apache.druid.guice.annotations.Smile;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.IOE;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StreamUtils;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.java.util.http.client.Request;
import org.apache.druid.java.util.http.client.response.ClientResponse;
import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
import org.apache.druid.java.util.http.client.response.SequenceInputStreamResponseHandler;
import org.apache.druid.query.lookup.LookupsState;
import org.apache.druid.server.http.HostAndPortWithScheme;
import org.apache.druid.server.listener.resource.ListenerResource;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpResponse;

import javax.annotation.Nullable;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Managers {@link org.apache.druid.query.lookup.LookupExtractorFactoryContainer} specifications, distributing them
 * to {@link org.apache.druid.query.lookup.LookupReferencesManager} around the cluster by monitoring the lookup
 * announce path for servers and utilizing their {@link org.apache.druid.query.lookup.LookupListeningResource} API
 * to load, drop, and update lookups around the cluster.
 */
public class LookupCoordinatorManager
{
  //key used in druid-0.10.0 with config manager
  public static final String OLD_LOOKUP_CONFIG_KEY = "lookups";

  public static final String LOOKUP_CONFIG_KEY = "lookupsConfig";
  public static final String LOOKUP_LISTEN_ANNOUNCE_KEY = "lookups";

  private static final String LOOKUP_BASE_REQUEST_PATH = ListenerResource.BASE_PATH
                                                         + "/"
                                                         + LOOKUP_LISTEN_ANNOUNCE_KEY;
  private static final String LOOKUP_UPDATE_REQUEST_PATH = LOOKUP_BASE_REQUEST_PATH + "/" + "updates";

  private static final TypeReference<LookupsState<LookupExtractorFactoryMapContainer>> LOOKUPS_STATE_TYPE_REFERENCE =
      new TypeReference<LookupsState<LookupExtractorFactoryMapContainer>>()
      {
      };

  private static final EmittingLogger LOG = new EmittingLogger(LookupCoordinatorManager.class);

  private final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider;
  private LookupNodeDiscovery lookupNodeDiscovery;

  private final JacksonConfigManager configManager;
  private final LookupCoordinatorManagerConfig lookupCoordinatorManagerConfig;
  private final LookupsCommunicator lookupsCommunicator;

  // Known lookup state across various cluster nodes is managed in the reference here. On each lookup management loop
  // state is rediscovered and updated in the ref here. If some lookup nodes have disappeared since last lookup
  // management loop, then they get discarded automatically.
  @VisibleForTesting
  final AtomicReference<Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>>> knownOldState =
      new AtomicReference<>(ImmutableMap.of());

  // Updated by config watching service
  private AtomicReference<Map<String, Map<String, LookupExtractorFactoryMapContainer>>> lookupMapConfigRef;

  private final LifecycleLock lifecycleLock = new LifecycleLock();

  private ListeningScheduledExecutorService executorService;
  private ListenableScheduledFuture<?> backgroundManagerFuture;
  private CountDownLatch backgroundManagerExitedLatch;

  @Inject
  public LookupCoordinatorManager(
      final @EscalatedGlobal HttpClient httpClient,
      final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider,
      final @Smile ObjectMapper smileMapper,
      final JacksonConfigManager configManager,
      final LookupCoordinatorManagerConfig lookupCoordinatorManagerConfig
  )
  {
    this(
        druidNodeDiscoveryProvider,
        configManager,
        lookupCoordinatorManagerConfig,
        new LookupsCommunicator(
            httpClient,
            lookupCoordinatorManagerConfig,
            smileMapper
        ),
        null
    );
  }

  @VisibleForTesting
  LookupCoordinatorManager(
      final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider,
      final JacksonConfigManager configManager,
      final LookupCoordinatorManagerConfig lookupCoordinatorManagerConfig,
      final LookupsCommunicator lookupsCommunicator,
      final LookupNodeDiscovery lookupNodeDiscovery
  )
  {
    this.druidNodeDiscoveryProvider = druidNodeDiscoveryProvider;
    this.configManager = configManager;
    this.lookupCoordinatorManagerConfig = lookupCoordinatorManagerConfig;
    this.lookupsCommunicator = lookupsCommunicator;
    this.lookupNodeDiscovery = lookupNodeDiscovery;
  }

  public boolean updateLookup(
      final String tier,
      final String lookupName,
      LookupExtractorFactoryMapContainer spec,
      final AuditInfo auditInfo
  )
  {
    return updateLookups(
        ImmutableMap.of(tier, ImmutableMap.of(lookupName, spec)),
        auditInfo
    );
  }

  public boolean updateLookups(final Map<String, Map<String, LookupExtractorFactoryMapContainer>> updateSpec, AuditInfo auditInfo)
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");

    if (updateSpec.isEmpty() && lookupMapConfigRef.get() != null) {
      return true;
    }

    //ensure all the lookups specs have version specified. ideally this should be done in the LookupExtractorFactoryMapContainer
    //constructor but that allows null to enable backward compatibility with 0.10.0 lookup specs
    for (final Map.Entry<String, Map<String, LookupExtractorFactoryMapContainer>> tierEntry : updateSpec.entrySet()) {
      for (Map.Entry<String, LookupExtractorFactoryMapContainer> e : tierEntry.getValue().entrySet()) {
        Preconditions.checkNotNull(
            e.getValue().getVersion(),
            "lookup [%s]:[%s] does not have version.", tierEntry.getKey(), e.getKey()
        );
      }
    }

    synchronized (this) {
      final Map<String, Map<String, LookupExtractorFactoryMapContainer>> priorSpec = getKnownLookups();
      if (priorSpec == null && !updateSpec.isEmpty()) {
        // To prevent accidentally erasing configs if we haven't updated our cache of the values
        throw new ISE("Not initialized. If this is the first lookup, post an empty map to initialize");
      }
      final Map<String, Map<String, LookupExtractorFactoryMapContainer>> updatedSpec;

      // Only add or update here, don't delete.
      if (priorSpec == null) {
        // all new
        updatedSpec = updateSpec;
      } else {
        // Needs update
        updatedSpec = new HashMap<>(priorSpec);
        for (final Map.Entry<String, Map<String, LookupExtractorFactoryMapContainer>> tierEntry : updateSpec.entrySet()) {
          final String tier = tierEntry.getKey();
          final Map<String, LookupExtractorFactoryMapContainer> updateTierSpec = tierEntry.getValue();
          final Map<String, LookupExtractorFactoryMapContainer> priorTierSpec = priorSpec.get(tier);

          if (priorTierSpec == null) {
            // New tier
            updatedSpec.put(tier, updateTierSpec);
          } else {
            // Update existing tier
            final Map<String, LookupExtractorFactoryMapContainer> updatedTierSpec = new HashMap<>(priorTierSpec);

            for (Map.Entry<String, LookupExtractorFactoryMapContainer> e : updateTierSpec.entrySet()) {
              if (updatedTierSpec.containsKey(e.getKey()) && !e.getValue().replaces(updatedTierSpec.get(e.getKey()))) {
                throw new IAE(
                    "given update for lookup [%s]:[%s] can't replace existing spec [%s].",
                    tier,
                    e.getKey(),
                    updatedTierSpec.get(e.getKey())
                );
              }
            }
            updatedTierSpec.putAll(updateTierSpec);
            updatedSpec.put(tier, updatedTierSpec);
          }
        }
      }
      return configManager.set(LOOKUP_CONFIG_KEY, updatedSpec, auditInfo).isOk();
    }
  }

  public Map<String, Map<String, LookupExtractorFactoryMapContainer>> getKnownLookups()
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");
    return lookupMapConfigRef.get();
  }

  public boolean deleteTier(final String tier, AuditInfo auditInfo)
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");

    synchronized (this) {
      final Map<String, Map<String, LookupExtractorFactoryMapContainer>> priorSpec = getKnownLookups();
      if (priorSpec == null) {
        LOG.warn("Requested delete tier [%s]. But no lookups exist!", tier);
        return false;
      }
      final Map<String, Map<String, LookupExtractorFactoryMapContainer>> updateSpec = new HashMap<>(priorSpec);

      if (updateSpec.remove(tier) == null) {
        LOG.warn("Requested delete of tier [%s] that does not exist!", tier);
        return false;
      }

      return configManager.set(LOOKUP_CONFIG_KEY, updateSpec, auditInfo).isOk();
    }
  }

  public boolean deleteLookup(final String tier, final String lookup, AuditInfo auditInfo)
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");

    synchronized (this) {
      final Map<String, Map<String, LookupExtractorFactoryMapContainer>> priorSpec = getKnownLookups();
      if (priorSpec == null) {
        LOG.warn("Requested delete lookup [%s]/[%s]. But no lookups exist!", tier, lookup);
        return false;
      }
      final Map<String, Map<String, LookupExtractorFactoryMapContainer>> updateSpec = new HashMap<>(priorSpec);
      final Map<String, LookupExtractorFactoryMapContainer> priorTierSpec = updateSpec.get(tier);
      if (priorTierSpec == null) {
        LOG.warn("Requested delete of lookup [%s]/[%s] but tier does not exist!", tier, lookup);
        return false;
      }

      if (!priorTierSpec.containsKey(lookup)) {
        LOG.warn("Requested delete of lookup [%s]/[%s] but lookup does not exist!", tier, lookup);
        return false;
      }

      final Map<String, LookupExtractorFactoryMapContainer> updateTierSpec = new HashMap<>(priorTierSpec);
      updateTierSpec.remove(lookup);

      if (updateTierSpec.isEmpty()) {
        updateSpec.remove(tier);
      } else {
        updateSpec.put(tier, updateTierSpec);
      }
      return configManager.set(LOOKUP_CONFIG_KEY, updateSpec, auditInfo).isOk();
    }
  }

  public Set<String> discoverTiers()
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");
    return lookupNodeDiscovery.getAllTiers();
  }

  public Collection<HostAndPort> discoverNodesInTier(String tier)
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");
    return Collections2.transform(
        lookupNodeDiscovery.getNodesInTier(tier),
        new Function<HostAndPortWithScheme, HostAndPort>()
        {
          @Override
          public HostAndPort apply(HostAndPortWithScheme input)
          {
            return input.getHostAndPort();
          }
        }
    );
  }

  public Map<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> getLastKnownLookupsStateOnNodes()
  {
    Preconditions.checkState(lifecycleLock.awaitStarted(5, TimeUnit.SECONDS), "not started");
    return knownOldState.get();
  }

  /**
   * Try to find a lookupName spec for the specified lookupName.
   *
   * @param lookupName The lookupName to look for
   *
   * @return The lookupName spec if found or null if not found or if no lookups at all are found
   */
  @Nullable
  public LookupExtractorFactoryMapContainer getLookup(final String tier, final String lookupName)
  {
    final Map<String, Map<String, LookupExtractorFactoryMapContainer>> prior = getKnownLookups();
    if (prior == null) {
      LOG.warn("Requested tier [%s] lookupName [%s]. But no lookups exist!", tier, lookupName);
      return null;
    }
    final Map<String, LookupExtractorFactoryMapContainer> tierLookups = prior.get(tier);
    if (tierLookups == null) {
      LOG.warn("Tier [%s] does not exist", tier);
      return null;
    }
    return tierLookups.get(lookupName);
  }

  public boolean isStarted()
  {
    return lifecycleLock.isStarted();
  }

  @VisibleForTesting
  boolean awaitStarted(long waitTimeMs)
  {
    return lifecycleLock.awaitStarted(waitTimeMs, TimeUnit.MILLISECONDS);
  }

  // start() and stop() are synchronized so that they never run in parallel in case of ZK acting funny or druid bug and
  // coordinator becomes leader and drops leadership in quick succession.
  public void start()
  {
    synchronized (lifecycleLock) {
      if (!lifecycleLock.canStart()) {
        throw new ISE("LookupCoordinatorManager can't start.");
      }

      try {
        LOG.debug("Starting.");

        if (lookupNodeDiscovery == null) {
          lookupNodeDiscovery = new LookupNodeDiscovery(druidNodeDiscoveryProvider);
        }

        //first ensure that previous executorService from last cycle of start/stop has finished completely.
        //so that we don't have multiple live executorService instances lying around doing lookup management.
        if (executorService != null &&
            !executorService.awaitTermination(
                lookupCoordinatorManagerConfig.getHostTimeout().getMillis() * 10,
                TimeUnit.MILLISECONDS
            )) {
          throw new ISE("LookupCoordinatorManager executor from last start() hasn't finished. Failed to Start.");
        }

        executorService = MoreExecutors.listeningDecorator(
            Executors.newScheduledThreadPool(
                lookupCoordinatorManagerConfig.getThreadPoolSize(),
                Execs.makeThreadFactory("LookupCoordinatorManager--%s")
            )
        );

        initializeLookupsConfigWatcher();

        this.backgroundManagerExitedLatch = new CountDownLatch(1);
        this.backgroundManagerFuture = executorService.scheduleWithFixedDelay(
            this::lookupManagementLoop,
            lookupCoordinatorManagerConfig.getInitialDelay(),
            lookupCoordinatorManagerConfig.getPeriod(),
            TimeUnit.MILLISECONDS
        );
        Futures.addCallback(
            backgroundManagerFuture,
            new FutureCallback<Object>()
            {
              @Override
              public void onSuccess(@Nullable Object result)
              {
                backgroundManagerExitedLatch.countDown();
                LOG.debug("Exited background lookup manager");
              }

              @Override
              public void onFailure(Throwable t)
              {
                backgroundManagerExitedLatch.countDown();
                if (backgroundManagerFuture.isCancelled()) {
                  LOG.debug("Exited background lookup manager due to cancellation.");
                } else {
                  LOG.makeAlert(t, "Background lookup manager exited with error!").emit();
                }
              }
            }
        );

        LOG.debug("Started");
      }
      catch (Exception ex) {
        LOG.makeAlert(ex, "Got Exception while start()").emit();
      }
      finally {
        //so that subsequent stop() would happen, even if start() failed with exception
        lifecycleLock.started();
        lifecycleLock.exitStart();
      }
    }
  }

  public void stop()
  {
    synchronized (lifecycleLock) {
      if (!lifecycleLock.canStop()) {
        throw new ISE("LookupCoordinatorManager can't stop.");
      }

      try {
        LOG.debug("Stopping");

        if (backgroundManagerFuture != null && !backgroundManagerFuture.cancel(true)) {
          LOG.warn("Background lookup manager thread could not be cancelled");
        }

        // signal the executorService to shut down ASAP, if this coordinator becomes leader again
        // then start() would ensure that this executorService is finished before starting a
        // new one.
        if (executorService != null) {
          executorService.shutdownNow();
        }

        LOG.debug("Stopped");
      }
      catch (Exception ex) {
        LOG.makeAlert(ex, "Got Exception while stop()").emit();
      }
      finally {
        //so that subsequent start() would happen, even if stop() failed with exception
        lifecycleLock.exitStopAndReset();
      }
    }
  }

  private void initializeLookupsConfigWatcher()
  {
    //Note: this call is idempotent, so multiple start() would not cause any problems.
    lookupMapConfigRef = configManager.watch(
        LOOKUP_CONFIG_KEY,
        new TypeReference<Map<String, Map<String, LookupExtractorFactoryMapContainer>>>()
        {
        },
        null
    );

    // backward compatibility with 0.10.0
    if (lookupMapConfigRef.get() == null) {
      Map<String, Map<String, Map<String, Object>>> oldLookups = configManager.watch(
          OLD_LOOKUP_CONFIG_KEY,
          new TypeReference<Map<String, Map<String, Map<String, Object>>>>()
          {
          },
          null
      ).get();

      if (oldLookups != null) {
        Map<String, Map<String, LookupExtractorFactoryMapContainer>> converted = new HashMap<>();
        oldLookups.forEach(
            (tier, oldTierLookups) -> {
              if (oldTierLookups != null && !oldTierLookups.isEmpty()) {
                converted.put(tier, convertTierLookups(oldTierLookups));
              }
            }
        );

        configManager.set(
            LOOKUP_CONFIG_KEY,
            converted,
            new AuditInfo("autoConversion", "autoConversion", "127.0.0.1")
        );
      }
    }
  }

  private Map<String, LookupExtractorFactoryMapContainer> convertTierLookups(
      Map<String, Map<String, Object>> oldTierLookups
  )
  {
    Map<String, LookupExtractorFactoryMapContainer> convertedTierLookups = new HashMap<>();
    oldTierLookups.forEach(
        (lookup, lookupExtractorFactory) -> {
          convertedTierLookups.put(lookup, new LookupExtractorFactoryMapContainer(null, lookupExtractorFactory));
        }
    );
    return convertedTierLookups;
  }

  @VisibleForTesting
  void lookupManagementLoop()
  {
    // Sanity check for if we are shutting down
    if (Thread.currentThread().isInterrupted() || !lifecycleLock.awaitStarted(15, TimeUnit.SECONDS)) {
      LOG.info("Not updating lookups because process was interrupted or not finished starting yet.");
      return;
    }

    final Map<String, Map<String, LookupExtractorFactoryMapContainer>> allLookupTiers = lookupMapConfigRef.get();

    if (allLookupTiers == null) {
      LOG.info("Not updating lookups because no data exists");
      return;
    }

    LOG.debug("Starting lookup sync for on all nodes.");

    try {
      List<ListenableFuture<Map.Entry>> futures = new ArrayList<>();

      Set<String> discoveredLookupTiers = lookupNodeDiscovery.getAllTiers();

      // Check and Log warnings about lookups configured by user in DB but no nodes discovered to load those.
      for (String tierInDB : allLookupTiers.keySet()) {
        if (!discoveredLookupTiers.contains(tierInDB) &&
            !allLookupTiers.getOrDefault(tierInDB, ImmutableMap.of()).isEmpty()) {
          LOG.warn("Found lookups for tier [%s] in DB, but no nodes discovered for it", tierInDB);
        }
      }

      for (String tier : discoveredLookupTiers) {

        LOG.debug("Starting lookup mgmt for tier [%s].", tier);

        final Map<String, LookupExtractorFactoryMapContainer> tierLookups = allLookupTiers.getOrDefault(tier, ImmutableMap.of());
        for (final HostAndPortWithScheme node : lookupNodeDiscovery.getNodesInTier(tier)) {

          LOG.debug(
              "Starting lookup mgmt for tier [%s] and host [%s:%s:%s].",
              tier,
              node.getScheme(),
              node.getHostText(),
              node.getPort()
          );

          futures.add(
              executorService.submit(
                  () -> {
                    try {
                      return new AbstractMap.SimpleImmutableEntry<>(node.getHostAndPort(), doLookupManagementOnNode(node, tierLookups));
                    }
                    catch (InterruptedException ex) {
                      LOG.warn(ex, "lookup management on node [%s:%s:%s] interrupted.", node.getScheme(), node.getHostText(), node.getPort());
                      return null;
                    }
                    catch (Exception ex) {
                      LOG.makeAlert(
                          ex,
                          "Failed to finish lookup management on node [%s:%s:%s]",
                          node.getScheme(),
                          node.getHostText(),
                          node.getPort()
                      ).emit();
                      return null;
                    }
                  }
              )
          );
        }
      }

      final ListenableFuture<List<Map.Entry>> allFuture = Futures.allAsList(futures);
      try {
        ImmutableMap.Builder<HostAndPort, LookupsState<LookupExtractorFactoryMapContainer>> stateBuilder = ImmutableMap.builder();
        allFuture.get(lookupCoordinatorManagerConfig.getAllHostTimeout().getMillis(), TimeUnit.MILLISECONDS)
                 .stream()
                 .filter(Objects::nonNull)
                 .forEach(stateBuilder::put);
        knownOldState.set(stateBuilder.build());
      }
      catch (InterruptedException ex) {
        allFuture.cancel(true);
        Thread.currentThread().interrupt();
        throw ex;
      }
      catch (Exception ex) {
        allFuture.cancel(true);
        throw ex;
      }

    }
    catch (Exception ex) {
      LOG.makeAlert(ex, "Failed to finish lookup management loop.").emit();
    }

    LOG.debug("Finished lookup sync for on all nodes.");
  }

  private LookupsState<LookupExtractorFactoryMapContainer> doLookupManagementOnNode(
      HostAndPortWithScheme node,
      Map<String, LookupExtractorFactoryMapContainer> nodeTierLookupsToBe
  ) throws IOException, InterruptedException, ExecutionException
  {
    LOG.debug("Starting lookup sync for node [%s].", node);

    LookupsState<LookupExtractorFactoryMapContainer> currLookupsStateOnNode = lookupsCommunicator.getLookupStateForNode(
        node
    );
    LOG.debug("Received lookups state from node [%s].", node);


    // Compare currLookupsStateOnNode with nodeTierLookupsToBe to find what are the lookups
    // we need to further ask node to load/drop
    Map<String, LookupExtractorFactoryMapContainer> toLoad = getToBeLoadedOnNode(
        currLookupsStateOnNode,
        nodeTierLookupsToBe
    );
    Set<String> toDrop = getToBeDroppedFromNode(currLookupsStateOnNode, nodeTierLookupsToBe);

    if (!toLoad.isEmpty() || !toDrop.isEmpty()) {
      // Send POST request to the node asking to load and drop the lookups necessary.
      // no need to send "current" in the LookupsStateWithMap , that is not required
      currLookupsStateOnNode = lookupsCommunicator.updateNode(node, new LookupsState<>(null, toLoad, toDrop));

      LOG.debug(
          "Sent lookup toAdd[%s] and toDrop[%s] updates to node [%s].",
          toLoad.keySet(),
          toDrop,
          node
      );
    }

    LOG.debug("Finished lookup sync for node [%s].", node);
    return currLookupsStateOnNode;
  }

  // Returns the Map<lookup-name, lookup-spec> that needs to be loaded by the node and it does not know about
  // those already.
  // It is assumed that currLookupsStateOnNode "toLoad" and "toDrop" are disjoint.
  @VisibleForTesting
  Map<String, LookupExtractorFactoryMapContainer> getToBeLoadedOnNode(
      LookupsState<LookupExtractorFactoryMapContainer> currLookupsStateOnNode,
      Map<String, LookupExtractorFactoryMapContainer> nodeTierLookupsToBe
  )
  {
    Map<String, LookupExtractorFactoryMapContainer> toLoad = new HashMap<>();
    for (Map.Entry<String, LookupExtractorFactoryMapContainer> e : nodeTierLookupsToBe.entrySet()) {
      String name = e.getKey();
      LookupExtractorFactoryMapContainer lookupToBe = e.getValue();

      // get it from the current pending notices list on the node
      LookupExtractorFactoryMapContainer current = currLookupsStateOnNode.getToLoad().get(name);

      if (current == null) {
        //ok, not on pending list, get from currently loaded lookups on node
        current = currLookupsStateOnNode.getCurrent().get(name);
      }

      if (current == null || //lookup is neither pending nor already loaded on the node OR
          currLookupsStateOnNode.getToDrop().contains(name) || //it is being dropped on the node OR
          lookupToBe.replaces(current) //lookup is already know to node, but lookupToBe overrides that
          ) {
        toLoad.put(name, lookupToBe);
      }
    }
    return toLoad;
  }

  // Returns Set<lookup-name> that should be dropped from the node which has them already either in pending to load
  // state or loaded
  // It is assumed that currLookupsStateOnNode "toLoad" and "toDrop" are disjoint.
  @VisibleForTesting
  Set<String> getToBeDroppedFromNode(
      LookupsState<LookupExtractorFactoryMapContainer> currLookupsStateOnNode,
      Map<String, LookupExtractorFactoryMapContainer> nodeTierLookupsToBe
  )
  {
    Set<String> toDrop = new HashSet<>();

    // {currently loading/loaded on the node} - {currently pending deletion on node} - {lookups node should actually have}
    toDrop.addAll(currLookupsStateOnNode.getCurrent().keySet());
    toDrop.addAll(currLookupsStateOnNode.getToLoad().keySet());
    toDrop = Sets.difference(toDrop, currLookupsStateOnNode.getToDrop());
    toDrop = Sets.difference(toDrop, nodeTierLookupsToBe.keySet());
    return toDrop;
  }

  static URL getLookupsURL(HostAndPortWithScheme druidNode) throws MalformedURLException
  {
    return new URL(
        druidNode.getScheme(),
        druidNode.getHostText(),
        druidNode.getPortOrDefault(-1),
        LOOKUP_BASE_REQUEST_PATH
    );
  }

  static URL getLookupsUpdateURL(HostAndPortWithScheme druidNode) throws MalformedURLException
  {
    return new URL(
        druidNode.getScheme(),
        druidNode.getHostText(),
        druidNode.getPortOrDefault(-1),
        LOOKUP_UPDATE_REQUEST_PATH
    );
  }

  private static boolean httpStatusIsSuccess(int statusCode)
  {
    return statusCode >= 200 && statusCode < 300;
  }

  @VisibleForTesting
  boolean backgroundManagerIsRunning()
  {
    ListenableScheduledFuture backgroundManagerFuture = this.backgroundManagerFuture;
    return backgroundManagerFuture != null && !backgroundManagerFuture.isDone();
  }

  @VisibleForTesting
  boolean waitForBackgroundTermination(long timeout) throws InterruptedException
  {
    return backgroundManagerExitedLatch.await(timeout, TimeUnit.MILLISECONDS);
  }

  @VisibleForTesting
  public static class LookupsCommunicator
  {
    private final HttpClient httpClient;
    private final LookupCoordinatorManagerConfig lookupCoordinatorManagerConfig;
    private final ObjectMapper smileMapper;

    public LookupsCommunicator(
        HttpClient httpClient,
        LookupCoordinatorManagerConfig lookupCoordinatorManagerConfig,
        ObjectMapper smileMapper
    )
    {
      this.httpClient = httpClient;
      this.lookupCoordinatorManagerConfig = lookupCoordinatorManagerConfig;
      this.smileMapper = smileMapper;
    }

    public LookupsState<LookupExtractorFactoryMapContainer> updateNode(
        HostAndPortWithScheme node,
        LookupsState<LookupExtractorFactoryMapContainer> lookupsUpdate
    )
        throws IOException, InterruptedException, ExecutionException
    {
      final AtomicInteger returnCode = new AtomicInteger(0);
      final AtomicReference<String> reasonString = new AtomicReference<>(null);

      final URL url = getLookupsUpdateURL(node);

      LOG.debug("Sending lookups load/drop request to [%s]. Request [%s]", url, lookupsUpdate);

      try (final InputStream result = httpClient.go(
          new Request(HttpMethod.POST, url)
              .addHeader(HttpHeaders.Names.ACCEPT, SmileMediaTypes.APPLICATION_JACKSON_SMILE)
              .addHeader(HttpHeaders.Names.CONTENT_TYPE, SmileMediaTypes.APPLICATION_JACKSON_SMILE)
              .setContent(smileMapper.writeValueAsBytes(lookupsUpdate)),
          makeResponseHandler(returnCode, reasonString),
          lookupCoordinatorManagerConfig.getHostTimeout()
      ).get()) {
        if (httpStatusIsSuccess(returnCode.get())) {
          try {
            final LookupsState<LookupExtractorFactoryMapContainer> response = smileMapper.readValue(
                result,
                LOOKUPS_STATE_TYPE_REFERENCE
            );
            LOG.debug(
                "Update on [%s], Status: %s reason: [%s], Response [%s].", url, returnCode.get(), reasonString.get(),
                response
            );
            return response;
          }
          catch (IOException ex) {
            throw new IOE(ex, "Failed to parse update response from [%s]. response [%s]", url, result);
          }
        } else {
          final ByteArrayOutputStream baos = new ByteArrayOutputStream();
          try {
            StreamUtils.copyAndClose(result, baos);
          }
          catch (IOException e2) {
            LOG.warn(e2, "Error reading response");
          }

          throw new IOE(
              "Bad update request to [%s] : [%d] : [%s]  Response: [%s]",
              url,
              returnCode.get(),
              reasonString.get(),
              StringUtils.fromUtf8(baos.toByteArray())
          );
        }
      }
    }

    public LookupsState<LookupExtractorFactoryMapContainer> getLookupStateForNode(
        HostAndPortWithScheme node
    ) throws IOException, InterruptedException, ExecutionException
    {
      final URL url = getLookupsURL(node);
      final AtomicInteger returnCode = new AtomicInteger(0);
      final AtomicReference<String> reasonString = new AtomicReference<>(null);

      LOG.debug("Getting lookups from [%s]", url);

      try (final InputStream result = httpClient.go(
          new Request(HttpMethod.GET, url)
              .addHeader(HttpHeaders.Names.ACCEPT, SmileMediaTypes.APPLICATION_JACKSON_SMILE),
          makeResponseHandler(returnCode, reasonString),
          lookupCoordinatorManagerConfig.getHostTimeout()
      ).get()) {
        if (returnCode.get() == HttpURLConnection.HTTP_OK) {
          try {
            final LookupsState<LookupExtractorFactoryMapContainer> response = smileMapper.readValue(
                result,
                LOOKUPS_STATE_TYPE_REFERENCE
            );
            LOG.debug(
                "Get on [%s], Status: [%s] reason: [%s], Response [%s].",
                url,
                returnCode.get(),
                reasonString.get(),
                response
            );
            return response;
          }
          catch (IOException ex) {
            throw new IOE(ex, "Failed to parser GET lookups response from [%s]. response [%s].", url, result);
          }
        } else {
          final ByteArrayOutputStream baos = new ByteArrayOutputStream();
          try {
            StreamUtils.copyAndClose(result, baos);
          }
          catch (IOException ex) {
            LOG.warn(ex, "Error reading response from GET on url [%s]", url);
          }

          throw new IOE(
              "GET request failed to [%s] : [%d] : [%s]  Response: [%s]",
              url,
              returnCode.get(),
              reasonString.get(),
              StringUtils.fromUtf8(baos.toByteArray())
          );
        }
      }
    }

    @VisibleForTesting
    HttpResponseHandler<InputStream, InputStream> makeResponseHandler(
        final AtomicInteger returnCode,
        final AtomicReference<String> reasonString
    )
    {
      return new SequenceInputStreamResponseHandler()
      {
        @Override
        public ClientResponse<InputStream> handleResponse(HttpResponse response, TrafficCop trafficCop)
        {
          returnCode.set(response.getStatus().getCode());
          reasonString.set(response.getStatus().getReasonPhrase());
          return super.handleResponse(response, trafficCop);
        }
      };
    }
  }
}