| /* |
| * 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.sync.SyncClientAdaptor; |
| import org.apache.iotdb.cluster.client.sync.SyncClientPool; |
| import org.apache.iotdb.cluster.config.ClusterConfig; |
| import org.apache.iotdb.cluster.config.ClusterConstant; |
| import org.apache.iotdb.cluster.config.ClusterDescriptor; |
| import org.apache.iotdb.cluster.exception.CheckConsistencyException; |
| import org.apache.iotdb.cluster.exception.LogExecutionException; |
| import org.apache.iotdb.cluster.exception.UnknownLogTypeException; |
| import org.apache.iotdb.cluster.log.CommitLogCallback; |
| import org.apache.iotdb.cluster.log.CommitLogTask; |
| import org.apache.iotdb.cluster.log.HardState; |
| import org.apache.iotdb.cluster.log.Log; |
| import org.apache.iotdb.cluster.log.LogDispatcher; |
| import org.apache.iotdb.cluster.log.LogDispatcher.SendLogRequest; |
| import org.apache.iotdb.cluster.log.LogParser; |
| import org.apache.iotdb.cluster.log.catchup.CatchUpTask; |
| import org.apache.iotdb.cluster.log.logtypes.PhysicalPlanLog; |
| import org.apache.iotdb.cluster.log.manage.RaftLogManager; |
| import org.apache.iotdb.cluster.rpc.thrift.AppendEntriesRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.AppendEntryRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.ElectionRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.ExecutNonQueryReq; |
| import org.apache.iotdb.cluster.rpc.thrift.HeartBeatRequest; |
| import org.apache.iotdb.cluster.rpc.thrift.HeartBeatResponse; |
| import org.apache.iotdb.cluster.rpc.thrift.Node; |
| import org.apache.iotdb.cluster.rpc.thrift.RaftService.AsyncClient; |
| import org.apache.iotdb.cluster.rpc.thrift.RaftService.Client; |
| import org.apache.iotdb.cluster.rpc.thrift.RequestCommitIndexResponse; |
| import org.apache.iotdb.cluster.server.NodeCharacter; |
| import org.apache.iotdb.cluster.server.RaftServer; |
| import org.apache.iotdb.cluster.server.Response; |
| import org.apache.iotdb.cluster.server.handlers.caller.AppendNodeEntryHandler; |
| import org.apache.iotdb.cluster.server.handlers.caller.GenericHandler; |
| 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.ClientUtils; |
| import org.apache.iotdb.cluster.utils.IOUtils; |
| import org.apache.iotdb.cluster.utils.PlanSerializer; |
| import org.apache.iotdb.cluster.utils.StatusUtils; |
| import org.apache.iotdb.db.exception.BatchProcessException; |
| import org.apache.iotdb.db.exception.IoTDBException; |
| import org.apache.iotdb.db.exception.metadata.IllegalPathException; |
| import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException; |
| import org.apache.iotdb.db.exception.metadata.PathNotExistException; |
| import org.apache.iotdb.db.exception.metadata.StorageGroupAlreadySetException; |
| import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException; |
| import org.apache.iotdb.db.qp.physical.PhysicalPlan; |
| import org.apache.iotdb.db.utils.TestOnly; |
| import org.apache.iotdb.rpc.RpcUtils; |
| import org.apache.iotdb.rpc.TSStatusCode; |
| import org.apache.iotdb.service.rpc.thrift.TSStatus; |
| |
| import com.google.common.util.concurrent.ThreadFactoryBuilder; |
| import org.apache.thrift.TException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.net.SocketTimeoutException; |
| import java.nio.BufferUnderflowException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.ConcurrentModificationException; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Random; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.ThreadPoolExecutor; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| /** |
| * RaftMember process the common raft logic like leader election, log appending, catch-up and so on. |
| */ |
| @SuppressWarnings("java:S3077") // reference volatile is enough |
| public abstract class RaftMember { |
| private static final Logger logger = LoggerFactory.getLogger(RaftMember.class); |
| public static final boolean USE_LOG_DISPATCHER = false; |
| |
| private static final String MSG_FORWARD_TIMEOUT = "{}: Forward {} to {} time out"; |
| private static final String MSG_FORWARD_ERROR = |
| "{}: encountered an error when forwarding {} to" + " {}"; |
| private static final String MSG_NO_LEADER_COMMIT_INDEX = |
| "{}: Cannot request commit index from {}"; |
| private static final String MSG_NO_LEADER_IN_SYNC = "{}: No leader is found when synchronizing"; |
| public static final String MSG_LOG_IS_ACCEPTED = "{}: log {} is accepted"; |
| /** |
| * when there is no leader, wait for waitLeaderTimeMs before return a NoLeader response to the |
| * client. |
| */ |
| private static long waitLeaderTimeMs = 60 * 1000L; |
| |
| /** |
| * when the leader of this node changes, the condition will be notified so other threads that wait |
| * on this may be woken. |
| */ |
| private final Object waitLeaderCondition = new Object(); |
| /** the lock is to make sure that only one thread can apply snapshot at the same time */ |
| private final Object snapshotApplyLock = new Object(); |
| |
| protected Node thisNode = ClusterConstant.EMPTY_NODE; |
| /** the nodes that belong to the same raft group as thisNode. */ |
| protected List<Node> allNodes; |
| |
| ClusterConfig config = ClusterDescriptor.getInstance().getConfig(); |
| /** the name of the member, to distinguish several members in the logs. */ |
| String name; |
| /** to choose nodes to send request of joining cluster randomly. */ |
| Random random = new Random(); |
| /** when the node is a leader, this map is used to track log progress of each follower. */ |
| Map<Node, Peer> peerMap; |
| /** |
| * the current term of the node, this object also works as lock of some transactions of the member |
| * like elections. |
| */ |
| AtomicLong term = new AtomicLong(0); |
| |
| volatile NodeCharacter character = NodeCharacter.ELECTOR; |
| AtomicReference<Node> leader = new AtomicReference<>(ClusterConstant.EMPTY_NODE); |
| /** |
| * the node that thisNode has voted for in this round of election, which prevents a node voting |
| * twice in a single election. |
| */ |
| volatile Node voteFor; |
| /** |
| * when this node is a follower, this records the unix-epoch timestamp when the last heartbeat |
| * arrived, and is reported in the timed member report to show how long the leader has been |
| * offline. |
| */ |
| volatile long lastHeartbeatReceivedTime; |
| /** the raft logs are all stored and maintained in the log manager */ |
| RaftLogManager logManager; |
| /** |
| * the single thread pool that runs the heartbeat thread, which send heartbeats to the follower |
| * when this node is a leader, or start elections when this node is an elector. |
| */ |
| ExecutorService heartBeatService; |
| /** |
| * if set to true, the node will reject all log appends when the header of a group is removed from |
| * the cluster, the members of the group should no longer accept writes, but they still can be |
| * candidates for weak consistency reads and provide snapshots for the new data holders |
| */ |
| volatile boolean readOnly = false; |
| /** |
| * lastLogIndex when generating the previous member report, to show the log ingestion rate of the |
| * member by comparing it with the current last log index. |
| */ |
| long lastReportedLogIndex; |
| /** the thread pool that runs catch-up tasks */ |
| private ExecutorService catchUpService; |
| /** |
| * lastCatchUpResponseTime records when is the latest response of each node's catch-up. There |
| * should be only one catch-up task for each node to avoid duplication, but the task may time out |
| * or the task may corrupt unexpectedly, and in that case, the next catch up should be enabled. So |
| * if we find a catch-up task that does not respond for long, we will start a new one instead of |
| * waiting for the previous one to finish. |
| */ |
| private Map<Node, Long> lastCatchUpResponseTime = new ConcurrentHashMap<>(); |
| /** |
| * the pool that provides reusable Thrift clients to connect to other RaftMembers and execute RPC |
| * requests. It will be initialized according to the implementation of the subclasses |
| */ |
| private AsyncClientPool asyncClientPool; |
| |
| private AsyncClientPool asyncSendLogClientPool; |
| private SyncClientPool syncClientPool; |
| private AsyncClientPool asyncHeartbeatClientPool; |
| private SyncClientPool syncHeartbeatClientPool; |
| /** |
| * when the commit progress is updated by a heartbeat, this object is notified so that we may know |
| * if this node is up-to-date with the leader, and whether the given consistency is reached |
| */ |
| private Object syncLock = new Object(); |
| /** |
| * when this node sends logs to the followers, the send is performed in parallel in this pool, so |
| * that a slow or unavailable node will not block other nodes. |
| */ |
| private ExecutorService appendLogThreadPool; |
| /** |
| * when using sync server, this thread pool is used to convert serial operations (like sending |
| * heartbeats and asking for votes) into paralleled ones, so the process will not be blocked by |
| * one slow node. |
| */ |
| private ExecutorService serialToParallelPool; |
| /** a thread pool that is used to do commit log tasks asynchronous in heartbeat thread */ |
| private ExecutorService commitLogPool; |
| /** |
| * logDispatcher buff the logs orderly according to their log indexes and send them sequentially, |
| * which avoids the followers receiving out-of-order logs, forcing them to wait for previous logs. |
| */ |
| private LogDispatcher logDispatcher; |
| |
| protected RaftMember() {} |
| |
| protected RaftMember( |
| String name, |
| AsyncClientPool asyncPool, |
| SyncClientPool syncPool, |
| AsyncClientPool asyncHeartbeatPool, |
| SyncClientPool syncHeartbeatPool) { |
| this.name = name; |
| this.asyncClientPool = asyncPool; |
| this.syncClientPool = syncPool; |
| this.asyncHeartbeatClientPool = asyncHeartbeatPool; |
| this.syncHeartbeatClientPool = syncHeartbeatPool; |
| this.asyncSendLogClientPool = asyncClientPool; |
| } |
| |
| protected RaftMember( |
| String name, |
| AsyncClientPool asyncPool, |
| SyncClientPool syncPool, |
| AsyncClientPool asyncHeartbeatPool, |
| SyncClientPool syncHeartbeatPool, |
| AsyncClientPool asyncSendLogClientPool) { |
| this.name = name; |
| this.asyncClientPool = asyncPool; |
| this.syncClientPool = syncPool; |
| this.asyncHeartbeatClientPool = asyncHeartbeatPool; |
| this.syncHeartbeatClientPool = syncHeartbeatPool; |
| this.asyncSendLogClientPool = asyncSendLogClientPool; |
| } |
| |
| /** |
| * Start the heartbeat thread and the catch-up thread pool. Calling the method twice does not |
| * induce side effects. |
| */ |
| public void start() { |
| if (heartBeatService != null) { |
| return; |
| } |
| |
| startBackGroundThreads(); |
| logger.info("{} started", name); |
| } |
| |
| void startBackGroundThreads() { |
| heartBeatService = |
| Executors.newSingleThreadScheduledExecutor( |
| r -> new Thread(r, name + "-HeartbeatThread@" + System.currentTimeMillis())); |
| catchUpService = |
| Executors.newCachedThreadPool( |
| new ThreadFactoryBuilder().setNameFormat(getName() + "-CatchUpThread%d").build()); |
| appendLogThreadPool = |
| Executors.newFixedThreadPool( |
| Runtime.getRuntime().availableProcessors() * 10, |
| new ThreadFactoryBuilder().setNameFormat(getName() + "-AppendLog%d").build()); |
| if (!config.isUseAsyncServer()) { |
| serialToParallelPool = |
| new ThreadPoolExecutor( |
| allNodes.size(), |
| Math.max(allNodes.size(), Runtime.getRuntime().availableProcessors()), |
| 1000L, |
| TimeUnit.MILLISECONDS, |
| new LinkedBlockingQueue<>(), |
| new ThreadFactoryBuilder().setNameFormat(getName() + "-SerialToParallel%d").build()); |
| } |
| commitLogPool = |
| Executors.newSingleThreadExecutor( |
| new ThreadFactoryBuilder().setNameFormat(getName() + "-CommitLog%d").build()); |
| } |
| |
| public String getName() { |
| return name; |
| } |
| |
| public RaftLogManager getLogManager() { |
| return logManager; |
| } |
| |
| @TestOnly |
| public void setLogManager(RaftLogManager logManager) { |
| if (this.logManager != null) { |
| this.logManager.close(); |
| } |
| this.logManager = logManager; |
| } |
| |
| /** |
| * Stop the heartbeat thread and the catch-up thread pool. Calling the method twice does not |
| * induce side effects. |
| */ |
| public void stop() { |
| closeLogManager(); |
| if (heartBeatService == null) { |
| return; |
| } |
| |
| heartBeatService.shutdownNow(); |
| catchUpService.shutdownNow(); |
| appendLogThreadPool.shutdownNow(); |
| try { |
| heartBeatService.awaitTermination(10, TimeUnit.SECONDS); |
| catchUpService.awaitTermination(10, TimeUnit.SECONDS); |
| appendLogThreadPool.awaitTermination(10, TimeUnit.SECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error( |
| "Unexpected interruption when waiting for heartBeatService and catchUpService " |
| + "to end", |
| e); |
| } |
| if (serialToParallelPool != null) { |
| serialToParallelPool.shutdownNow(); |
| try { |
| serialToParallelPool.awaitTermination(10, TimeUnit.SECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error("Unexpected interruption when waiting for asyncThreadPool to end", e); |
| } |
| } |
| |
| if (commitLogPool != null) { |
| commitLogPool.shutdownNow(); |
| try { |
| commitLogPool.awaitTermination(10, TimeUnit.SECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error("Unexpected interruption when waiting for commitLogPool to end", e); |
| } |
| } |
| catchUpService = null; |
| heartBeatService = null; |
| appendLogThreadPool = null; |
| logger.info("Member {} stopped", name); |
| } |
| |
| public void closeLogManager() { |
| if (logManager != null) { |
| logManager.close(); |
| } |
| } |
| |
| /** |
| * Process the HeartBeatRequest from the leader. If the term of the leader is smaller than the |
| * local term, reject the request by telling it the newest term. Else if the local logs are |
| * consistent with the leader's, commit them. Else help the leader find the last matched log. Also |
| * update the leadership, heartbeat timer and term of the local node. |
| */ |
| public HeartBeatResponse processHeartbeatRequest(HeartBeatRequest request) { |
| logger.trace("{} received a heartbeat", name); |
| synchronized (term) { |
| long thisTerm = term.get(); |
| long leaderTerm = request.getTerm(); |
| HeartBeatResponse response = new HeartBeatResponse(); |
| |
| if (leaderTerm < thisTerm) { |
| // a leader with a term lower than this node is invalid, send it the local term to inform |
| // it to resign |
| response.setTerm(thisTerm); |
| if (logger.isTraceEnabled()) { |
| logger.trace("{} received a heartbeat from a stale leader {}", name, request.getLeader()); |
| } |
| } else { |
| // try updating local term |
| stepDown(leaderTerm, true); |
| setLeader(request.getLeader()); |
| if (character != NodeCharacter.FOLLOWER) { |
| // interrupt current election |
| term.notifyAll(); |
| } |
| |
| // the heartbeat comes from a valid leader, process it with the sub-class logic |
| processValidHeartbeatReq(request, response); |
| |
| response.setTerm(Response.RESPONSE_AGREE); |
| // tell the leader who I am in case of catch-up |
| response.setFollower(thisNode); |
| |
| synchronized (logManager) { |
| // tell the leader the local log progress so it may decide whether to perform a catch up |
| response.setLastLogIndex(logManager.getLastLogIndex()); |
| response.setLastLogTerm(logManager.getLastLogTerm()); |
| } |
| |
| tryUpdateCommitIndex(leaderTerm, request.getCommitLogIndex(), request.getCommitLogTerm()); |
| |
| if (logger.isTraceEnabled()) { |
| logger.trace("{} received heartbeat from a valid leader {}", name, request.getLeader()); |
| } |
| } |
| return response; |
| } |
| } |
| |
| private void tryUpdateCommitIndex(long leaderTerm, long commitIndex, long commitTerm) { |
| if (leaderTerm >= term.get() && logManager.getCommitLogIndex() < commitIndex) { |
| // there are more local logs that can be committed, commit them in a ThreadPool so the |
| // heartbeat response will not be blocked |
| CommitLogTask commitLogTask = new CommitLogTask(logManager, commitIndex, commitTerm); |
| commitLogTask.registerCallback(new CommitLogCallback(this)); |
| // if the log is not consistent, the commitment will be blocked until the leader makes the |
| // node catch up |
| if (commitLogPool != null && !commitLogPool.isShutdown()) { |
| commitLogPool.submit(commitLogTask); |
| } |
| |
| logger.debug( |
| "{}: Inconsistent log found, leaderCommit: {}-{}, localCommit: {}-{}, " |
| + "localLast: {}-{}", |
| name, |
| commitIndex, |
| commitTerm, |
| logManager.getCommitLogIndex(), |
| logManager.getCommitLogTerm(), |
| logManager.getLastLogIndex(), |
| logManager.getLastLogTerm()); |
| } |
| } |
| |
| /** |
| * Process an ElectionRequest. If the request comes from the last leader, accept it. Else decide |
| * whether to accept by examining the log status of the elector. |
| */ |
| public long processElectionRequest(ElectionRequest electionRequest) { |
| synchronized (term) { |
| long currentTerm = term.get(); |
| long response = |
| checkElectorTerm(currentTerm, electionRequest.getTerm(), electionRequest.getElector()); |
| if (response != Response.RESPONSE_AGREE) { |
| return response; |
| } |
| |
| // compare the log progress of the elector with this node |
| response = checkElectorLogProgress(electionRequest); |
| logger.info( |
| "{} sending response {} to the elector {}", name, response, electionRequest.getElector()); |
| return response; |
| } |
| } |
| |
| private long checkElectorTerm(long currentTerm, long electorTerm, Node elector) { |
| if (electorTerm < currentTerm) { |
| // the elector has a smaller term thus the request is invalid |
| logger.info( |
| "{} sending localTerm {} to the elector {} because it's term {} is smaller.", |
| name, |
| currentTerm, |
| elector, |
| electorTerm); |
| return currentTerm; |
| } |
| if (currentTerm == electorTerm && voteFor != null && !Objects.equals(voteFor, elector)) { |
| // this node has voted in this round, but not for the elector, as one node cannot vote |
| // twice, reject the request |
| logger.info( |
| "{} sending rejection to the elector {} because member already has voted {} in this term {}.", |
| name, |
| elector, |
| voteFor, |
| currentTerm); |
| return Response.RESPONSE_REJECT; |
| } |
| if (electorTerm > currentTerm) { |
| // the elector has a larger term, this node should update its term first |
| logger.info( |
| "{} received an election from elector {} which has bigger term {} than localTerm {}, raftMember should step down first and then continue to decide whether to grant it's vote by log status.", |
| name, |
| elector, |
| electorTerm, |
| currentTerm); |
| stepDown(electorTerm, false); |
| } |
| return Response.RESPONSE_AGREE; |
| } |
| |
| /** |
| * Process an AppendEntryRequest. First check the term of the leader, then parse the log and |
| * finally see if we can find a position to append the log. |
| */ |
| public long appendEntry(AppendEntryRequest request) throws UnknownLogTypeException { |
| logger.debug("{} received an AppendEntryRequest: {}", name, request); |
| // the term checked here is that of the leader, not that of the log |
| long checkResult = checkRequestTerm(request.term, request.leader); |
| if (checkResult != Response.RESPONSE_AGREE) { |
| return checkResult; |
| } |
| |
| long startTime = Timer.Statistic.RAFT_RECEIVER_LOG_PARSE.getOperationStartTime(); |
| int logByteSize = request.getEntry().length; |
| Log log = LogParser.getINSTANCE().parse(request.entry); |
| log.setByteSize(logByteSize); |
| Timer.Statistic.RAFT_RECEIVER_LOG_PARSE.calOperationCostTimeFromStart(startTime); |
| |
| long result = appendEntry(request.prevLogIndex, request.prevLogTerm, request.leaderCommit, log); |
| logger.debug("{} AppendEntryRequest of {} completed with result {}", name, log, result); |
| |
| return result; |
| } |
| |
| /** Similar to appendEntry, while the incoming load is batch of logs instead of a single log. */ |
| public long appendEntries(AppendEntriesRequest request) throws UnknownLogTypeException { |
| logger.debug("{} received an AppendEntriesRequest", name); |
| |
| // the term checked here is that of the leader, not that of the log |
| long checkResult = checkRequestTerm(request.term, request.leader); |
| if (checkResult != Response.RESPONSE_AGREE) { |
| return checkResult; |
| } |
| |
| long response; |
| List<Log> logs = new ArrayList<>(); |
| int logByteSize = 0; |
| long startTime = Timer.Statistic.RAFT_RECEIVER_LOG_PARSE.getOperationStartTime(); |
| for (ByteBuffer buffer : request.getEntries()) { |
| buffer.mark(); |
| Log log; |
| logByteSize = buffer.limit() - buffer.position(); |
| try { |
| log = LogParser.getINSTANCE().parse(buffer); |
| log.setByteSize(logByteSize); |
| } catch (BufferUnderflowException e) { |
| buffer.reset(); |
| throw e; |
| } |
| logs.add(log); |
| } |
| |
| Timer.Statistic.RAFT_RECEIVER_LOG_PARSE.calOperationCostTimeFromStart(startTime); |
| |
| response = appendEntries(request.prevLogIndex, request.prevLogTerm, request.leaderCommit, logs); |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{} AppendEntriesRequest of log size {} completed with result {}", |
| name, |
| request.getEntries().size(), |
| response); |
| } |
| return response; |
| } |
| |
| /** |
| * Get an asynchronous heartbeat thrift client to the given node. |
| * |
| * @return an asynchronous thrift client or null if the caller tries to connect the local node. |
| */ |
| public AsyncClient getAsyncHeartbeatClient(Node node) { |
| return getAsyncClient(node, asyncHeartbeatClientPool, false); |
| } |
| |
| /** |
| * NOTICE: client.putBack() must be called after use. |
| * |
| * @return the heartbeat client for the node |
| */ |
| public Client getSyncHeartbeatClient(Node node) { |
| return getSyncClient(syncHeartbeatClientPool, node, false); |
| } |
| |
| public void sendLogAsync( |
| Log log, |
| AtomicInteger voteCounter, |
| Node node, |
| AtomicBoolean leaderShipStale, |
| AtomicLong newLeaderTerm, |
| AppendEntryRequest request, |
| Peer peer) { |
| AsyncClient client = getSendLogAsyncClient(node); |
| if (client != null) { |
| AppendNodeEntryHandler handler = |
| getAppendNodeEntryHandler(log, voteCounter, node, leaderShipStale, newLeaderTerm, peer); |
| try { |
| client.appendEntry(request, handler); |
| logger.debug("{} sending a log to {}: {}", name, node, log); |
| } catch (Exception e) { |
| logger.warn("{} cannot append log to node {}", name, node, e); |
| } |
| } |
| } |
| |
| private Client getSyncClient(SyncClientPool pool, Node node, boolean activatedOnly) { |
| if (ClusterConstant.EMPTY_NODE.equals(node) || node == null) { |
| return null; |
| } |
| |
| return pool.getClient(node, activatedOnly); |
| } |
| |
| public NodeCharacter getCharacter() { |
| return character; |
| } |
| |
| public void setCharacter(NodeCharacter character) { |
| if (!Objects.equals(character, this.character)) { |
| logger.info("{} has become a {}", name, character); |
| this.character = character; |
| } |
| } |
| |
| public long getLastHeartbeatReceivedTime() { |
| return lastHeartbeatReceivedTime; |
| } |
| |
| public void setLastHeartbeatReceivedTime(long lastHeartbeatReceivedTime) { |
| this.lastHeartbeatReceivedTime = lastHeartbeatReceivedTime; |
| } |
| |
| public Node getLeader() { |
| return leader.get(); |
| } |
| |
| public void setLeader(Node leader) { |
| if (!Objects.equals(leader, this.leader.get())) { |
| if (ClusterConstant.EMPTY_NODE.equals(leader) || leader == null) { |
| logger.info("{} has been set to null in term {}", getName(), term.get()); |
| } else if (!Objects.equals(leader, this.thisNode)) { |
| logger.info("{} has become a follower of {} in term {}", getName(), leader, term.get()); |
| } |
| synchronized (waitLeaderCondition) { |
| if (leader == null) { |
| this.leader.set(ClusterConstant.EMPTY_NODE); |
| } else { |
| this.leader.set(leader); |
| } |
| if (!ClusterConstant.EMPTY_NODE.equals(this.leader.get())) { |
| waitLeaderCondition.notifyAll(); |
| } |
| } |
| } |
| } |
| |
| public Collection<Node> getAllNodes() { |
| return allNodes; |
| } |
| |
| public void setAllNodes(List<Node> allNodes) { |
| this.allNodes = allNodes; |
| } |
| |
| public Map<Node, Long> getLastCatchUpResponseTime() { |
| return lastCatchUpResponseTime; |
| } |
| |
| /** Sub-classes will add their own process of HeartBeatResponse in this method. */ |
| public void processValidHeartbeatResp(HeartBeatResponse response, Node receiver) {} |
| |
| /** The actions performed when the node wins in an election (becoming a leader). */ |
| public void onElectionWins() {} |
| |
| /** |
| * Update the followers' log by sending logs whose index >= followerLastMatchedLogIndex to the |
| * follower. If some of the required logs are removed, also send the snapshot. <br> |
| * notice that if a part of data is in the snapshot, then it is not in the logs. |
| */ |
| public void catchUp(Node follower, long lastLogIdx) { |
| // for one follower, there is at most one ongoing catch-up, so the same data will not be sent |
| // twice to the node |
| synchronized (catchUpService) { |
| // check if the last catch-up is still ongoing and does not time out yet |
| Long lastCatchupResp = lastCatchUpResponseTime.get(follower); |
| if (lastCatchupResp != null |
| && System.currentTimeMillis() - lastCatchupResp < config.getCatchUpTimeoutMS()) { |
| logger.debug("{}: last catch up of {} is ongoing", name, follower); |
| return; |
| } else { |
| // record the start of the catch-up |
| lastCatchUpResponseTime.put(follower, System.currentTimeMillis()); |
| } |
| } |
| logger.info("{}: Start to make {} catch up", name, follower); |
| if (!catchUpService.isShutdown()) { |
| Future<?> future = |
| catchUpService.submit(new CatchUpTask(follower, peerMap.get(follower), this, lastLogIdx)); |
| catchUpService.submit( |
| () -> { |
| try { |
| future.get(); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } catch (ExecutionException e) { |
| logger.error("{}: Catch up task exits with unexpected exception", name, e); |
| } |
| }); |
| } |
| } |
| |
| /** |
| * If the node is not a leader, the request will be sent to the leader or reports an error if |
| * there is no leader. Otherwise execute the plan locally (whether to send it to followers depends |
| * on the type of the plan). |
| */ |
| public TSStatus executeNonQueryPlan(ExecutNonQueryReq request) |
| throws IOException, IllegalPathException { |
| // process the plan locally |
| PhysicalPlan plan = PhysicalPlan.Factory.create(request.planBytes); |
| |
| TSStatus answer = executeNonQueryPlan(plan); |
| logger.debug("{}: Received a plan {}, executed answer: {}", name, plan, answer); |
| return answer; |
| } |
| |
| /** |
| * Execute a non-query plan. Subclass may have their individual implements. |
| * |
| * @param plan a non-query plan. |
| * @return A TSStatus indicating the execution result. |
| */ |
| abstract TSStatus executeNonQueryPlan(PhysicalPlan plan); |
| |
| /** |
| * according to the consistency configuration, decide whether to execute syncLeader or not and |
| * throws exception when failed. Note that the write request will always try to sync leader |
| */ |
| public void syncLeaderWithConsistencyCheck(boolean isWriteRequest) |
| throws CheckConsistencyException { |
| if (isWriteRequest) { |
| syncLeader(new StrongCheckConsistency()); |
| } else { |
| switch (config.getConsistencyLevel()) { |
| case STRONG_CONSISTENCY: |
| syncLeader(new StrongCheckConsistency()); |
| return; |
| case MID_CONSISTENCY: |
| // if leaderCommitId bigger than localAppliedId a value, |
| // will throw CHECK_MID_CONSISTENCY_EXCEPTION |
| syncLeader(new MidCheckConsistency()); |
| return; |
| case WEAK_CONSISTENCY: |
| // do nothing |
| return; |
| default: |
| // this should not happen in theory |
| throw new CheckConsistencyException( |
| "unknown consistency=" + config.getConsistencyLevel().name()); |
| } |
| } |
| } |
| |
| /** call back after syncLeader */ |
| public interface CheckConsistency { |
| |
| /** |
| * deal leaderCommitId and localAppliedId after syncLeader |
| * |
| * @param leaderCommitId leader commit id |
| * @param localAppliedId local applied id |
| * @throws CheckConsistencyException maybe throw CheckConsistencyException, which is defined in |
| * implements. |
| */ |
| void postCheckConsistency(long leaderCommitId, long localAppliedId) |
| throws CheckConsistencyException; |
| } |
| |
| public static class MidCheckConsistency implements CheckConsistency { |
| |
| /** |
| * if leaderCommitId - localAppliedId > MaxReadLogLag, will throw |
| * CHECK_MID_CONSISTENCY_EXCEPTION |
| * |
| * @param leaderCommitId leader commit id |
| * @param localAppliedId local applied id |
| * @throws CheckConsistencyException |
| */ |
| @Override |
| public void postCheckConsistency(long leaderCommitId, long localAppliedId) |
| throws CheckConsistencyException { |
| if (leaderCommitId == Long.MAX_VALUE |
| || leaderCommitId == Long.MIN_VALUE |
| || leaderCommitId - localAppliedId |
| > ClusterDescriptor.getInstance().getConfig().getMaxReadLogLag()) { |
| throw CheckConsistencyException.CHECK_MID_CONSISTENCY_EXCEPTION; |
| } |
| } |
| } |
| |
| public static class StrongCheckConsistency implements CheckConsistency { |
| |
| /** |
| * if leaderCommitId > localAppliedId, will throw CHECK_STRONG_CONSISTENCY_EXCEPTION |
| * |
| * @param leaderCommitId leader commit id |
| * @param localAppliedId local applied id |
| * @throws CheckConsistencyException |
| */ |
| @Override |
| public void postCheckConsistency(long leaderCommitId, long localAppliedId) |
| throws CheckConsistencyException { |
| if (leaderCommitId > localAppliedId |
| || leaderCommitId == Long.MAX_VALUE |
| || leaderCommitId == Long.MIN_VALUE) { |
| throw CheckConsistencyException.CHECK_STRONG_CONSISTENCY_EXCEPTION; |
| } |
| } |
| } |
| |
| /** |
| * Request and check the leader's commitId to see whether this node has caught up. If not, wait |
| * until this node catches up. |
| * |
| * @param checkConsistency check after syncleader |
| * @return true if the node has caught up, false otherwise |
| * @throws CheckConsistencyException if leaderCommitId bigger than localAppliedId a threshold |
| * value after timeout |
| */ |
| public boolean syncLeader(CheckConsistency checkConsistency) throws CheckConsistencyException { |
| if (character == NodeCharacter.LEADER) { |
| return true; |
| } |
| waitLeader(); |
| if (leader.get() == null || ClusterConstant.EMPTY_NODE.equals(leader.get())) { |
| // the leader has not been elected, we must assume the node falls behind |
| logger.warn(MSG_NO_LEADER_IN_SYNC, name); |
| return false; |
| } |
| if (character == NodeCharacter.LEADER) { |
| return true; |
| } |
| logger.debug("{}: try synchronizing with the leader {}", name, leader.get()); |
| return waitUntilCatchUp(checkConsistency); |
| } |
| |
| /** Wait until the leader of this node becomes known or time out. */ |
| public void waitLeader() { |
| long startTime = System.currentTimeMillis(); |
| while (leader.get() == null || ClusterConstant.EMPTY_NODE.equals(leader.get())) { |
| synchronized (waitLeaderCondition) { |
| try { |
| waitLeaderCondition.wait(10); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error("Unexpected interruption when waiting for a leader", e); |
| } |
| } |
| long consumedTime = System.currentTimeMillis() - startTime; |
| if (consumedTime >= getWaitLeaderTimeMs()) { |
| logger.warn("{}: leader is still offline after {}ms", name, consumedTime); |
| break; |
| } |
| } |
| logger.debug("{}: current leader is {}", name, leader.get()); |
| } |
| |
| /** |
| * Request the leader's commit index and wait until the local commit index becomes not less than |
| * it. |
| * |
| * @return true if this node has caught up before timeout, false otherwise |
| * @throws CheckConsistencyException if leaderCommitId bigger than localAppliedId a threshold |
| * value after timeout |
| */ |
| protected boolean waitUntilCatchUp(CheckConsistency checkConsistency) |
| throws CheckConsistencyException { |
| long leaderCommitId = Long.MIN_VALUE; |
| RequestCommitIndexResponse response; |
| try { |
| response = config.isUseAsyncServer() ? requestCommitIdAsync() : requestCommitIdSync(); |
| leaderCommitId = response.getCommitLogIndex(); |
| |
| tryUpdateCommitIndex( |
| response.getTerm(), response.getCommitLogIndex(), response.getCommitLogTerm()); |
| |
| return syncLocalApply(leaderCommitId); |
| } catch (TException e) { |
| logger.error(MSG_NO_LEADER_COMMIT_INDEX, name, leader.get(), e); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error(MSG_NO_LEADER_COMMIT_INDEX, name, leader.get(), e); |
| } finally { |
| if (checkConsistency != null) { |
| checkConsistency.postCheckConsistency( |
| leaderCommitId, logManager.getMaxHaveAppliedCommitIndex()); |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * sync local applyId to leader commitId |
| * |
| * @param leaderCommitId leader commit id |
| * @return true if leaderCommitId <= localAppliedId |
| */ |
| private boolean syncLocalApply(long leaderCommitId) { |
| long startTime = System.currentTimeMillis(); |
| long waitedTime = 0; |
| long localAppliedId = 0; |
| while (waitedTime < RaftServer.getSyncLeaderMaxWaitMs()) { |
| try { |
| localAppliedId = logManager.getMaxHaveAppliedCommitIndex(); |
| logger.debug("{}: synchronizing commitIndex {}/{}", name, localAppliedId, leaderCommitId); |
| if (leaderCommitId <= localAppliedId) { |
| // this node has caught up |
| if (logger.isDebugEnabled()) { |
| waitedTime = System.currentTimeMillis() - startTime; |
| logger.debug("{}: synchronized with the leader after {}ms", name, waitedTime); |
| } |
| return true; |
| } |
| // wait for next heartbeat to catch up |
| // the local node will not perform a commit here according to the leaderCommitId because |
| // the node may have some inconsistent logs with the leader |
| waitedTime = System.currentTimeMillis() - startTime; |
| synchronized (syncLock) { |
| syncLock.wait(RaftServer.getHeartBeatIntervalMs()); |
| } |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.error(MSG_NO_LEADER_COMMIT_INDEX, name, leader.get(), e); |
| } |
| } |
| logger.warn("{}: Failed to synchronize with the leader after {}ms", name, waitedTime); |
| |
| return false; |
| } |
| |
| /** |
| * Create a log for "plan" and append it locally and to all followers. Only the group leader can |
| * call this method. Will commit the log locally and send it to followers |
| * |
| * @return OK if over half of the followers accept the log or null if the leadership is lost |
| * during the appending |
| */ |
| TSStatus processPlanLocally(PhysicalPlan plan) { |
| if (USE_LOG_DISPATCHER) { |
| return processPlanLocallyV2(plan); |
| } |
| |
| logger.debug("{}: Processing plan {}", name, plan); |
| if (readOnly) { |
| return StatusUtils.NODE_READ_ONLY; |
| } |
| long startTime = Timer.Statistic.RAFT_SENDER_APPEND_LOG.getOperationStartTime(); |
| PhysicalPlanLog log = new PhysicalPlanLog(); |
| // assign term and index to the new log and append it |
| synchronized (logManager) { |
| log.setCurrLogTerm(getTerm().get()); |
| log.setCurrLogIndex(logManager.getLastLogIndex() + 1); |
| |
| log.setPlan(plan); |
| plan.setIndex(log.getCurrLogIndex()); |
| // appendLogInGroup will serialize log, and set log size, and we will use the size after it |
| logManager.append(log); |
| } |
| Timer.Statistic.RAFT_SENDER_APPEND_LOG.calOperationCostTimeFromStart(startTime); |
| |
| try { |
| if (appendLogInGroup(log)) { |
| return StatusUtils.OK; |
| } |
| } catch (LogExecutionException e) { |
| return handleLogExecutionException(log, e); |
| } |
| return StatusUtils.TIME_OUT; |
| } |
| |
| private TSStatus processPlanLocallyV2(PhysicalPlan plan) { |
| logger.debug("{}: Processing plan {}", name, plan); |
| if (readOnly) { |
| return StatusUtils.NODE_READ_ONLY; |
| } |
| PhysicalPlanLog log = new PhysicalPlanLog(); |
| // assign term and index to the new log and append it |
| SendLogRequest sendLogRequest; |
| |
| long startTime = |
| Statistic.RAFT_SENDER_COMPETE_LOG_MANAGER_BEFORE_APPEND_V2.getOperationStartTime(); |
| synchronized (logManager) { |
| Statistic.RAFT_SENDER_COMPETE_LOG_MANAGER_BEFORE_APPEND_V2.calOperationCostTimeFromStart( |
| startTime); |
| |
| log.setCurrLogTerm(getTerm().get()); |
| log.setCurrLogIndex(logManager.getLastLogIndex() + 1); |
| log.setPlan(plan); |
| plan.setIndex(log.getCurrLogIndex()); |
| |
| startTime = Timer.Statistic.RAFT_SENDER_APPEND_LOG_V2.getOperationStartTime(); |
| // logDispatcher will serialize log, and set log size, and we will use the size after it |
| logManager.append(log); |
| Timer.Statistic.RAFT_SENDER_APPEND_LOG_V2.calOperationCostTimeFromStart(startTime); |
| |
| startTime = Statistic.RAFT_SENDER_BUILD_LOG_REQUEST.getOperationStartTime(); |
| sendLogRequest = buildSendLogRequest(log); |
| Statistic.RAFT_SENDER_BUILD_LOG_REQUEST.calOperationCostTimeFromStart(startTime); |
| |
| startTime = Statistic.RAFT_SENDER_OFFER_LOG.getOperationStartTime(); |
| log.setCreateTime(System.nanoTime()); |
| getLogDispatcher().offer(sendLogRequest); |
| Statistic.RAFT_SENDER_OFFER_LOG.calOperationCostTimeFromStart(startTime); |
| } |
| |
| try { |
| AppendLogResult appendLogResult = |
| waitAppendResult( |
| sendLogRequest.getVoteCounter(), |
| sendLogRequest.getLeaderShipStale(), |
| sendLogRequest.getNewLeaderTerm()); |
| Timer.Statistic.RAFT_SENDER_LOG_FROM_CREATE_TO_ACCEPT.calOperationCostTimeFromStart( |
| sendLogRequest.getLog().getCreateTime()); |
| switch (appendLogResult) { |
| case OK: |
| logger.debug(MSG_LOG_IS_ACCEPTED, name, log); |
| startTime = Timer.Statistic.RAFT_SENDER_COMMIT_LOG.getOperationStartTime(); |
| commitLog(log); |
| Timer.Statistic.RAFT_SENDER_COMMIT_LOG.calOperationCostTimeFromStart(startTime); |
| return StatusUtils.OK; |
| case TIME_OUT: |
| logger.debug("{}: log {} timed out...", name, log); |
| break; |
| case LEADERSHIP_STALE: |
| // abort the appending, the new leader will fix the local logs by catch-up |
| default: |
| break; |
| } |
| } catch (LogExecutionException e) { |
| return handleLogExecutionException(log, e); |
| } |
| return StatusUtils.TIME_OUT; |
| } |
| |
| public SendLogRequest buildSendLogRequest(Log log) { |
| AtomicInteger voteCounter = new AtomicInteger(allNodes.size() / 2); |
| AtomicBoolean leaderShipStale = new AtomicBoolean(false); |
| AtomicLong newLeaderTerm = new AtomicLong(term.get()); |
| |
| long startTime = Statistic.RAFT_SENDER_BUILD_APPEND_REQUEST.getOperationStartTime(); |
| AppendEntryRequest appendEntryRequest = buildAppendEntryRequest(log, false); |
| Statistic.RAFT_SENDER_BUILD_APPEND_REQUEST.calOperationCostTimeFromStart(startTime); |
| |
| return new SendLogRequest(log, voteCounter, leaderShipStale, newLeaderTerm, appendEntryRequest); |
| } |
| |
| /** |
| * The maximum time to wait if there is no leader in the group, after which a |
| * LeadNotFoundException will be thrown. |
| */ |
| static long getWaitLeaderTimeMs() { |
| return waitLeaderTimeMs; |
| } |
| |
| static void setWaitLeaderTimeMs(long waitLeaderTimeMs) { |
| RaftMember.waitLeaderTimeMs = waitLeaderTimeMs; |
| } |
| |
| @SuppressWarnings("java:S2274") // enable timeout |
| protected RequestCommitIndexResponse requestCommitIdAsync() |
| throws TException, InterruptedException { |
| // use Long.MAX_VALUE to indicate a timeout |
| RequestCommitIndexResponse response = |
| new RequestCommitIndexResponse(Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE); |
| AtomicReference<RequestCommitIndexResponse> commitIdResult = new AtomicReference<>(response); |
| AsyncClient client = getAsyncClient(leader.get()); |
| if (client == null) { |
| // cannot connect to the leader |
| logger.warn(MSG_NO_LEADER_IN_SYNC, name); |
| return commitIdResult.get(); |
| } |
| synchronized (commitIdResult) { |
| client.requestCommitIndex(getHeader(), new GenericHandler<>(leader.get(), commitIdResult)); |
| commitIdResult.wait(RaftServer.getSyncLeaderMaxWaitMs()); |
| } |
| return commitIdResult.get(); |
| } |
| |
| private RequestCommitIndexResponse requestCommitIdSync() throws TException { |
| Client client = getSyncClient(leader.get()); |
| RequestCommitIndexResponse response; |
| if (client == null) { |
| // cannot connect to the leader |
| logger.warn(MSG_NO_LEADER_IN_SYNC, name); |
| // use Long.MAX_VALUE to indicate a timeouts |
| response = new RequestCommitIndexResponse(Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE); |
| return response; |
| } |
| try { |
| response = client.requestCommitIndex(getHeader()); |
| } catch (TException e) { |
| client.getInputProtocol().getTransport().close(); |
| throw e; |
| } finally { |
| ClientUtils.putBackSyncClient(client); |
| } |
| return response; |
| } |
| |
| /** |
| * Tell the requester the current commit index if the local node is the leader of the group headed |
| * by header. Or forward it to the leader. Otherwise report an error. |
| * |
| * @return Long.MIN_VALUE if the node is not a leader, or the commitIndex |
| */ |
| public long getCommitIndex() { |
| if (character == NodeCharacter.LEADER) { |
| return logManager.getCommitLogIndex(); |
| } else { |
| return Long.MIN_VALUE; |
| } |
| } |
| |
| public void setReadOnly() { |
| synchronized (logManager) { |
| readOnly = true; |
| } |
| } |
| |
| public boolean isReadOnly() { |
| return readOnly; |
| } |
| |
| public void initPeerMap() { |
| peerMap = new ConcurrentHashMap<>(); |
| for (Node entry : allNodes) { |
| peerMap.computeIfAbsent(entry, k -> new Peer(logManager.getLastLogIndex())); |
| } |
| } |
| |
| public Map<Node, Peer> getPeerMap() { |
| return peerMap; |
| } |
| |
| /** @return true if there is a log whose index is "index" and term is "term", false otherwise */ |
| public boolean matchLog(long index, long term) { |
| boolean matched = logManager.matchTerm(term, index); |
| logger.debug("Log {}-{} matched: {}", index, term, matched); |
| return matched; |
| } |
| |
| public ExecutorService getSerialToParallelPool() { |
| return serialToParallelPool; |
| } |
| |
| public Object getSyncLock() { |
| return syncLock; |
| } |
| |
| /** Sub-classes will add their own process of HeartBeatRequest in this method. */ |
| void processValidHeartbeatReq(HeartBeatRequest request, HeartBeatResponse response) {} |
| |
| /** |
| * Verify the validity of an ElectionRequest, and make itself a follower of the elector if the |
| * request is valid. |
| * |
| * @return Response.RESPONSE_AGREE if the elector is valid or the local term if the elector has a |
| * smaller term or Response.RESPONSE_LOG_MISMATCH if the elector has older logs. |
| */ |
| long checkElectorLogProgress(ElectionRequest electionRequest) { |
| |
| long thatTerm = electionRequest.getTerm(); |
| long thatLastLogIndex = electionRequest.getLastLogIndex(); |
| long thatLastLogTerm = electionRequest.getLastLogTerm(); |
| |
| // check the log progress of the elector |
| long resp = checkLogProgress(thatLastLogIndex, thatLastLogTerm); |
| if (resp == Response.RESPONSE_AGREE) { |
| logger.info( |
| "{} accepted an election request, term:{}/{}, logIndex:{}/{}, logTerm:{}/{}", |
| name, |
| thatTerm, |
| term.get(), |
| thatLastLogIndex, |
| logManager.getLastLogIndex(), |
| thatLastLogTerm, |
| logManager.getLastLogTerm()); |
| setCharacter(NodeCharacter.FOLLOWER); |
| lastHeartbeatReceivedTime = System.currentTimeMillis(); |
| setVoteFor(electionRequest.getElector()); |
| updateHardState(thatTerm, getVoteFor()); |
| } else { |
| logger.info( |
| "{} rejected an election request, term:{}/{}, logIndex:{}/{}, logTerm:{}/{}", |
| name, |
| thatTerm, |
| term.get(), |
| thatLastLogIndex, |
| logManager.getLastLogIndex(), |
| thatLastLogTerm, |
| logManager.getLastLogTerm()); |
| } |
| return resp; |
| } |
| |
| /** |
| * Reject the election if the lastLogTerm of the candidate equals to the voter's but its |
| * lastLogIndex is smaller than the voter's Otherwise accept the election. |
| * |
| * @return Response.RESPONSE_AGREE if the elector is valid or the local term if the elector has a |
| * smaller term or Response.RESPONSE_LOG_MISMATCH if the elector has older logs. |
| */ |
| long checkLogProgress(long lastLogIndex, long lastLogTerm) { |
| long response; |
| synchronized (logManager) { |
| if (logManager.isLogUpToDate(lastLogTerm, lastLogIndex)) { |
| response = Response.RESPONSE_AGREE; |
| } else { |
| response = Response.RESPONSE_LOG_MISMATCH; |
| } |
| } |
| return response; |
| } |
| |
| /** |
| * Forward a non-query plan to a node using the default client. |
| * |
| * @param plan a non-query plan |
| * @param node cannot be the local node |
| * @param header must be set for data group communication, set to null for meta group |
| * communication |
| * @return a TSStatus indicating if the forwarding is successful. |
| */ |
| public TSStatus forwardPlan(PhysicalPlan plan, Node node, Node header) { |
| if (node == null || node.equals(thisNode)) { |
| logger.debug("{}: plan {} has no where to be forwarded", name, plan); |
| return StatusUtils.NO_LEADER; |
| } |
| logger.debug("{}: Forward {} to node {}", name, plan, node); |
| |
| TSStatus status; |
| if (config.isUseAsyncServer()) { |
| status = forwardPlanAsync(plan, node, header); |
| } else { |
| status = forwardPlanSync(plan, node, header); |
| } |
| if (status.getCode() == TSStatusCode.NO_CONNECTION.getStatusCode() |
| && (header == null || header.equals(getHeader())) |
| && (leader.get() != null) |
| && leader.get().equals(node)) { |
| // leader is down, trigger a new election by resetting heartbeat |
| lastHeartbeatReceivedTime = -1; |
| leader.set(null); |
| waitLeader(); |
| } |
| return status; |
| } |
| |
| /** |
| * Forward a non-query plan to "receiver" using "client". |
| * |
| * @param plan a non-query plan |
| * @param header to determine which DataGroupMember of "receiver" will process the request. |
| * @return a TSStatus indicating if the forwarding is successful. |
| */ |
| private TSStatus forwardPlanAsync(PhysicalPlan plan, Node receiver, Node header) { |
| AsyncClient client = getAsyncClient(receiver); |
| if (client == null) { |
| logger.debug("{}: can not get client for node={}", name, receiver); |
| return StatusUtils.NO_CONNECTION |
| .deepCopy() |
| .setMessage(String.format("%s cannot be reached", receiver)); |
| } |
| return forwardPlanAsync(plan, receiver, header, client); |
| } |
| |
| public TSStatus forwardPlanAsync( |
| PhysicalPlan plan, Node receiver, Node header, AsyncClient client) { |
| try { |
| TSStatus tsStatus = SyncClientAdaptor.executeNonQuery(client, plan, header, receiver); |
| if (tsStatus == null) { |
| tsStatus = StatusUtils.TIME_OUT; |
| logger.warn(MSG_FORWARD_TIMEOUT, name, plan, receiver); |
| } |
| return tsStatus; |
| } catch (IOException | TException e) { |
| logger.error(MSG_FORWARD_ERROR, name, plan, receiver, e); |
| return StatusUtils.getStatus(StatusUtils.INTERNAL_ERROR, e.getMessage()); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.warn("{}: forward {} to {} interrupted", name, plan, receiver); |
| return StatusUtils.TIME_OUT; |
| } |
| } |
| |
| private TSStatus forwardPlanSync(PhysicalPlan plan, Node receiver, Node header) { |
| Client client = getSyncClient(receiver); |
| if (client == null) { |
| logger.warn(MSG_FORWARD_TIMEOUT, name, plan, receiver); |
| return StatusUtils.TIME_OUT; |
| } |
| return forwardPlanSync(plan, receiver, header, client); |
| } |
| |
| public TSStatus forwardPlanSync(PhysicalPlan plan, Node receiver, Node header, Client client) { |
| try { |
| ExecutNonQueryReq req = new ExecutNonQueryReq(); |
| req.setPlanBytes(PlanSerializer.getInstance().serialize(plan)); |
| if (header != null) { |
| req.setHeader(header); |
| } |
| |
| TSStatus tsStatus = client.executeNonQueryPlan(req); |
| if (tsStatus == null) { |
| tsStatus = StatusUtils.TIME_OUT; |
| logger.warn(MSG_FORWARD_TIMEOUT, name, plan, receiver); |
| } |
| return tsStatus; |
| } catch (IOException e) { |
| logger.error(MSG_FORWARD_ERROR, name, plan, receiver, e); |
| return StatusUtils.getStatus(StatusUtils.INTERNAL_ERROR, e.getMessage()); |
| } catch (TException e) { |
| TSStatus status; |
| if (e.getCause() instanceof SocketTimeoutException) { |
| status = StatusUtils.TIME_OUT; |
| logger.warn(MSG_FORWARD_TIMEOUT, name, plan, receiver); |
| } else { |
| logger.error(MSG_FORWARD_ERROR, name, plan, receiver, e); |
| status = StatusUtils.getStatus(StatusUtils.INTERNAL_ERROR, e.getMessage()); |
| } |
| // the connection may be broken, close it to avoid it being reused |
| client.getInputProtocol().getTransport().close(); |
| return status; |
| } finally { |
| ClientUtils.putBackSyncClient(client); |
| } |
| } |
| |
| /** |
| * Get an asynchronous thrift client of the given node. |
| * |
| * @return an asynchronous thrift client or null if the caller tries to connect the local node or |
| * the node cannot be reached. |
| */ |
| public AsyncClient getAsyncClient(Node node) { |
| return getAsyncClient(node, asyncClientPool, true); |
| } |
| |
| public AsyncClient getAsyncClient(Node node, boolean activatedOnly) { |
| return getAsyncClient(node, asyncClientPool, activatedOnly); |
| } |
| |
| public AsyncClient getSendLogAsyncClient(Node node) { |
| return getAsyncClient(node, asyncSendLogClientPool, true); |
| } |
| |
| private AsyncClient getAsyncClient(Node node, AsyncClientPool pool, boolean activatedOnly) { |
| if (ClusterConstant.EMPTY_NODE.equals(node) || node == null) { |
| return null; |
| } |
| try { |
| return pool.getClient(node, activatedOnly); |
| } catch (IOException e) { |
| logger.warn("{} cannot connect to node {}", name, node, e); |
| return null; |
| } |
| } |
| |
| /** |
| * NOTICE: client.putBack() must be called after use. the caller needs to check to see if the |
| * return value is null |
| * |
| * @param node the node to connect |
| * @return the client if node is available, otherwise null |
| */ |
| public Client getSyncClient(Node node) { |
| return getSyncClient(syncClientPool, node, true); |
| } |
| |
| public Client getSyncClient(Node node, boolean activatedOnly) { |
| return getSyncClient(syncClientPool, node, activatedOnly); |
| } |
| |
| public AtomicLong getTerm() { |
| return term; |
| } |
| |
| private synchronized LogDispatcher getLogDispatcher() { |
| if (logDispatcher == null) { |
| logDispatcher = new LogDispatcher(this); |
| } |
| return logDispatcher; |
| } |
| |
| /** |
| * wait until "voteCounter" counts down to zero, which means the quorum has received the log, or |
| * one follower tells the node that it is no longer a valid leader, or a timeout is triggered. |
| */ |
| @SuppressWarnings({"java:S2445"}) // safe synchronized |
| private AppendLogResult waitAppendResult( |
| AtomicInteger voteCounter, AtomicBoolean leaderShipStale, AtomicLong newLeaderTerm) { |
| // wait for the followers to vote |
| long startTime = Timer.Statistic.RAFT_SENDER_VOTE_COUNTER.getOperationStartTime(); |
| synchronized (voteCounter) { |
| long waitStart = System.currentTimeMillis(); |
| long alreadyWait = 0; |
| while (voteCounter.get() > 0 |
| && alreadyWait < RaftServer.getWriteOperationTimeoutMS() |
| && voteCounter.get() != Integer.MAX_VALUE) { |
| try { |
| voteCounter.wait(RaftServer.getWriteOperationTimeoutMS()); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.warn("Unexpected interruption when sending a log", e); |
| } |
| alreadyWait = System.currentTimeMillis() - waitStart; |
| } |
| } |
| Timer.Statistic.RAFT_SENDER_VOTE_COUNTER.calOperationCostTimeFromStart(startTime); |
| |
| // a node has a larger term than the local node, so this node is no longer a valid leader |
| if (leaderShipStale.get()) { |
| stepDown(newLeaderTerm.get(), false); |
| return AppendLogResult.LEADERSHIP_STALE; |
| } |
| // the node knows it is no long the leader from other requests |
| if (character != NodeCharacter.LEADER) { |
| return AppendLogResult.LEADERSHIP_STALE; |
| } |
| |
| // cannot get enough agreements within a certain amount of time |
| if (voteCounter.get() > 0) { |
| return AppendLogResult.TIME_OUT; |
| } |
| |
| // voteCounter has counted down to zero |
| return AppendLogResult.OK; |
| } |
| |
| @SuppressWarnings("java:S2445") |
| void commitLog(Log log) throws LogExecutionException { |
| long startTime = |
| Statistic.RAFT_SENDER_COMPETE_LOG_MANAGER_BEFORE_COMMIT.getOperationStartTime(); |
| synchronized (logManager) { |
| Statistic.RAFT_SENDER_COMPETE_LOG_MANAGER_BEFORE_COMMIT.calOperationCostTimeFromStart( |
| startTime); |
| |
| startTime = Statistic.RAFT_SENDER_COMMIT_LOG_IN_MANAGER.getOperationStartTime(); |
| logManager.commitTo(log.getCurrLogIndex()); |
| } |
| Statistic.RAFT_SENDER_COMMIT_LOG_IN_MANAGER.calOperationCostTimeFromStart(startTime); |
| // when using async applier, the log here may not be applied. To return the execution |
| // result, we must wait until the log is applied. |
| startTime = Statistic.RAFT_SENDER_COMMIT_WAIT_LOG_APPLY.getOperationStartTime(); |
| synchronized (log) { |
| while (!log.isApplied()) { |
| // wait until the log is applied |
| try { |
| log.wait(5); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new LogExecutionException(e); |
| } |
| } |
| } |
| Statistic.RAFT_SENDER_COMMIT_WAIT_LOG_APPLY.calOperationCostTimeFromStart(startTime); |
| if (log.getException() != null) { |
| throw new LogExecutionException(log.getException()); |
| } |
| } |
| |
| private TSStatus handleLogExecutionException(PhysicalPlanLog log, LogExecutionException e) { |
| Throwable cause = IOUtils.getRootCause(e); |
| if (cause instanceof BatchProcessException) { |
| return RpcUtils.getStatus(Arrays.asList(((BatchProcessException) cause).getFailingStatus())); |
| } |
| TSStatus tsStatus = |
| StatusUtils.getStatus(StatusUtils.EXECUTE_STATEMENT_ERROR, cause.getMessage()); |
| if (cause instanceof RuntimeException) { |
| logger.error("RuntimeException during executing {}", log, cause); |
| } |
| if (cause instanceof IoTDBException) { |
| tsStatus.setCode(((IoTDBException) cause).getErrorCode()); |
| } |
| if (!(cause instanceof PathNotExistException) |
| && !(cause instanceof StorageGroupNotSetException) |
| && !(cause instanceof PathAlreadyExistException) |
| && !(cause instanceof StorageGroupAlreadySetException)) { |
| logger.debug("{} cannot be executed because ", log, cause); |
| } |
| return tsStatus; |
| } |
| |
| AppendEntryRequest buildAppendEntryRequest(Log log, boolean serializeNow) { |
| AppendEntryRequest request = new AppendEntryRequest(); |
| request.setTerm(term.get()); |
| if (serializeNow) { |
| ByteBuffer byteBuffer = log.serialize(); |
| log.setByteSize(byteBuffer.array().length); |
| request.setEntry(byteBuffer); |
| } |
| request.setLeader(getThisNode()); |
| // don't need lock because even if it's larger than the commitIndex when appending this log to |
| // logManager, the follower can handle the larger commitIndex with no effect |
| request.setLeaderCommit(logManager.getCommitLogIndex()); |
| request.setPrevLogIndex(log.getCurrLogIndex() - 1); |
| try { |
| request.setPrevLogTerm(logManager.getTerm(log.getCurrLogIndex() - 1)); |
| } catch (Exception e) { |
| logger.error("getTerm failed for newly append entries", e); |
| } |
| if (getHeader() != null) { |
| // data groups use header to find a particular DataGroupMember |
| request.setHeader(getHeader()); |
| } |
| return request; |
| } |
| |
| /** |
| * If "newTerm" is larger than the local term, give up the leadership, become a follower and reset |
| * heartbeat timer. |
| * |
| * @param fromLeader true if the request is from a leader, false if the request is from an |
| * elector. |
| */ |
| public void stepDown(long newTerm, boolean fromLeader) { |
| synchronized (term) { |
| long currTerm = term.get(); |
| // confirm that the heartbeat of the new leader hasn't come |
| if (currTerm < newTerm) { |
| logger.info("{} has update it's term to {}", getName(), newTerm); |
| term.set(newTerm); |
| setVoteFor(null); |
| setLeader(ClusterConstant.EMPTY_NODE); |
| updateHardState(newTerm, getVoteFor()); |
| } |
| |
| if (fromLeader) { |
| // only when the request is from a leader should we update lastHeartbeatReceivedTime, |
| // otherwise the node may be stuck in FOLLOWER state by a stale node. |
| setCharacter(NodeCharacter.FOLLOWER); |
| lastHeartbeatReceivedTime = System.currentTimeMillis(); |
| } |
| } |
| } |
| |
| public Node getThisNode() { |
| return thisNode; |
| } |
| |
| public void setThisNode(Node thisNode) { |
| this.thisNode = thisNode; |
| } |
| |
| /** @return the header of the data raft group or null if this is in a meta group. */ |
| public Node getHeader() { |
| return null; |
| } |
| |
| public void updateHardState(long currentTerm, Node voteFor) { |
| HardState state = logManager.getHardState(); |
| state.setCurrentTerm(currentTerm); |
| state.setVoteFor(voteFor); |
| logManager.updateHardState(state); |
| } |
| |
| public Node getVoteFor() { |
| return voteFor; |
| } |
| |
| public void setVoteFor(Node voteFor) { |
| if (!Objects.equals(voteFor, this.voteFor)) { |
| logger.info("{} has update it's voteFor to {}", getName(), voteFor); |
| this.voteFor = voteFor; |
| } |
| } |
| |
| /** |
| * Append a log to all followers in the group until half of them accept the log or the leadership |
| * is lost. |
| * |
| * @return true if the log is accepted by the quorum of the group, false otherwise |
| */ |
| boolean appendLogInGroup(Log log) throws LogExecutionException { |
| if (allNodes.size() == 1) { |
| // single node group, no followers |
| long startTime = Timer.Statistic.RAFT_SENDER_COMMIT_LOG.getOperationStartTime(); |
| logger.debug(MSG_LOG_IS_ACCEPTED, name, log); |
| commitLog(log); |
| Timer.Statistic.RAFT_SENDER_COMMIT_LOG.calOperationCostTimeFromStart(startTime); |
| return true; |
| } |
| |
| int retryTime = 0; |
| while (true) { |
| long startTime = Timer.Statistic.RAFT_SENDER_SEND_LOG_TO_FOLLOWERS.getOperationStartTime(); |
| logger.debug("{}: Send log {} to other nodes, retry times: {}", name, log, retryTime); |
| if (character != NodeCharacter.LEADER) { |
| logger.debug("Has lose leadership, so need not to send log"); |
| return false; |
| } |
| AppendLogResult result = sendLogToFollowers(log, allNodes.size() / 2); |
| Timer.Statistic.RAFT_SENDER_SEND_LOG_TO_FOLLOWERS.calOperationCostTimeFromStart(startTime); |
| switch (result) { |
| case OK: |
| startTime = Timer.Statistic.RAFT_SENDER_COMMIT_LOG.getOperationStartTime(); |
| logger.debug(MSG_LOG_IS_ACCEPTED, name, log); |
| commitLog(log); |
| Timer.Statistic.RAFT_SENDER_COMMIT_LOG.calOperationCostTimeFromStart(startTime); |
| return true; |
| case TIME_OUT: |
| logger.debug("{}: log {} timed out, retrying...", name, log); |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| retryTime++; |
| if (retryTime > 5) { |
| return false; |
| } |
| break; |
| case LEADERSHIP_STALE: |
| // abort the appending, the new leader will fix the local logs by catch-up |
| default: |
| return false; |
| } |
| } |
| } |
| |
| /** |
| * Send the given log to all the followers and decide the result by how many followers return a |
| * success. |
| * |
| * @param requiredQuorum the number of votes needed to make the log valid, when requiredQuorum <= |
| * 0, half of the cluster size will be used. |
| * @return an AppendLogResult |
| */ |
| private AppendLogResult sendLogToFollowers(Log log, int requiredQuorum) { |
| if (requiredQuorum <= 0) { |
| // use half of the members' size as the quorum |
| return sendLogToFollowers(log, new AtomicInteger(allNodes.size() / 2)); |
| } else { |
| // make sure quorum does not exceed the number of members - 1 |
| return sendLogToFollowers( |
| log, new AtomicInteger(Math.min(requiredQuorum, allNodes.size() - 1))); |
| } |
| } |
| |
| /** |
| * Send the log to each follower. Every time a follower returns a success, "voteCounter" is |
| * decreased by 1 and when it counts to 0, return an OK. If any follower returns a higher term |
| * than the local term, retire from leader and return a LEADERSHIP_STALE. If "voteCounter" is |
| * still positive after a certain time, return TIME_OUT. |
| * |
| * @param voteCounter a decreasing vote counter |
| * @return an AppendLogResult indicating a success or a failure and why |
| */ |
| private AppendLogResult sendLogToFollowers(Log log, AtomicInteger voteCounter) { |
| if (allNodes.size() == 1) { |
| // single node group, does not need the agreement of others |
| return AppendLogResult.OK; |
| } |
| logger.debug("{} sending a log to followers: {}", name, log); |
| |
| // if a follower has larger term than this node, leaderShipStale will be set to true and |
| // newLeaderTerm will store the follower's term |
| AtomicBoolean leaderShipStale = new AtomicBoolean(false); |
| AtomicLong newLeaderTerm = new AtomicLong(term.get()); |
| |
| AppendEntryRequest request = buildAppendEntryRequest(log, true); |
| |
| try { |
| if (allNodes.size() > 2) { |
| // if there are more than one followers, send the requests in parallel so that one slow |
| // follower will not be blocked |
| for (Node node : allNodes) { |
| appendLogThreadPool.submit( |
| () -> |
| sendLogToFollower( |
| log, voteCounter, node, leaderShipStale, newLeaderTerm, request)); |
| if (character != NodeCharacter.LEADER) { |
| return AppendLogResult.LEADERSHIP_STALE; |
| } |
| } |
| } else { |
| // there is only one member, send to it within this thread to reduce thread switching |
| // overhead |
| for (Node node : allNodes) { |
| sendLogToFollower(log, voteCounter, node, leaderShipStale, newLeaderTerm, request); |
| if (character != NodeCharacter.LEADER) { |
| return AppendLogResult.LEADERSHIP_STALE; |
| } |
| } |
| } |
| |
| } catch (ConcurrentModificationException e) { |
| // retry if allNodes has changed |
| return AppendLogResult.TIME_OUT; |
| } |
| |
| return waitAppendResult(voteCounter, leaderShipStale, newLeaderTerm); |
| } |
| |
| /** Send "log" to "node". */ |
| public void sendLogToFollower( |
| Log log, |
| AtomicInteger voteCounter, |
| Node node, |
| AtomicBoolean leaderShipStale, |
| AtomicLong newLeaderTerm, |
| AppendEntryRequest request) { |
| if (node.equals(thisNode)) { |
| return; |
| } |
| /** |
| * if the peer's log progress is too stale, wait until it catches up, otherwise, there may be |
| * too many waiting requests on the peer's side. |
| */ |
| long startTime = Timer.Statistic.RAFT_SENDER_WAIT_FOR_PREV_LOG.getOperationStartTime(); |
| Peer peer = peerMap.computeIfAbsent(node, k -> new Peer(logManager.getLastLogIndex())); |
| if (!waitForPrevLog(peer, log)) { |
| logger.warn("{}: node {} timed out when appending {}", name, node, log); |
| return; |
| } |
| Timer.Statistic.RAFT_SENDER_WAIT_FOR_PREV_LOG.calOperationCostTimeFromStart(startTime); |
| |
| if (character != NodeCharacter.LEADER) { |
| return; |
| } |
| |
| if (config.isUseAsyncServer()) { |
| sendLogAsync(log, voteCounter, node, leaderShipStale, newLeaderTerm, request, peer); |
| } else { |
| sendLogSync(log, voteCounter, node, leaderShipStale, newLeaderTerm, request, peer); |
| } |
| } |
| |
| /** |
| * wait until the difference of log index between the matched log of peer and the given log become |
| * no bigger than maxLogDiff. |
| */ |
| @SuppressWarnings("java:S2445") // safe synchronized |
| public boolean waitForPrevLog(Peer peer, Log log) { |
| final int maxLogDiff = config.getMaxNumOfLogsInMem(); |
| long waitStart = System.currentTimeMillis(); |
| long alreadyWait = 0; |
| // if the peer falls behind too much, wait until it catches up, otherwise there may be too |
| // many client threads in the peer |
| while (peer.getMatchIndex() < log.getCurrLogIndex() - maxLogDiff |
| && character == NodeCharacter.LEADER |
| && alreadyWait <= RaftServer.getWriteOperationTimeoutMS()) { |
| synchronized (peer) { |
| try { |
| peer.wait(RaftServer.getWriteOperationTimeoutMS()); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.warn("Waiting for peer to catch up interrupted"); |
| return false; |
| } |
| } |
| alreadyWait = System.currentTimeMillis() - waitStart; |
| } |
| return alreadyWait <= RaftServer.getWriteOperationTimeoutMS(); |
| } |
| |
| private void sendLogSync( |
| Log log, |
| AtomicInteger voteCounter, |
| Node node, |
| AtomicBoolean leaderShipStale, |
| AtomicLong newLeaderTerm, |
| AppendEntryRequest request, |
| Peer peer) { |
| Client client = getSyncClient(node); |
| if (client != null) { |
| AppendNodeEntryHandler handler = |
| getAppendNodeEntryHandler(log, voteCounter, node, leaderShipStale, newLeaderTerm, peer); |
| try { |
| logger.debug("{} sending a log to {}: {}", name, node, log); |
| long result = client.appendEntry(request); |
| handler.onComplete(result); |
| } catch (TException e) { |
| client.getInputProtocol().getTransport().close(); |
| handler.onError(e); |
| } catch (Exception e) { |
| handler.onError(e); |
| } finally { |
| ClientUtils.putBackSyncClient(client); |
| } |
| } |
| } |
| |
| public AppendNodeEntryHandler getAppendNodeEntryHandler( |
| Log log, |
| AtomicInteger voteCounter, |
| Node node, |
| AtomicBoolean leaderShipStale, |
| AtomicLong newLeaderTerm, |
| Peer peer) { |
| AppendNodeEntryHandler handler = new AppendNodeEntryHandler(); |
| handler.setReceiver(node); |
| handler.setVoteCounter(voteCounter); |
| handler.setLeaderShipStale(leaderShipStale); |
| handler.setLog(log); |
| handler.setMember(this); |
| handler.setPeer(peer); |
| handler.setReceiverTerm(newLeaderTerm); |
| return handler; |
| } |
| |
| @TestOnly |
| public void setAppendLogThreadPool(ExecutorService appendLogThreadPool) { |
| this.appendLogThreadPool = appendLogThreadPool; |
| } |
| |
| public Object getSnapshotApplyLock() { |
| return snapshotApplyLock; |
| } |
| |
| /** |
| * Find the local previous log of "log". If such log is found, discard all local logs behind it |
| * and append "log" to it. Otherwise report a log mismatch. |
| * |
| * @return Response.RESPONSE_AGREE when the log is successfully appended or Response |
| * .RESPONSE_LOG_MISMATCH if the previous log of "log" is not found. |
| */ |
| private long appendEntry(long prevLogIndex, long prevLogTerm, long leaderCommit, Log log) { |
| long resp = checkPrevLogIndex(prevLogIndex); |
| if (resp != Response.RESPONSE_AGREE) { |
| return resp; |
| } |
| |
| long startTime = Timer.Statistic.RAFT_RECEIVER_APPEND_ENTRY.getOperationStartTime(); |
| long success; |
| synchronized (logManager) { |
| success = logManager.maybeAppend(prevLogIndex, prevLogTerm, leaderCommit, log); |
| } |
| Timer.Statistic.RAFT_RECEIVER_APPEND_ENTRY.calOperationCostTimeFromStart(startTime); |
| if (success != -1) { |
| logger.debug("{} append a new log {}", name, log); |
| resp = Response.RESPONSE_AGREE; |
| } else { |
| // the incoming log points to an illegal position, reject it |
| resp = Response.RESPONSE_LOG_MISMATCH; |
| } |
| return resp; |
| } |
| |
| /** Wait until all logs before "prevLogIndex" arrive or a timeout is reached. */ |
| private boolean waitForPrevLog(long prevLogIndex) { |
| long waitStart = System.currentTimeMillis(); |
| long alreadyWait = 0; |
| Object logUpdateCondition = logManager.getLogUpdateCondition(prevLogIndex); |
| long lastLogIndex = logManager.getLastLogIndex(); |
| while (lastLogIndex < prevLogIndex && alreadyWait <= RaftServer.getWriteOperationTimeoutMS()) { |
| try { |
| // each time new logs are appended, this will be notified |
| synchronized (logUpdateCondition) { |
| logUpdateCondition.wait(1); |
| } |
| lastLogIndex = logManager.getLastLogIndex(); |
| if (lastLogIndex >= prevLogIndex) { |
| return true; |
| } |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| return false; |
| } |
| alreadyWait = System.currentTimeMillis() - waitStart; |
| } |
| |
| return alreadyWait <= RaftServer.getWriteOperationTimeoutMS(); |
| } |
| |
| private long checkPrevLogIndex(long prevLogIndex) { |
| long lastLogIndex = logManager.getLastLogIndex(); |
| long startTime = Timer.Statistic.RAFT_RECEIVER_WAIT_FOR_PREV_LOG.getOperationStartTime(); |
| if (lastLogIndex < prevLogIndex && !waitForPrevLog(prevLogIndex)) { |
| // there are logs missing between the incoming log and the local last log, and such logs |
| // did not come within a timeout, report a mismatch to the sender and it shall fix this |
| // through catch-up |
| Timer.Statistic.RAFT_RECEIVER_INDEX_DIFF.add(prevLogIndex - lastLogIndex); |
| return Response.RESPONSE_LOG_MISMATCH; |
| } |
| Timer.Statistic.RAFT_RECEIVER_WAIT_FOR_PREV_LOG.calOperationCostTimeFromStart(startTime); |
| return Response.RESPONSE_AGREE; |
| } |
| |
| /** |
| * Find the local previous log of "log". If such log is found, discard all local logs behind it |
| * and append "log" to it. Otherwise report a log mismatch. |
| * |
| * @param logs append logs |
| * @return Response.RESPONSE_AGREE when the log is successfully appended or Response |
| * .RESPONSE_LOG_MISMATCH if the previous log of "log" is not found. |
| */ |
| private long appendEntries( |
| long prevLogIndex, long prevLogTerm, long leaderCommit, List<Log> logs) { |
| logger.debug( |
| "{}, prevLogIndex={}, prevLogTerm={}, leaderCommit={}", |
| name, |
| prevLogIndex, |
| prevLogTerm, |
| leaderCommit); |
| if (logs.isEmpty()) { |
| return Response.RESPONSE_AGREE; |
| } |
| |
| long resp = checkPrevLogIndex(prevLogIndex); |
| if (resp != Response.RESPONSE_AGREE) { |
| return resp; |
| } |
| |
| synchronized (logManager) { |
| long startTime = Timer.Statistic.RAFT_RECEIVER_APPEND_ENTRY.getOperationStartTime(); |
| resp = logManager.maybeAppend(prevLogIndex, prevLogTerm, leaderCommit, logs); |
| Timer.Statistic.RAFT_RECEIVER_APPEND_ENTRY.calOperationCostTimeFromStart(startTime); |
| if (resp != -1) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} append a new log list {}, commit to {}", name, logs, leaderCommit); |
| } |
| resp = Response.RESPONSE_AGREE; |
| } else { |
| // the incoming log points to an illegal position, reject it |
| resp = Response.RESPONSE_LOG_MISMATCH; |
| } |
| } |
| return resp; |
| } |
| |
| /** |
| * Check the term of the AppendEntryRequest. The term checked is the term of the leader, not the |
| * term of the log. A new leader can still send logs of old leaders. |
| * |
| * @return -1 if the check is passed, >0 otherwise |
| */ |
| private long checkRequestTerm(long leaderTerm, Node leader) { |
| long localTerm; |
| |
| synchronized (term) { |
| // if the request comes before the heartbeat arrives, the local term may be smaller than the |
| // leader term |
| localTerm = term.get(); |
| if (leaderTerm < localTerm) { |
| logger.debug( |
| "{} rejected the AppendEntriesRequest for term: {}/{}", name, leaderTerm, localTerm); |
| return localTerm; |
| } else { |
| if (leaderTerm > localTerm) { |
| stepDown(leaderTerm, true); |
| } else { |
| lastHeartbeatReceivedTime = System.currentTimeMillis(); |
| } |
| setLeader(leader); |
| if (character != NodeCharacter.FOLLOWER) { |
| term.notifyAll(); |
| } |
| } |
| } |
| logger.debug("{} accepted the AppendEntryRequest for term: {}", name, localTerm); |
| return Response.RESPONSE_AGREE; |
| } |
| |
| enum AppendLogResult { |
| OK, |
| TIME_OUT, |
| LEADERSHIP_STALE |
| } |
| } |
