This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch expr
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/expr by this push:
new 2cc6cd0 add leader changed response
2cc6cd0 is described below
commit 2cc6cd0a42a1d245b60c2d89041e303cc0f203a1
Author: jt <[email protected]>
AuthorDate: Thu Dec 2 09:27:15 2021 +0800
add leader changed response
---
.../cluster/log/appender/BlockingLogAppender.java | 16 +-
.../iotdb/cluster/log/appender/LogAppender.java | 7 +-
.../cluster/log/appender/LogAppenderFactory.java | 1 -
.../log/appender/SlidingWindowLogAppender.java | 5 +-
.../cluster/server/member/DataGroupMember.java | 14 ++
.../iotdb/cluster/server/member/RaftMember.java | 171 ++++++++-----------
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 11 ++
.../org/apache/iotdb/db/service/TSServiceImpl.java | 15 +-
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 2 +
.../apache/iotdb/session/SessionConnection.java | 181 ++++++++++++++++++---
thrift/src/main/thrift/rpc.thrift | 11 ++
11 files changed, 294 insertions(+), 140 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
index e93969c..7cb614a 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.cluster.log.appender;
-import java.util.List;
import org.apache.iotdb.cluster.config.ClusterConstant;
import org.apache.iotdb.cluster.log.Log;
import org.apache.iotdb.cluster.log.manage.RaftLogManager;
@@ -27,9 +26,12 @@ import org.apache.iotdb.cluster.rpc.thrift.AppendEntryResult;
import org.apache.iotdb.cluster.server.Response;
import org.apache.iotdb.cluster.server.member.RaftMember;
import org.apache.iotdb.cluster.server.monitor.Timer;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
+
/**
* BlockingLogAppender wait for a certain amount of time when it receives
out-of-order entries
* (entries with indices larger than local last entry's index + 1), if the
local log is updated
@@ -53,7 +55,7 @@ public class BlockingLogAppender implements LogAppender {
* 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.
+ * .RESPONSE_LOG_MISMATCH if the previous log of "log" is not found.
*/
public AppendEntryResult appendEntry(
long prevLogIndex, long prevLogTerm, long leaderCommit, Log log) {
@@ -83,9 +85,7 @@ public class BlockingLogAppender implements LogAppender {
return result;
}
- /**
- * Wait until all logs before "prevLogIndex" arrive or a timeout is reached.
- */
+ /** Wait until all logs before "prevLogIndex" arrive or a timeout is
reached. */
private boolean waitForPrevLog(long prevLogIndex) {
long waitStart = System.currentTimeMillis();
long alreadyWait = 0;
@@ -132,7 +132,7 @@ public class BlockingLogAppender implements LogAppender {
*
* @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.
+ * .RESPONSE_LOG_MISMATCH if the previous log of "log" is not found.
*/
public AppendEntryResult appendEntries(
long prevLogIndex, long prevLogTerm, long leaderCommit, List<Log> logs) {
@@ -158,8 +158,8 @@ public class BlockingLogAppender implements LogAppender {
Timer.Statistic.RAFT_RECEIVER_APPEND_ENTRY.calOperationCostTimeFromStart(startTime);
if (resp != -1) {
if (logger.isDebugEnabled()) {
- logger.debug("{} append a new log list {}, commit to {}",
member.getName(), logs,
- leaderCommit);
+ logger.debug(
+ "{} append a new log list {}, commit to {}", member.getName(),
logs, leaderCommit);
}
result.status = Response.RESPONSE_STRONG_ACCEPT;
result.setLastLogIndex(logManager.getLastLogIndex());
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
index f997b06..b5bd35f 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
@@ -17,13 +17,13 @@
* under the License.
*/
-
package org.apache.iotdb.cluster.log.appender;
-import java.util.List;
import org.apache.iotdb.cluster.log.Log;
import org.apache.iotdb.cluster.rpc.thrift.AppendEntryResult;
+import java.util.List;
+
/**
* LogAppender appends newly incoming entries to the local log of a member,
providing different
* policies for out-of-order entries and other cases.
@@ -33,6 +33,5 @@ public interface LogAppender {
AppendEntryResult appendEntries(
long prevLogIndex, long prevLogTerm, long leaderCommit, List<Log> logs);
- AppendEntryResult appendEntry(
- long prevLogIndex, long prevLogTerm, long leaderCommit, Log log);
+ AppendEntryResult appendEntry(long prevLogIndex, long prevLogTerm, long
leaderCommit, Log log);
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppenderFactory.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppenderFactory.java
index 1f1e83c..0b4a4d4 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppenderFactory.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppenderFactory.java
@@ -17,7 +17,6 @@
* under the License.
*/
-
package org.apache.iotdb.cluster.log.appender;
import org.apache.iotdb.cluster.server.member.RaftMember;
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
index 928f960..d2ecb1c 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
@@ -19,8 +19,6 @@
package org.apache.iotdb.cluster.log.appender;
-import java.util.Arrays;
-import java.util.List;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.log.Log;
import org.apache.iotdb.cluster.log.manage.RaftLogManager;
@@ -30,6 +28,9 @@ import org.apache.iotdb.cluster.server.member.RaftMember;
import org.apache.iotdb.cluster.server.monitor.Timer;
import org.apache.iotdb.cluster.server.monitor.Timer.Statistic;
+import java.util.Arrays;
+import java.util.List;
+
public class SlidingWindowLogAppender implements LogAppender {
private int windowCapacity =
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
index 6472835..975bba2 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
@@ -32,6 +32,8 @@ import org.apache.iotdb.cluster.log.LogApplier;
import org.apache.iotdb.cluster.log.LogParser;
import org.apache.iotdb.cluster.log.Snapshot;
import org.apache.iotdb.cluster.log.VotingLog;
+import org.apache.iotdb.cluster.log.appender.BlockingLogAppender;
+import org.apache.iotdb.cluster.log.appender.SlidingWindowLogAppender;
import org.apache.iotdb.cluster.log.applier.AsyncDataLogApplier;
import org.apache.iotdb.cluster.log.applier.DataLogApplier;
import org.apache.iotdb.cluster.log.logtypes.AddNodeLog;
@@ -192,6 +194,10 @@ public class DataGroupMember extends RaftMember implements
DataGroupMemberMBean
setQueryManager(new ClusterQueryManager());
localQueryExecutor = new LocalQueryExecutor(this);
lastAppliedPartitionTableVersion = new
LastAppliedPatitionTableVersion(getMemberDir());
+ appenderFactory =
+
ClusterDescriptor.getInstance().getConfig().isUseFollowerSlidingWindow()
+ ? new SlidingWindowLogAppender.Factory()
+ : new BlockingLogAppender.Factory();
}
DataGroupMember(TProtocolFactory factory, PartitionGroup nodes,
MetaGroupMember metaGroupMember) {
@@ -232,6 +238,10 @@ public class DataGroupMember extends RaftMember implements
DataGroupMemberMBean
voteFor = logManager.getHardState().getVoteFor();
localQueryExecutor = new LocalQueryExecutor(this);
lastAppliedPartitionTableVersion = new
LastAppliedPatitionTableVersion(getMemberDir());
+ appenderFactory =
+
ClusterDescriptor.getInstance().getConfig().isUseFollowerSlidingWindow()
+ ? new SlidingWindowLogAppender.Factory()
+ : new BlockingLogAppender.Factory();
}
/**
@@ -801,6 +811,10 @@ public class DataGroupMember extends RaftMember implements
DataGroupMemberMBean
private TSStatus executeNonQueryPlanWithKnownLeader(PhysicalPlan plan) {
if (character == NodeCharacter.LEADER) {
+ if (plan.getTargetedTerm() > 0 && plan.getTargetedTerm() != term.get()) {
+ return
StatusUtils.getStatus(TSStatusCode.LEADER_CHANGED).setMessage(term.get() + "");
+ }
+
long startTime =
Statistic.DATA_GROUP_MEMBER_LOCAL_EXECUTION.getOperationStartTime();
TSStatus status = processPlanLocally(plan);
boolean hasCreated = false;
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
index 4d4587d..ce8579a 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
@@ -44,7 +44,6 @@ import org.apache.iotdb.cluster.log.VotingLogList;
import org.apache.iotdb.cluster.log.appender.BlockingLogAppender;
import org.apache.iotdb.cluster.log.appender.LogAppender;
import org.apache.iotdb.cluster.log.appender.LogAppenderFactory;
-import org.apache.iotdb.cluster.log.appender.SlidingWindowLogAppender;
import org.apache.iotdb.cluster.log.catchup.CatchUpTask;
import org.apache.iotdb.cluster.log.logtypes.PhysicalPlanLog;
import org.apache.iotdb.cluster.log.manage.RaftLogManager;
@@ -93,6 +92,7 @@ import
org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.executor.PlanExecutor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.sys.LogPlan;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.rpc.RpcUtils;
@@ -130,8 +130,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static
org.apache.iotdb.cluster.config.ClusterConstant.THREAD_POLL_WAIT_TERMINATION_TIME_S;
/**
- * RaftMember process the common raft logic like leader election, log
appending, catch-up and so
- * on.
+ * 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 implements RaftMemberMBean {
@@ -140,12 +139,10 @@ public abstract class RaftMember implements
RaftMemberMBean {
public static boolean USE_LOG_DISPATCHER = false;
private static final boolean USE_INDIRECT_LOG_DISPATCHER =
ClusterDescriptor.getInstance().getConfig().isUseIndirectBroadcasting();
- private static final boolean ENABLE_WEAK_ACCEPTANCE =
ClusterDescriptor.getInstance().getConfig()
- .isEnableWeakAcceptance();
+ private static final boolean ENABLE_WEAK_ACCEPTANCE =
+ ClusterDescriptor.getInstance().getConfig().isEnableWeakAcceptance();
- private static final LogAppenderFactory APPENDER_FACTORY =
- ClusterDescriptor.getInstance().getConfig().isUseFollowerSlidingWindow()
?
- new SlidingWindowLogAppender.Factory() : new
BlockingLogAppender.Factory();
+ protected LogAppenderFactory appenderFactory = new
BlockingLogAppender.Factory();
protected static final LogSequencerFactory SEQUENCER_FACTORY =
ClusterDescriptor.getInstance().getConfig().isUseAsyncSequencing()
? new Factory()
@@ -169,32 +166,22 @@ public abstract class RaftMember implements
RaftMemberMBean {
* 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
- */
+ /** the lock is to make sure that only one thread can apply snapshot at the
same time */
private final Object snapshotApplyLock = new Object();
private final Object heartBeatWaitObject = new Object();
protected Node thisNode = ClusterIoTDB.getInstance().getThisNode();
- /**
- * the nodes that belong to the same raft group as thisNode.
- */
+ /** the nodes that belong to the same raft group as thisNode. */
protected PartitionGroup allNodes;
ClusterConfig config = ClusterDescriptor.getInstance().getConfig();
- /**
- * the name of the member, to distinguish several members in the logs.
- */
+ /** the name of the member, to distinguish several members in the logs. */
String name;
- /**
- * to choose nodes to send request of joining cluster randomly.
- */
+ /** 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.
- */
+ /** 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
@@ -216,9 +203,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
*/
volatile long lastHeartbeatReceivedTime;
- /**
- * the raft logs are all stored and maintained in the log manager
- */
+ /** the raft logs are all stored and maintained in the log manager */
protected RaftLogManager logManager;
/**
@@ -237,9 +222,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
* member by comparing it with the current last log index.
*/
long lastReportedLogIndex;
- /**
- * the thread pool that runs catch-up tasks
- */
+ /** 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
@@ -270,32 +253,24 @@ public abstract class RaftMember implements
RaftMemberMBean {
* one slow node.
*/
private ExecutorService serialToParallelPool;
- /**
- * a thread pool that is used to do commit log tasks asynchronous in
heartbeat thread
- */
+ /** 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.
+ * which avoids the followers receiving out-of-order logs, forcing them to
wait for previous logs.
*/
private volatile LogDispatcher logDispatcher;
- /**
- * If this node can not be the leader, this parameter will be set true.
- */
+ /** If this node can not be the leader, this parameter will be set true. */
private volatile boolean skipElection = false;
/**
- * localExecutor is used to directly execute plans like load configuration
in the underlying
- * IoTDB
+ * localExecutor is used to directly execute plans like load configuration
in the underlying IoTDB
*/
protected PlanExecutor localExecutor;
- /**
- * (logIndex, logTerm) -> append handler
- */
+ /** (logIndex, logTerm) -> append handler */
protected Map<Pair<Long, Long>, AppendNodeEntryHandler> sentLogHandlers =
new ConcurrentHashMap<>();
@@ -305,8 +280,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
private volatile LogAppender logAppender;
- protected RaftMember() {
- }
+ protected RaftMember() {}
protected RaftMember(String name, ClientManager clientManager) {
this.name = name;
@@ -368,7 +342,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
if (logAppender == null) {
synchronized (this) {
if (logAppender == null) {
- logAppender = APPENDER_FACTORY.create(this);
+ logAppender = appenderFactory.create(this);
}
}
}
@@ -662,9 +636,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
}
}
- /**
- * Similar to appendEntry, while the incoming load is batch of logs instead
of a single log.
- */
+ /** Similar to appendEntry, while the incoming load is batch of logs instead
of a single log. */
public AppendEntryResult appendEntries(AppendEntriesRequest request)
throws UnknownLogTypeException {
logger.debug("{} received an AppendEntriesRequest", name);
@@ -695,9 +667,9 @@ public abstract class RaftMember implements RaftMemberMBean
{
Timer.Statistic.RAFT_RECEIVER_LOG_PARSE.calOperationCostTimeFromStart(startTime);
- response = getLogAppender()
- .appendEntries(request.prevLogIndex, request.prevLogTerm,
request.leaderCommit,
- logs);
+ response =
+ getLogAppender()
+ .appendEntries(request.prevLogIndex, request.prevLogTerm,
request.leaderCommit, logs);
if (logger.isDebugEnabled()) {
logger.debug(
"{} AppendEntriesRequest of log size {} completed with result {}",
@@ -818,22 +790,16 @@ public abstract class RaftMember implements
RaftMemberMBean {
return lastCatchUpResponseTime;
}
- /**
- * Sub-classes will add their own process of HeartBeatResponse in this
method.
- */
- public void processValidHeartbeatResp(HeartBeatResponse response, Node
receiver) {
- }
+ /** 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() {
- }
+ /** 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.
+ * 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
@@ -927,9 +893,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
"%s:%s=%s", "org.apache.iotdb.cluster.service",
IoTDBConstant.JMX_TYPE, "Engine");
}
- /**
- * call back after syncLeader
- */
+ /** call back after syncLeader */
public interface CheckConsistency {
/**
@@ -938,7 +902,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
* @param leaderCommitId leader commit id
* @param localAppliedId local applied id
* @throws CheckConsistencyException maybe throw
CheckConsistencyException, which is defined in
- * implements.
+ * implements.
*/
void postCheckConsistency(long leaderCommitId, long localAppliedId)
throws CheckConsistencyException;
@@ -947,7 +911,8 @@ public abstract class RaftMember implements RaftMemberMBean
{
public static class MidCheckConsistency implements CheckConsistency {
/**
- * if leaderCommitId - localAppliedId > MaxReadLogLag, will throw
CHECK_MID_CONSISTENCY_EXCEPTION
+ * if leaderCommitId - localAppliedId > MaxReadLogLag, will throw
+ * CHECK_MID_CONSISTENCY_EXCEPTION
*
* @param leaderCommitId leader commit id
* @param localAppliedId local applied id
@@ -959,7 +924,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
if (leaderCommitId == Long.MAX_VALUE
|| leaderCommitId == Long.MIN_VALUE
|| leaderCommitId - localAppliedId
- > ClusterDescriptor.getInstance().getConfig().getMaxReadLogLag()) {
+ >
ClusterDescriptor.getInstance().getConfig().getMaxReadLogLag()) {
throw CheckConsistencyException.CHECK_MID_CONSISTENCY_EXCEPTION;
}
}
@@ -992,7 +957,7 @@ public abstract class RaftMember implements RaftMemberMBean
{
* @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
+ * value after timeout
*/
public boolean syncLeader(CheckConsistency checkConsistency) throws
CheckConsistencyException {
if (character == NodeCharacter.LEADER) {
@@ -1011,9 +976,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
return waitUntilCatchUp(checkConsistency);
}
- /**
- * Wait until the leader of this node becomes known or time out.
- */
+ /** 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())) {
@@ -1040,7 +1003,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
*
* @return true if this node has caught up before timeout, false otherwise
* @throws CheckConsistencyException if leaderCommitId bigger than
localAppliedId a threshold
- * value after timeout
+ * value after timeout
*/
protected boolean waitUntilCatchUp(CheckConsistency checkConsistency)
throws CheckConsistencyException {
@@ -1073,7 +1036,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
* sync local applyId to leader commitId
*
* @param leaderCommitId leader commit id
- * @param fastFail if enable, when log differ too much, return false
directly.
+ * @param fastFail if enable, when log differ too much, return false
directly.
* @return true if leaderCommitId <= localAppliedId
*/
public boolean syncLocalApply(long leaderCommitId, boolean fastFail) {
@@ -1126,7 +1089,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
* 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
+ * during the appending
*/
public TSStatus processPlanLocally(PhysicalPlan plan) {
if (USE_LOG_DISPATCHER) {
@@ -1234,7 +1197,8 @@ public abstract class RaftMember implements
RaftMemberMBean {
Statistic.LOG_DISPATCHER_FROM_CREATE_TO_OK.calOperationCostTimeFromStart(
log.getCreateTime());
Statistic.LOG_DISPATCHER_TOTAL.calOperationCostTimeFromStart(totalStartTime);
- return StatusUtils.OK;
+ return StatusUtils.getStatus(TSStatusCode.WEAKLY_ACCEPTED)
+ .setMessage(log.getCurrLogIndex() + "-" + log.getCurrLogTerm());
case OK:
logger.debug(MSG_LOG_IS_ACCEPTED, name, log);
startTime =
Timer.Statistic.RAFT_SENDER_COMMIT_LOG.getOperationStartTime();
@@ -1243,7 +1207,9 @@ public abstract class RaftMember implements
RaftMemberMBean {
Statistic.LOG_DISPATCHER_FROM_CREATE_TO_OK.calOperationCostTimeFromStart(
log.getCreateTime());
Statistic.LOG_DISPATCHER_TOTAL.calOperationCostTimeFromStart(totalStartTime);
- return StatusUtils.OK;
+ return StatusUtils.OK
+ .deepCopy()
+ .setMessage(log.getCurrLogIndex() + "-" + log.getCurrLogTerm());
case TIME_OUT:
logger.debug("{}: log {} timed out...", name, log);
break;
@@ -1363,9 +1329,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
return peerMap;
}
- /**
- * @return true if there is a log whose index is "index" and term is "term",
false otherwise
- */
+ /** @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);
@@ -1384,18 +1348,15 @@ public abstract class RaftMember implements
RaftMemberMBean {
return syncLock;
}
- /**
- * Sub-classes will add their own process of HeartBeatRequest in this method.
- */
- void processValidHeartbeatReq(HeartBeatRequest request, HeartBeatResponse
response) {
- }
+ /** 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.
+ * smaller term or Response.RESPONSE_LOG_MISMATCH if the elector has
older logs.
*/
long checkElectorLogProgress(ElectionRequest electionRequest) {
@@ -1439,7 +1400,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
* 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.
+ * smaller term or Response.RESPONSE_LOG_MISMATCH if the elector has
older logs.
*/
long checkLogProgress(long lastLogIndex, long lastLogTerm) {
long response;
@@ -1456,10 +1417,10 @@ public abstract class RaftMember implements
RaftMemberMBean {
/**
* 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 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
+ * communication
* @return a TSStatus indicating if the forwarding is successful.
*/
public TSStatus forwardPlan(PhysicalPlan plan, Node node, RaftNode header) {
@@ -1490,7 +1451,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
/**
* Forward a non-query plan to "receiver" using "client".
*
- * @param plan a non-query plan
+ * @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.
*/
@@ -1572,7 +1533,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
* 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.
+ * the node cannot be reached.
*/
public AsyncClient getAsyncClient(Node node) {
try {
@@ -1680,6 +1641,14 @@ public abstract class RaftMember implements
RaftMemberMBean {
return logDispatcher;
}
+ private boolean canBeWeaklyAccepted(Log log) {
+ if (!(log instanceof PhysicalPlanLog)) {
+ return false;
+ }
+ PhysicalPlanLog physicalPlanLog = (PhysicalPlanLog) log;
+ return physicalPlanLog.getPlan() instanceof InsertPlan;
+ }
+
/**
* 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.
@@ -1699,9 +1668,9 @@ public abstract class RaftMember implements
RaftMemberMBean {
long waitStart = System.currentTimeMillis();
long alreadyWait = 0;
while (stronglyAcceptedNodeNum < quorumSize
- && (!ENABLE_WEAK_ACCEPTANCE
- || (totalAccepted < allNodes.size() - 1)
- || votingLogList.size() > config.getMaxNumOfLogsInMem())
+ && (!(ENABLE_WEAK_ACCEPTANCE && canBeWeaklyAccepted(log.getLog()))
+ || (totalAccepted < allNodes.size() - 1)
+ || votingLogList.size() > config.getMaxNumOfLogsInMem())
&& alreadyWait < ClusterConstant.getWriteOperationTimeoutMS()
&& !log.getStronglyAcceptedNodeIds().contains(Integer.MAX_VALUE)) {
try {
@@ -1863,7 +1832,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
* heartbeat timer.
*
* @param fromLeader true if the request is from a leader, false if the
request is from an
- * elector.
+ * elector.
*/
public void stepDown(long newTerm, boolean fromLeader) {
synchronized (term) {
@@ -1895,9 +1864,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
this.thisNode = thisNode;
}
- /**
- * @return the header of the data raft group or null if this is in a meta
group.
- */
+ /** @return the header of the data raft group or null if this is in a meta
group. */
public RaftNode getHeader() {
return null;
}
@@ -2065,9 +2032,7 @@ public abstract class RaftMember implements
RaftMemberMBean {
log, node, leaderShipStale, newLeaderTerm, request, quorumSize,
Collections.emptyList());
}
- /**
- * Send "log" to "node".
- */
+ /** Send "log" to "node". */
public void sendLogToFollower(
VotingLog log,
Node node,
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index 33f1ca7..356c572 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -103,6 +103,9 @@ public abstract class PhysicalPlan {
// a bridge from a cluster raft log to a physical plan
protected long index;
+ // if set, only the associated leader can execute the plan to guarantee
serializability and other
+ // leaders should return a LEADER_CHANGED response
+ protected long targetedTerm = -1;
private boolean debug;
@@ -564,4 +567,12 @@ public abstract class PhysicalPlan {
* @throws QueryProcessException when the check fails
*/
public void checkIntegrity() throws QueryProcessException {}
+
+ public long getTargetedTerm() {
+ return targetedTerm;
+ }
+
+ public void setTargetedTerm(long targetedTerm) {
+ this.targetedTerm = targetedTerm;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 4516588..76fdb33 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -380,6 +380,8 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
try {
PhysicalPlan physicalPlan =
processor.parseSQLToPhysicalPlan(statement,
sessionManager.getZoneId(req.sessionId));
+ physicalPlan.setTargetedTerm(req.latestTerm);
+
if (physicalPlan.isQuery() || physicalPlan.isSelectInto()) {
throw new QueryInBatchStatementException(statement);
}
@@ -480,6 +482,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
PhysicalPlan physicalPlan =
processor.parseSQLToPhysicalPlan(statement,
sessionManager.getZoneId(req.getSessionId()));
+ physicalPlan.setTargetedTerm(req.latestTerm);
return physicalPlan.isQuery()
? internalExecuteQueryStatement(
@@ -1112,6 +1115,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
try {
PhysicalPlan physicalPlan =
processor.parseSQLToPhysicalPlan(req.statement,
sessionManager.getZoneId(req.sessionId));
+ physicalPlan.setTargetedTerm(req.latestTerm);
return physicalPlan.isQuery()
? RpcUtils.getTSExecuteStatementResp(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query
statement.")
@@ -1155,7 +1159,8 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
return status != null
? new TSExecuteStatementResp(status)
: RpcUtils.getTSExecuteStatementResp(executeNonQueryPlan(plan))
- .setQueryId(sessionManager.requestQueryId(false));
+ .setQueryId(sessionManager.requestQueryId(false))
+ .setOperationType(plan.getOperatorType().name());
}
protected void handleClientExit() {
@@ -1263,6 +1268,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
e, OperationType.INSERT_RECORDS,
TSStatusCode.INTERNAL_SERVER_ERROR));
}
}
+ insertRowsPlan.setTargetedTerm(req.latestTerm);
TSStatus tsStatus = executeNonQueryPlan(insertRowsPlan);
return judgeFinalTsStatus(
@@ -1314,6 +1320,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
req.getValuesList().toArray(new ByteBuffer[0]),
req.isAligned);
TSStatus status = checkAuthority(plan, req.getSessionId());
+ plan.setTargetedTerm(req.latestTerm);
statusList.add(status != null ? status : executeNonQueryPlan(plan));
} catch (IoTDBException e) {
statusList.add(
@@ -1383,6 +1390,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
allCheckSuccess = false;
}
}
+ insertRowsPlan.setTargetedTerm(req.latestTerm);
TSStatus tsStatus = executeNonQueryPlan(insertRowsPlan);
return judgeFinalTsStatus(
@@ -1470,6 +1478,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
req.values,
req.isAligned);
TSStatus status = checkAuthority(plan, req.getSessionId());
+ plan.setTargetedTerm(req.latestTerm);
return status != null ? status : executeNonQueryPlan(plan);
} catch (IoTDBException e) {
return onIoTDBException(e, OperationType.INSERT_RECORD,
e.getErrorCode());
@@ -1501,6 +1510,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
plan.setNeedInferType(true);
plan.setAligned(req.isAligned);
TSStatus status = checkAuthority(plan, req.getSessionId());
+ plan.setTargetedTerm(req.latestTerm);
return status != null ? status : executeNonQueryPlan(plan);
} catch (IoTDBException e) {
return onIoTDBException(e, OperationType.INSERT_STRING_RECORD,
e.getErrorCode());
@@ -1556,7 +1566,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
insertTabletPlan.setDataTypes(req.types);
insertTabletPlan.setAligned(req.isAligned);
TSStatus status = checkAuthority(insertTabletPlan, req.getSessionId());
-
+ insertTabletPlan.setTargetedTerm(req.latestTerm);
return status != null ? status : executeNonQueryPlan(insertTabletPlan);
} catch (IoTDBException e) {
return onIoTDBException(e, OperationType.INSERT_TABLET,
e.getErrorCode());
@@ -1626,6 +1636,7 @@ public class TSServiceImpl extends BasicServiceProvider
implements TSIService.If
}
insertMultiTabletPlan.setInsertTabletPlanList(insertTabletPlanList);
+ insertMultiTabletPlan.setTargetedTerm(req.latestTerm);
return executeNonQueryPlan(insertMultiTabletPlan);
}
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index ebc1354..51ac057 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -27,6 +27,7 @@ public enum TSStatusCode {
STILL_EXECUTING_STATUS(201),
INVALID_HANDLE_STATUS(202),
INCOMPATIBLE_VERSION(203),
+ WEAKLY_ACCEPTED(204),
NODE_DELETE_FAILED_ERROR(298),
ALIAS_ALREADY_EXIST_ERROR(299),
@@ -101,6 +102,7 @@ public enum TSStatusCode {
NO_CONNECTION(706),
NEED_REDIRECTION(707),
PARSE_LOG_ERROR(708),
+ LEADER_CHANGED(709),
// configuration
CONFIG_ERROR(800);
diff --git
a/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
b/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index de6350e..2c061c3 100644
--- a/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++ b/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.rpc.RedirectException;
import org.apache.iotdb.rpc.RpcTransportFactory;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.EndPoint;
import org.apache.iotdb.service.rpc.thrift.TSAppendSchemaTemplateReq;
import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
@@ -83,6 +84,9 @@ public class SessionConnection {
private List<EndPoint> endPointList = new ArrayList<>();
private boolean enableRedirect = false;
+ private List<UnconfirmedRequest> unconfirmedRequests = new ArrayList<>();
+ private long latestTerm;
+
// TestOnly
public SessionConnection() {}
@@ -364,16 +368,26 @@ public class SessionConnection {
protected void executeNonQueryStatement(String sql)
throws IoTDBConnectionException, StatementExecutionException {
TSExecuteStatementReq execReq = new TSExecuteStatementReq(sessionId, sql,
statementId);
+ TSExecuteStatementResp execResp;
try {
execReq.setEnableRedirectQuery(enableRedirect);
- TSExecuteStatementResp execResp = client.executeUpdateStatement(execReq);
- RpcUtils.verifySuccess(execResp.getStatus());
+ execResp = client.executeUpdateStatement(execReq);
+ if (execResp.operationType != null &&
execResp.operationType.contains("INSERT")) {
+ verifyInsertionSuccess(execResp.getStatus(), execReq);
+ } else {
+ RpcUtils.verifySuccess(execResp.getStatus());
+ }
} catch (TException e) {
if (reconnect()) {
try {
execReq.setSessionId(sessionId);
execReq.setStatementId(statementId);
-
RpcUtils.verifySuccess(client.executeUpdateStatement(execReq).status);
+ execResp = client.executeUpdateStatement(execReq);
+ if (execResp.operationType != null &&
execResp.operationType.contains("INSERT")) {
+ verifyInsertionSuccess(execResp.getStatus(), execReq);
+ } else {
+ RpcUtils.verifySuccess(execResp.getStatus());
+ }
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -462,13 +476,14 @@ public class SessionConnection {
protected void insertRecord(TSInsertRecordReq request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
- RpcUtils.verifySuccessWithRedirection(client.insertRecord(request));
+ verifyInsertionSuccessWithRedirection(client.insertRecord(request),
request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertRecord(request));
+ verifyInsertionSuccess(client.insertRecord(request), request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -481,13 +496,14 @@ public class SessionConnection {
protected void insertRecord(TSInsertStringRecordReq request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
-
RpcUtils.verifySuccessWithRedirection(client.insertStringRecord(request));
+
verifyInsertionSuccessWithRedirection(client.insertStringRecord(request),
request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertStringRecord(request));
+ verifyInsertionSuccess(client.insertStringRecord(request), request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -500,14 +516,15 @@ public class SessionConnection {
protected void insertRecords(TSInsertRecordsReq request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
- RpcUtils.verifySuccessWithRedirectionForMultiDevices(
- client.insertRecords(request), request.getPrefixPaths());
+ verifyInsertionWithRedirectionForMultiDevices(
+ client.insertRecords(request), request.getPrefixPaths(), request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertRecords(request));
+ verifyInsertionSuccess(client.insertRecords(request), request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -520,14 +537,15 @@ public class SessionConnection {
protected void insertRecords(TSInsertStringRecordsReq request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
- RpcUtils.verifySuccessWithRedirectionForMultiDevices(
- client.insertStringRecords(request), request.getPrefixPaths());
+ verifyInsertionWithRedirectionForMultiDevices(
+ client.insertStringRecords(request), request.getPrefixPaths(),
request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertStringRecords(request));
+ verifyInsertionSuccess(client.insertStringRecords(request), request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -540,13 +558,14 @@ public class SessionConnection {
protected void insertRecordsOfOneDevice(TSInsertRecordsOfOneDeviceReq
request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
-
RpcUtils.verifySuccessWithRedirection(client.insertRecordsOfOneDevice(request));
+
verifyInsertionSuccessWithRedirection(client.insertRecordsOfOneDevice(request),
request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertRecordsOfOneDevice(request));
+ verifyInsertionSuccess(client.insertRecordsOfOneDevice(request),
request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -559,13 +578,14 @@ public class SessionConnection {
protected void insertTablet(TSInsertTabletReq request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
- RpcUtils.verifySuccessWithRedirection(client.insertTablet(request));
+ verifyInsertionSuccessWithRedirection(client.insertTablet(request),
request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertTablet(request));
+ verifyInsertionSuccess(client.insertTablet(request), request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -578,14 +598,15 @@ public class SessionConnection {
protected void insertTablets(TSInsertTabletsReq request)
throws IoTDBConnectionException, StatementExecutionException,
RedirectException {
request.setSessionId(sessionId);
+ request.setLatestTerm(latestTerm);
try {
- RpcUtils.verifySuccessWithRedirectionForMultiDevices(
- client.insertTablets(request), request.getPrefixPaths());
+ verifyInsertionWithRedirectionForMultiDevices(
+ client.insertTablets(request), request.getPrefixPaths(), request);
} catch (TException e) {
if (reconnect()) {
try {
request.setSessionId(sessionId);
- RpcUtils.verifySuccess(client.insertTablets(request));
+ verifyInsertionSuccess(client.insertTablets(request), request);
} catch (TException tException) {
throw new IoTDBConnectionException(tException);
}
@@ -913,8 +934,128 @@ public class SessionConnection {
this.endPoint = endPoint;
}
+ private void verifyInsertionSuccess(TSStatus status, Object request)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ onStronglyAccepted(status.getMessage());
+ } else if (status.getCode() ==
TSStatusCode.WEAKLY_ACCEPTED.getStatusCode()) {
+ onWeaklyAccepted(status.getMessage(), request);
+ } else if (status.getCode() ==
TSStatusCode.LEADER_CHANGED.getStatusCode()) {
+ onLeaderChange(status.getMessage(), request);
+ } else {
+ RpcUtils.verifySuccess(status);
+ }
+ }
+
+ private void verifyInsertionSuccessWithRedirection(TSStatus status, Object
request)
+ throws StatementExecutionException, RedirectException,
IoTDBConnectionException {
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ onStronglyAccepted(status.getMessage());
+ } else if (status.getCode() ==
TSStatusCode.WEAKLY_ACCEPTED.getStatusCode()) {
+ onWeaklyAccepted(status.getMessage(), request);
+ } else if (status.getCode() ==
TSStatusCode.LEADER_CHANGED.getStatusCode()) {
+ onLeaderChange(status.getMessage(), request);
+ } else {
+ RpcUtils.verifySuccessWithRedirection(status);
+ }
+ }
+
+ private void verifyInsertionWithRedirectionForMultiDevices(
+ TSStatus status, List<String> devices, Object request)
+ throws StatementExecutionException, RedirectException,
IoTDBConnectionException {
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ onStronglyAccepted(status.getMessage());
+ } else if (status.getCode() ==
TSStatusCode.WEAKLY_ACCEPTED.getStatusCode()) {
+ onWeaklyAccepted(status.getMessage(), request);
+ } else if (status.getCode() ==
TSStatusCode.LEADER_CHANGED.getStatusCode()) {
+ onLeaderChange(status.getMessage(), request);
+ } else {
+ RpcUtils.verifySuccessWithRedirectionForMultiDevices(status, devices);
+ }
+ }
+
+ private void onStronglyAccepted(String message) {
+ if (message == null) {
+ return;
+ }
+ try {
+ String[] split = message.split("-");
+ long index = Long.parseLong(split[0]);
+ long term = Long.parseLong(split[1]);
+ if (!unconfirmedRequests.isEmpty() && unconfirmedRequests.get(0).term ==
term) {
+ unconfirmedRequests.removeIf(r -> r.index < index);
+ }
+ latestTerm = term;
+ } catch (IndexOutOfBoundsException | NumberFormatException e) {
+ // ignore
+ }
+ }
+
+ private void onWeaklyAccepted(String message, Object request) {
+ if (message == null) {
+ return;
+ }
+ try {
+ String[] split = message.split("-");
+ long index = Long.parseLong(split[0]);
+ long term = Long.parseLong(split[1]);
+ UnconfirmedRequest unconfirmedRequest = new UnconfirmedRequest();
+ unconfirmedRequest.index = index;
+ unconfirmedRequest.term = term;
+ unconfirmedRequest.request = request;
+ unconfirmedRequests.add(unconfirmedRequest);
+ } catch (IndexOutOfBoundsException | NumberFormatException e) {
+ // ignore
+ }
+ }
+
+ private void onLeaderChange(String message, Object request)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (message == null) {
+ return;
+ }
+ try {
+ latestTerm = Long.parseLong(message);
+ List<UnconfirmedRequest> tempUnconfirmedRequests = new
ArrayList<>(unconfirmedRequests);
+ unconfirmedRequests.clear();
+ for (UnconfirmedRequest tempUnconfirmedRequest :
tempUnconfirmedRequests) {
+ retryInsertion(tempUnconfirmedRequest.request);
+ }
+ retryInsertion(request);
+ } catch (IndexOutOfBoundsException | NumberFormatException |
RedirectException e) {
+ // ignore
+ }
+ }
+
+ private void retryInsertion(Object request)
+ throws StatementExecutionException, IoTDBConnectionException,
RedirectException {
+ if (request instanceof TSInsertRecordReq) {
+ insertRecord(((TSInsertRecordReq) request));
+ } else if (request instanceof TSInsertStringRecordReq) {
+ insertRecord(((TSInsertStringRecordReq) request));
+ } else if (request instanceof TSInsertRecordsReq) {
+ insertRecords(((TSInsertRecordsReq) request));
+ } else if (request instanceof TSInsertStringRecordsReq) {
+ insertRecords(((TSInsertStringRecordsReq) request));
+ } else if (request instanceof TSInsertRecordsOfOneDeviceReq) {
+ insertRecordsOfOneDevice(((TSInsertRecordsOfOneDeviceReq) request));
+ } else if (request instanceof TSInsertTabletReq) {
+ insertTablet(((TSInsertTabletReq) request));
+ } else if (request instanceof TSInsertTabletsReq) {
+ insertTablets(((TSInsertTabletsReq) request));
+ } else {
+ logger.error("Unknown request type during retry: {}", request);
+ }
+ }
+
@Override
public String toString() {
return "SessionConnection{" + " endPoint=" + endPoint + "}";
}
+
+ static class UnconfirmedRequest {
+ private long index;
+ private long term;
+ private Object request;
+ }
}
diff --git a/thrift/src/main/thrift/rpc.thrift
b/thrift/src/main/thrift/rpc.thrift
index a981a8c..534cd17 100644
--- a/thrift/src/main/thrift/rpc.thrift
+++ b/thrift/src/main/thrift/rpc.thrift
@@ -138,6 +138,8 @@ struct TSExecuteStatementReq {
6: optional bool enableRedirectQuery;
7: optional bool jdbcQuery;
+
+ 8: optional i64 latestTerm
}
struct TSExecuteBatchStatementReq{
@@ -146,6 +148,8 @@ struct TSExecuteBatchStatementReq{
// The statements to be executed (DML, DDL, SET, etc)
2: required list<string> statements
+
+ 3: optional i64 latestTerm
}
struct TSGetOperationStatusReq {
@@ -218,6 +222,7 @@ struct TSInsertRecordReq {
4: required binary values
5: required i64 timestamp
6: optional bool isAligned
+ 7: optional i64 latestTerm
}
struct TSInsertStringRecordReq {
@@ -227,6 +232,7 @@ struct TSInsertStringRecordReq {
4: required list<string> values
5: required i64 timestamp
6: optional bool isAligned
+ 7: optional i64 latestTerm
}
struct TSInsertTabletReq {
@@ -238,6 +244,7 @@ struct TSInsertTabletReq {
6: required list<i32> types
7: required i32 size
8: optional bool isAligned
+ 9: optional i64 latestTerm
}
struct TSInsertTabletsReq {
@@ -249,6 +256,7 @@ struct TSInsertTabletsReq {
6: required list<list<i32>> typesList
7: required list<i32> sizeList
8: optional bool isAligned
+ 9: optional i64 latestTerm
}
struct TSInsertRecordsReq {
@@ -258,6 +266,7 @@ struct TSInsertRecordsReq {
4: required list<binary> valuesList
5: required list<i64> timestamps
6: optional bool isAligned
+ 7: optional i64 latestTerm
}
struct TSInsertRecordsOfOneDeviceReq {
@@ -267,6 +276,7 @@ struct TSInsertRecordsOfOneDeviceReq {
4: required list<binary> valuesList
5: required list<i64> timestamps
6: optional bool isAligned
+ 7: optional i64 latestTerm
}
struct TSInsertStringRecordsReq {
@@ -276,6 +286,7 @@ struct TSInsertStringRecordsReq {
4: required list<list<string>> valuesList
5: required list<i64> timestamps
6: optional bool isAligned
+ 7: optional i64 latestTerm
}
struct TSDeleteDataReq {
