/*
 * 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.iotdb.cluster.server.member;

import org.apache.iotdb.cluster.client.async.AsyncClientPool;
import org.apache.iotdb.cluster.client.async.AsyncDataClient;
import org.apache.iotdb.cluster.client.async.AsyncDataClient.SingleManagerFactory;
import org.apache.iotdb.cluster.client.async.AsyncDataHeartbeatClient;
import org.apache.iotdb.cluster.client.sync.SyncClientPool;
import org.apache.iotdb.cluster.client.sync.SyncDataClient;
import org.apache.iotdb.cluster.client.sync.SyncDataHeartbeatClient;
import org.apache.iotdb.cluster.config.ClusterConstant;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.exception.LogExecutionException;
import org.apache.iotdb.cluster.exception.SnapshotInstallationException;
import org.apache.iotdb.cluster.log.LogApplier;
import org.apache.iotdb.cluster.log.Snapshot;
import org.apache.iotdb.cluster.log.applier.AsyncDataLogApplier;
import org.apache.iotdb.cluster.log.applier.DataLogApplier;
import org.apache.iotdb.cluster.log.logtypes.CloseFileLog;
import org.apache.iotdb.cluster.log.manage.FilePartitionedSnapshotLogManager;
import org.apache.iotdb.cluster.log.manage.PartitionedSnapshotLogManager;
import org.apache.iotdb.cluster.log.snapshot.FileSnapshot;
import org.apache.iotdb.cluster.log.snapshot.PartitionedSnapshot;
import org.apache.iotdb.cluster.log.snapshot.PullSnapshotTask;
import org.apache.iotdb.cluster.log.snapshot.PullSnapshotTaskDescriptor;
import org.apache.iotdb.cluster.partition.NodeAdditionResult;
import org.apache.iotdb.cluster.partition.NodeRemovalResult;
import org.apache.iotdb.cluster.partition.PartitionGroup;
import org.apache.iotdb.cluster.partition.slot.SlotManager;
import org.apache.iotdb.cluster.partition.slot.SlotManager.SlotStatus;
import org.apache.iotdb.cluster.partition.slot.SlotNodeAdditionResult;
import org.apache.iotdb.cluster.partition.slot.SlotNodeRemovalResult;
import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable;
import org.apache.iotdb.cluster.query.LocalQueryExecutor;
import org.apache.iotdb.cluster.query.manage.ClusterQueryManager;
import org.apache.iotdb.cluster.rpc.thrift.ElectionRequest;
import org.apache.iotdb.cluster.rpc.thrift.Node;
import org.apache.iotdb.cluster.rpc.thrift.PullSnapshotRequest;
import org.apache.iotdb.cluster.rpc.thrift.PullSnapshotResp;
import org.apache.iotdb.cluster.rpc.thrift.SendSnapshotRequest;
import org.apache.iotdb.cluster.server.NodeCharacter;
import org.apache.iotdb.cluster.server.PullSnapshotHintService;
import org.apache.iotdb.cluster.server.Response;
import org.apache.iotdb.cluster.server.heartbeat.DataHeartbeatThread;
import org.apache.iotdb.cluster.server.monitor.NodeReport.DataMemberReport;
import org.apache.iotdb.cluster.server.monitor.NodeStatusManager;
import org.apache.iotdb.cluster.server.monitor.Peer;
import org.apache.iotdb.cluster.server.monitor.Timer;
import org.apache.iotdb.cluster.server.monitor.Timer.Statistic;
import org.apache.iotdb.cluster.utils.StatusUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.executor.PlanExecutor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.sys.FlushPlan;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.service.rpc.thrift.EndPoint;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.iotdb.tsfile.utils.Pair;

import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class DataGroupMember extends RaftMember {

  private static final Logger logger = LoggerFactory.getLogger(DataGroupMember.class);

  /**
   * The MetaGroupMember that in charge of the DataGroupMember. Mainly for providing partition table
   * and MetaLogManager.
   */
  private MetaGroupMember metaGroupMember;

  /** The thread pool that runs the pull snapshot tasks. Pool size is the # of CPU cores. */
  private ExecutorService pullSnapshotService;

  /**
   * When the member applies a pulled snapshot, it register hints in this service which will
   * periodically inform the data source that one member has pulled snapshot.
   */
  private PullSnapshotHintService pullSnapshotHintService;

  /**
   * "queryManger" records the remote nodes which have queried this node, and the readers or
   * executors this member has created for those queries. When the queries end, an EndQueryRequest
   * will be sent to this member and related resources will be released.
   */
  private ClusterQueryManager queryManager;

  /**
   * "slotManager" tracks the status of slots during data transfers so that we can know whether the
   * slot has non-pulled data.
   */
  protected SlotManager slotManager;

  private LocalQueryExecutor localQueryExecutor;

  /**
   * When a new partition table is installed, all data members will be checked if unchanged. If not,
   * such members will be removed.
   */
  private boolean unchanged;

  @TestOnly
  public DataGroupMember() {
    // constructor for test
    setQueryManager(new ClusterQueryManager());
    localQueryExecutor = new LocalQueryExecutor(this);
  }

  DataGroupMember(
      TProtocolFactory factory,
      PartitionGroup nodes,
      Node thisNode,
      MetaGroupMember metaGroupMember) {
    super(
        "Data(" + nodes.getHeader().getInternalIp() + ":" + nodes.getHeader().getMetaPort() + ")",
        new AsyncClientPool(new AsyncDataClient.FactoryAsync(factory)),
        new SyncClientPool(new SyncDataClient.FactorySync(factory)),
        new AsyncClientPool(new AsyncDataHeartbeatClient.FactoryAsync(factory)),
        new SyncClientPool(new SyncDataHeartbeatClient.FactorySync(factory)),
        new AsyncClientPool(new SingleManagerFactory(factory)));
    this.thisNode = thisNode;
    this.metaGroupMember = metaGroupMember;
    allNodes = nodes;
    setQueryManager(new ClusterQueryManager());
    slotManager = new SlotManager(ClusterConstant.SLOT_NUM, getMemberDir());
    LogApplier applier = new DataLogApplier(metaGroupMember, this);
    if (ClusterDescriptor.getInstance().getConfig().isUseAsyncApplier()) {
      applier = new AsyncDataLogApplier(applier, name);
    }
    logManager =
        new FilePartitionedSnapshotLogManager(
            applier, metaGroupMember.getPartitionTable(), allNodes.get(0), thisNode, this);
    initPeerMap();
    term.set(logManager.getHardState().getCurrentTerm());
    voteFor = logManager.getHardState().getVoteFor();
    localQueryExecutor = new LocalQueryExecutor(this);
  }

  /**
   * Start heartbeat, catch-up, pull snapshot services and start all unfinished pull-snapshot-tasks.
   * Calling the method twice does not induce side effects.
   *
   * @throws TTransportException
   */
  @Override
  public void start() {
    if (heartBeatService != null) {
      return;
    }
    super.start();
    heartBeatService.submit(new DataHeartbeatThread(this));
    pullSnapshotService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    pullSnapshotHintService = new PullSnapshotHintService(this);
    resumePullSnapshotTasks();
  }

  /**
   * Stop heartbeat, catch-up and pull snapshot services and release all query resources. Calling
   * the method twice does not induce side effects.
   */
  @Override
  public void stop() {
    logger.info("{}: stopping...", name);
    super.stop();
    if (pullSnapshotService != null) {
      pullSnapshotService.shutdownNow();
      try {
        pullSnapshotService.awaitTermination(10, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        logger.error("Unexpected interruption when waiting for pullSnapshotService to end", e);
      }
      pullSnapshotService = null;
      pullSnapshotHintService.stop();
    }

    try {
      getQueryManager().endAllQueries();
    } catch (StorageEngineException e) {
      logger.error("Cannot release queries of {}", name, e);
    }
    logger.info("{}: stopped", name);
  }

  /**
   * The first node (on the hash ring) in this data group is the header. It determines the duty
   * (what range on the ring do the group take responsibility for) of the group and although other
   * nodes in this may change, this node is unchangeable unless the data group is dismissed. It is
   * also the identifier of this data group.
   */
  @Override
  public Node getHeader() {
    return allNodes.get(0);
  }

  public ClusterQueryManager getQueryManager() {
    return queryManager;
  }

  protected void setQueryManager(ClusterQueryManager queryManager) {
    this.queryManager = queryManager;
  }

  public static class Factory {

    private TProtocolFactory protocolFactory;
    private MetaGroupMember metaGroupMember;

    Factory(TProtocolFactory protocolFactory, MetaGroupMember metaGroupMember) {
      this.protocolFactory = protocolFactory;
      this.metaGroupMember = metaGroupMember;
    }

    public DataGroupMember create(PartitionGroup partitionGroup, Node thisNode) {
      return new DataGroupMember(protocolFactory, partitionGroup, thisNode, metaGroupMember);
    }
  }

  /**
   * Try to add a Node into the group to which the member belongs.
   *
   * @param node
   * @return true if this node should leave the group because of the addition of the node, false
   *     otherwise
   */
  public synchronized boolean addNode(Node node, NodeAdditionResult result) {
    // when a new node is added, start an election instantly to avoid the stale leader still
    // taking the leadership, which guarantees the valid leader will not have the stale
    // partition table
    synchronized (term) {
      term.incrementAndGet();
      setLeader(ClusterConstant.EMPTY_NODE);
      setVoteFor(thisNode);
      updateHardState(term.get(), getVoteFor());
      setLastHeartbeatReceivedTime(System.currentTimeMillis());
      setCharacter(NodeCharacter.ELECTOR);
    }

    // mark slots that do not belong to this group any more
    Set<Integer> lostSlots =
        ((SlotNodeAdditionResult) result)
            .getLostSlots()
            .getOrDefault(getHeader(), Collections.emptySet());
    for (Integer lostSlot : lostSlots) {
      slotManager.setToSending(lostSlot);
    }

    synchronized (allNodes) {
      int insertIndex = -1;
      // find the position to insert the new node, the nodes are ordered by their identifiers
      for (int i = 0; i < allNodes.size() - 1; i++) {
        Node prev = allNodes.get(i);
        Node next = allNodes.get(i + 1);
        if (prev.nodeIdentifier < node.nodeIdentifier && node.nodeIdentifier < next.nodeIdentifier
            || prev.nodeIdentifier < node.nodeIdentifier
                && next.nodeIdentifier < prev.nodeIdentifier
            || node.nodeIdentifier < next.nodeIdentifier
                && next.nodeIdentifier < prev.nodeIdentifier) {
          insertIndex = i + 1;
          break;
        }
      }
      if (insertIndex > 0) {
        allNodes.add(insertIndex, node);
        peerMap.putIfAbsent(node, new Peer(logManager.getLastLogIndex()));
        // remove the last node because the group size is fixed to replication number
        Node removedNode = allNodes.remove(allNodes.size() - 1);
        peerMap.remove(removedNode);
        // if the local node is the last node and the insertion succeeds, this node should leave
        // the group
        logger.debug("{}: Node {} is inserted into the data group {}", name, node, allNodes);
        return removedNode.equals(thisNode);
      }
      return false;
    }
  }

  /**
   * Process the election request from another node in the group. To win the vote from the local
   * member, a node must have both meta and data logs no older than then local member, or it will be
   * turned down.
   *
   * @param electionRequest
   * @return Response.RESPONSE_META_LOG_STALE if the meta logs of the elector fall behind
   *     Response.RESPONSE_LOG_MISMATCH if the data logs of the elector fall behind Response.SUCCESS
   *     if the vote is given to the elector the term of local member if the elector's term is no
   *     bigger than the local member
   */
  @Override
  long checkElectorLogProgress(ElectionRequest electionRequest) {
    // to be a data group leader, a node should also be qualified to be the meta group leader
    // which guarantees the data group leader has the newest partition table.
    long thatTerm = electionRequest.getTerm();
    long thatMetaLastLogIndex = electionRequest.getLastLogIndex();
    long thatMetaLastLogTerm = electionRequest.getLastLogTerm();
    long thatDataLastLogIndex = electionRequest.getDataLogLastIndex();
    long thatDataLastLogTerm = electionRequest.getDataLogLastTerm();
    logger.info(
        "{} received an dataGroup election request, term:{}, metaLastLogIndex:{}, metaLastLogTerm:{}, dataLastLogIndex:{}, dataLastLogTerm:{}",
        name,
        thatTerm,
        thatMetaLastLogIndex,
        thatMetaLastLogTerm,
        thatDataLastLogIndex,
        thatDataLastLogTerm);

    // check meta logs
    // term of the electors' MetaGroupMember is not verified, so 0 and 1 are used to make sure
    // the verification does not fail
    long metaResponse = metaGroupMember.checkLogProgress(thatMetaLastLogIndex, thatMetaLastLogTerm);
    if (metaResponse == Response.RESPONSE_LOG_MISMATCH) {
      return Response.RESPONSE_META_LOG_STALE;
    }

    long resp = checkLogProgress(thatDataLastLogIndex, thatDataLastLogTerm);
    if (resp == Response.RESPONSE_AGREE) {
      logger.info(
          "{} accepted an dataGroup election request, term:{}/{}, dataLogIndex:{}/{}, dataLogTerm:{}/{}, metaLogIndex:{}/{},metaLogTerm:{}/{}",
          name,
          thatTerm,
          term.get(),
          thatDataLastLogIndex,
          logManager.getLastLogIndex(),
          thatDataLastLogTerm,
          logManager.getLastLogTerm(),
          thatMetaLastLogIndex,
          metaGroupMember.getLogManager().getLastLogIndex(),
          thatMetaLastLogTerm,
          metaGroupMember.getLogManager().getLastLogTerm());
      setCharacter(NodeCharacter.FOLLOWER);
      lastHeartbeatReceivedTime = System.currentTimeMillis();
      setVoteFor(electionRequest.getElector());
      updateHardState(thatTerm, getVoteFor());
    } else {
      logger.info(
          "{} rejected an dataGroup election request, term:{}/{}, dataLogIndex:{}/{}, dataLogTerm:{}/{}, metaLogIndex:{}/{},metaLogTerm:{}/{}",
          name,
          thatTerm,
          term.get(),
          thatDataLastLogIndex,
          logManager.getLastLogIndex(),
          thatDataLastLogTerm,
          logManager.getLastLogTerm(),
          thatMetaLastLogIndex,
          metaGroupMember.getLogManager().getLastLogIndex(),
          thatMetaLastLogTerm,
          metaGroupMember.getLogManager().getLastLogTerm());
    }
    return resp;
  }

  /**
   * Deserialize and install a snapshot sent by the leader. The type of the snapshot must be
   * currently PartitionedSnapshot with FileSnapshot inside.
   *
   * @param request
   */
  public void receiveSnapshot(SendSnapshotRequest request) throws SnapshotInstallationException {
    logger.info(
        "{}: received a snapshot from {} with size {}",
        name,
        request.getHeader(),
        request.getSnapshotBytes().length);
    PartitionedSnapshot<FileSnapshot> snapshot =
        new PartitionedSnapshot<>(FileSnapshot.Factory.INSTANCE);

    snapshot.deserialize(ByteBuffer.wrap(request.getSnapshotBytes()));
    if (logger.isDebugEnabled()) {
      logger.debug("{} received a snapshot {}", name, snapshot);
    }
    snapshot.getDefaultInstaller(this).install(snapshot, -1);
  }

  /**
   * Send the requested snapshots to the applier node.
   *
   * @param request
   */
  public PullSnapshotResp getSnapshot(PullSnapshotRequest request) throws IOException {
    waitLeader();
    if (character != NodeCharacter.LEADER && !readOnly) {
      return null;
    }
    // if the requester pulls the snapshots because the header of the group is removed, then the
    // member should no longer receive new data
    if (request.isRequireReadOnly()) {
      setReadOnly();
    }

    List<Integer> requiredSlots = request.getRequiredSlots();
    for (Integer requiredSlot : requiredSlots) {
      // wait if the data of the slot is in another node
      slotManager.waitSlot(requiredSlot);
    }
    if (logger.isDebugEnabled()) {
      logger.debug(
          "{}: {} slots are requested, first:{}, last: {}",
          name,
          requiredSlots.size(),
          requiredSlots.get(0),
          requiredSlots.get(requiredSlots.size() - 1));
    }

    // If the logs between [currCommitLogIndex, currLastLogIndex] are committed after the
    // snapshot is generated, they will be invisible to the new slot owner and thus lost forever
    long currLastLogIndex = logManager.getLastLogIndex();
    logger.info(
        "{}: Waiting for logs to commit before snapshot, {}/{}",
        name,
        logManager.getCommitLogIndex(),
        currLastLogIndex);
    while (logManager.getCommitLogIndex() < currLastLogIndex) {
      try {
        Thread.sleep(10);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        logger.warn("{}: Unexpected interruption when waiting for logs to commit", name, e);
      }
    }

    // this synchronized should work with the one in AppendEntry when a log is going to commit,
    // which may prevent the newly arrived data from being invisible to the new header.
    synchronized (logManager) {
      PullSnapshotResp resp = new PullSnapshotResp();
      Map<Integer, ByteBuffer> resultMap = new HashMap<>();
      logManager.takeSnapshot();

      PartitionedSnapshot<Snapshot> allSnapshot = (PartitionedSnapshot) logManager.getSnapshot();
      for (int requiredSlot : requiredSlots) {
        Snapshot snapshot = allSnapshot.getSnapshot(requiredSlot);
        if (snapshot != null) {
          resultMap.put(requiredSlot, snapshot.serialize());
        }
      }
      resp.setSnapshotBytes(resultMap);
      logger.debug("{}: Sending {} snapshots to the requester", name, resultMap.size());
      return resp;
    }
  }

  /**
   * Pull snapshots from the previous holders after newNode joins the cluster.
   *
   * @param slots
   * @param newNode
   */
  public void pullNodeAdditionSnapshots(List<Integer> slots, Node newNode) {
    synchronized (logManager) {
      logger.info("{} pulling {} slots from remote", name, slots.size());
      PartitionedSnapshot<Snapshot> snapshot = (PartitionedSnapshot) logManager.getSnapshot();
      Map<Integer, Node> prevHolders =
          ((SlotPartitionTable) metaGroupMember.getPartitionTable()).getPreviousNodeMap(newNode);

      // group the slots by their owners
      Map<Node, List<Integer>> holderSlotsMap = new HashMap<>();
      for (int slot : slots) {
        // skip the slot if the corresponding data is already replicated locally
        if (snapshot.getSnapshot(slot) == null) {
          Node node = prevHolders.get(slot);
          if (node != null) {
            holderSlotsMap.computeIfAbsent(node, n -> new ArrayList<>()).add(slot);
          }
        }
      }

      // pull snapshots from each owner's data group
      for (Entry<Node, List<Integer>> entry : holderSlotsMap.entrySet()) {
        Node node = entry.getKey();
        List<Integer> nodeSlots = entry.getValue();
        PullSnapshotTaskDescriptor taskDescriptor =
            new PullSnapshotTaskDescriptor(
                metaGroupMember.getPartitionTable().getHeaderGroup(node), nodeSlots, false);
        pullFileSnapshot(taskDescriptor, null);
      }
    }
  }

  /**
   * Pull FileSnapshots (timeseries schemas and lists of TsFiles) of "nodeSlots" from one of the
   * "prevHolders". The actual pulling will be performed in a separate thread.
   *
   * @param descriptor
   * @param snapshotSave set to the corresponding disk file if the task is resumed from disk, or set
   *     ot null otherwise
   */
  private void pullFileSnapshot(PullSnapshotTaskDescriptor descriptor, File snapshotSave) {
    Iterator<Integer> iterator = descriptor.getSlots().iterator();
    while (iterator.hasNext()) {
      Integer nodeSlot = iterator.next();
      SlotStatus status = slotManager.getStatus(nodeSlot);
      if (status != SlotStatus.NULL) {
        // the pulling may already be issued during restart, skip it in that case
        iterator.remove();
      } else {
        // mark the slot as pulling to control reads and writes of the pulling slot
        slotManager.setToPulling(nodeSlot, descriptor.getPreviousHolders().getHeader());
      }
    }

    if (descriptor.getSlots().isEmpty()) {
      return;
    }
    if (logger.isInfoEnabled()) {
      logger.info(
          "{}: {} and other {} slots are set to pulling",
          name,
          descriptor.getSlots().get(0),
          descriptor.getSlots().size() - 1);
    }

    pullSnapshotService.submit(
        new PullSnapshotTask<>(descriptor, this, FileSnapshot.Factory.INSTANCE, snapshotSave));
  }

  /** Restart all unfinished pull-snapshot-tasks of the member. */
  private void resumePullSnapshotTasks() {
    File snapshotTaskDir = new File(getPullSnapshotTaskDir());
    if (!snapshotTaskDir.exists()) {
      return;
    }

    File[] files = snapshotTaskDir.listFiles();
    if (files != null) {
      for (File file : files) {
        if (!file.getName().endsWith(PullSnapshotTask.TASK_SUFFIX)) {
          continue;
        }
        try (DataInputStream dataInputStream =
            new DataInputStream(new BufferedInputStream(new FileInputStream(file)))) {
          PullSnapshotTaskDescriptor descriptor = new PullSnapshotTaskDescriptor();
          descriptor.deserialize(dataInputStream);
          pullFileSnapshot(descriptor, file);
        } catch (IOException e) {
          logger.error("Cannot resume pull-snapshot-task in file {}", file, e);
          try {
            Files.delete(file.toPath());
          } catch (IOException ex) {
            logger.debug("Cannot remove pull snapshot task file {}", file, e);
          }
        }
      }
    }
  }

  /** @return a directory that stores the information of ongoing pulling snapshot tasks. */
  public String getPullSnapshotTaskDir() {
    return getMemberDir() + "snapshot_task" + File.separator;
  }

  /** @return the path of the directory that is provided exclusively for the member. */
  private String getMemberDir() {
    return IoTDBDescriptor.getInstance().getConfig().getSystemDir()
        + File.separator
        + "raft"
        + File.separator
        + getHeader().nodeIdentifier
        + File.separator;
  }

  public MetaGroupMember getMetaGroupMember() {
    return metaGroupMember;
  }

  /**
   * If the member is the leader, let all members in the group close the specified partition of a
   * storage group, else just return false.
   *
   * @param storageGroupName
   * @param partitionId
   * @param isSeq
   */
  void closePartition(String storageGroupName, long partitionId, boolean isSeq) {
    if (character != NodeCharacter.LEADER) {
      return;
    }
    CloseFileLog log = new CloseFileLog(storageGroupName, partitionId, isSeq);
    synchronized (logManager) {
      log.setCurrLogTerm(getTerm().get());
      log.setCurrLogIndex(logManager.getLastLogIndex() + 1);

      logManager.append(log);

      logger.info("Send the close file request of {} to other nodes", log);
    }
    try {
      appendLogInGroup(log);
    } catch (LogExecutionException e) {
      logger.error("Cannot close partition {}#{} seq:{}", storageGroupName, partitionId, isSeq, e);
    }
  }

  public boolean flushFileWhenDoSnapshot(
      Map<String, List<Pair<Long, Boolean>>> storageGroupPartitions) {
    if (character != NodeCharacter.LEADER) {
      return false;
    }

    Map<PartialPath, List<Pair<Long, Boolean>>> localDataMemberStorageGroupPartitions =
        new HashMap<>();
    for (Entry<String, List<Pair<Long, Boolean>>> entry : storageGroupPartitions.entrySet()) {
      List<Pair<Long, Boolean>> localListPair = new ArrayList<>();

      String storageGroupName = entry.getKey();
      List<Pair<Long, Boolean>> tmpPairList = entry.getValue();
      for (Pair<Long, Boolean> pair : tmpPairList) {
        long partitionId = pair.left;
        Node header =
            metaGroupMember
                .getPartitionTable()
                .routeToHeaderByTime(
                    storageGroupName, partitionId * StorageEngine.getTimePartitionInterval());
        DataGroupMember localDataMember = metaGroupMember.getLocalDataMember(header);
        if (localDataMember.getHeader().equals(this.getHeader())) {
          localListPair.add(new Pair<>(partitionId, pair.right));
        }
      }
      try {
        localDataMemberStorageGroupPartitions.put(new PartialPath(storageGroupName), localListPair);
      } catch (IllegalPathException e) {
        // ignore
      }
    }

    if (localDataMemberStorageGroupPartitions.size() <= 0) {
      logger.info("{}: have no data to flush", name);
      return true;
    }
    FlushPlan flushPlan = new FlushPlan(null, true, localDataMemberStorageGroupPartitions);
    try {
      PlanExecutor.flushSpecifiedStorageGroups(flushPlan);
      return true;
    } catch (StorageGroupNotSetException e) {
      logger.error("Some SGs are missing while flushing", e);
    }
    return false;
  }

  /**
   * Execute a non-query plan. If the member is a leader, a log for the plan will be created and
   * process through the raft procedure, otherwise the plan will be forwarded to the leader.
   *
   * @param plan a non-query plan.
   * @return
   */
  @Override
  public TSStatus executeNonQueryPlan(PhysicalPlan plan) {
    TSStatus status = executeNonQueryPlanWithKnownLeader(plan);
    if (!StatusUtils.NO_LEADER.equals(status)) {
      return status;
    }

    long startTime = Timer.Statistic.DATA_GROUP_MEMBER_WAIT_LEADER.getOperationStartTime();
    waitLeader();
    Timer.Statistic.DATA_GROUP_MEMBER_WAIT_LEADER.calOperationCostTimeFromStart(startTime);

    return executeNonQueryPlanWithKnownLeader(plan);
  }

  private TSStatus executeNonQueryPlanWithKnownLeader(PhysicalPlan plan) {
    if (character == NodeCharacter.LEADER) {
      long startTime = Statistic.DATA_GROUP_MEMBER_LOCAL_EXECUTION.getOperationStartTime();
      TSStatus status = processPlanLocally(plan);
      Statistic.DATA_GROUP_MEMBER_LOCAL_EXECUTION.calOperationCostTimeFromStart(startTime);
      if (status != null) {
        return status;
      }
    } else if (leader.get() != null && !ClusterConstant.EMPTY_NODE.equals(leader.get())) {
      long startTime = Timer.Statistic.DATA_GROUP_MEMBER_FORWARD_PLAN.getOperationStartTime();
      TSStatus result = forwardPlan(plan, leader.get(), getHeader());
      Timer.Statistic.DATA_GROUP_MEMBER_FORWARD_PLAN.calOperationCostTimeFromStart(startTime);
      if (!StatusUtils.NO_LEADER.equals(result)) {
        result.setRedirectNode(
            new EndPoint(leader.get().getClientIp(), leader.get().getClientPort()));
        return result;
      }
    }
    return StatusUtils.NO_LEADER;
  }

  /**
   * When the node does not play a member in a group any more, the corresponding local data should
   * be removed.
   */
  public void removeLocalData(List<Integer> slots) {
    if (slots.isEmpty()) {
      return;
    }

    Set<Integer> slotSet = new HashSet<>(slots);
    List<PartialPath> allStorageGroupNames = IoTDB.metaManager.getAllStorageGroupPaths();
    TimePartitionFilter filter =
        (storageGroupName, timePartitionId) -> {
          int slot =
              SlotPartitionTable.getSlotStrategy()
                  .calculateSlotByPartitionNum(
                      storageGroupName, timePartitionId, ClusterConstant.SLOT_NUM);
          return slotSet.contains(slot);
        };
    for (PartialPath sg : allStorageGroupNames) {
      StorageEngine.getInstance().removePartitions(sg, filter);
    }
    for (Integer slot : slots) {
      slotManager.setToNull(slot);
    }

    if (logger.isInfoEnabled()) {
      logger.info(
          "{}: data of {} and other {} slots are removed", name, slots.get(0), slots.size() - 1);
    }
  }

  /**
   * When a node is removed and IT IS NOT THE HEADER of the group, the member should take over some
   * slots from the removed group, and add a new node to the group the removed node was in the
   * group.
   */
  @SuppressWarnings("java:S2445") // the reference of allNodes is unchanged
  public void removeNode(Node removedNode, NodeRemovalResult removalResult) {
    synchronized (allNodes) {
      if (allNodes.contains(removedNode)) {
        // update the group if the deleted node was in it
        allNodes = metaGroupMember.getPartitionTable().getHeaderGroup(getHeader());
        initPeerMap();
        if (removedNode.equals(leader.get())) {
          // if the leader is removed, also start an election immediately
          synchronized (term) {
            setCharacter(NodeCharacter.ELECTOR);
            setLastHeartbeatReceivedTime(Long.MIN_VALUE);
          }
        }
      }
      List<Integer> slotsToPull =
          ((SlotNodeRemovalResult) removalResult).getNewSlotOwners().get(getHeader());
      if (slotsToPull != null) {
        // pull the slots that should be taken over
        PullSnapshotTaskDescriptor taskDescriptor =
            new PullSnapshotTaskDescriptor(removalResult.getRemovedGroup(), slotsToPull, true);
        pullFileSnapshot(taskDescriptor, null);
      }
    }
  }

  /**
   * Generate a report containing the character, leader, term, last log term, last log index, header
   * and readOnly or not of this member.
   *
   * @return
   */
  public DataMemberReport genReport() {
    long prevLastLogIndex = lastReportedLogIndex;
    lastReportedLogIndex = logManager.getLastLogIndex();
    return new DataMemberReport(
        character,
        leader.get(),
        term.get(),
        logManager.getLastLogTerm(),
        lastReportedLogIndex,
        logManager.getCommitLogIndex(),
        logManager.getCommitLogTerm(),
        getHeader(),
        readOnly,
        NodeStatusManager.getINSTANCE().getLastResponseLatency(getHeader()),
        lastHeartbeatReceivedTime,
        prevLastLogIndex,
        logManager.getMaxHaveAppliedCommitIndex());
  }

  @TestOnly
  public void setMetaGroupMember(MetaGroupMember metaGroupMember) {
    this.metaGroupMember = metaGroupMember;
    this.localQueryExecutor = new LocalQueryExecutor(this);
  }

  @TestOnly
  void setLogManager(PartitionedSnapshotLogManager<Snapshot> logManager) {
    if (this.logManager != null) {
      this.logManager.close();
    }
    this.logManager = logManager;
    super.setLogManager(logManager);
    initPeerMap();
  }

  public SlotManager getSlotManager() {
    return slotManager;
  }

  public boolean onSnapshotInstalled(List<Integer> slots) {
    List<Integer> removableSlots = new ArrayList<>();
    for (Integer slot : slots) {
      int sentReplicaNum = slotManager.sentOneReplication(slot);
      if (sentReplicaNum >= ClusterDescriptor.getInstance().getConfig().getReplicationNum()) {
        removableSlots.add(slot);
      }
    }
    removeLocalData(removableSlots);
    return true;
  }

  public void registerPullSnapshotHint(PullSnapshotTaskDescriptor descriptor) {
    pullSnapshotHintService.registerHint(descriptor);
  }

  public LocalQueryExecutor getLocalQueryExecutor() {
    return localQueryExecutor;
  }

  @TestOnly
  public void setLocalQueryExecutor(LocalQueryExecutor localQueryExecutor) {
    this.localQueryExecutor = localQueryExecutor;
  }

  public boolean isUnchanged() {
    return unchanged;
  }

  public void setUnchanged(boolean unchanged) {
    this.unchanged = unchanged;
  }
}
