You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2021/08/26 10:51:30 UTC
[GitHub] [iotdb] chengjianyun opened a new pull request #3848: [discussion-3833] refine exception handling logic in commitTo in Raft…
chengjianyun opened a new pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848
The idea come from the discussion: https://github.com/apache/iotdb/discussions/3833
Here I want to refine the error handling logic in `commitTo` to make cluster more robust and make code a bit more clear.
The idea here is don't let Raft group run into a invalid state when some error happened from Raft log persistence. And refine try catch block make the code clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] LebronAl merged pull request #3848: [ISSUE-3856] refine exception handling logic in commitTo in Raft…
Posted by GitBox <gi...@apache.org>.
LebronAl merged pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #3848: [discussion-3833] refine exception handling logic in commitTo in Raft…
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848#discussion_r701751910
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
##########
@@ -585,71 +585,87 @@ public void commitTo(long newCommitIndex) throws LogExecutionException {
.subList(0, (int) (getCommitLogIndex() - entries.get(0).getCurrLogIndex() + 1))
.clear();
}
- try {
- boolean needToCompactLog = false;
- int numToReserveForNew = minNumOfLogsInMem;
- if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
- needToCompactLog = true;
- numToReserveForNew = maxNumOfLogsInMem - entries.size();
- }
- long newEntryMemSize = 0;
- for (Log entry : entries) {
- if (entry.getByteSize() == 0) {
- logger.debug(
- "{} should not go here, must be send to the follower, "
- + "so the log has been serialized exclude single node mode",
- entry);
- entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
- }
- newEntryMemSize += entry.getByteSize();
- }
- int sizeToReserveForNew = minNumOfLogsInMem;
- if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
- needToCompactLog = true;
- sizeToReserveForNew =
- committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ boolean needToCompactLog = false;
+ int numToReserveForNew = minNumOfLogsInMem;
+ if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
+ needToCompactLog = true;
+ numToReserveForNew = maxNumOfLogsInMem - entries.size();
+ }
+
+ long newEntryMemSize = 0;
+ for (Log entry : entries) {
+ if (entry.getByteSize() == 0) {
+ logger.debug(
+ "{} should not go here, must be send to the follower, "
+ + "so the log has been serialized exclude single node mode",
+ entry);
+ entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
}
+ newEntryMemSize += entry.getByteSize();
+ }
+ int sizeToReserveForNew = minNumOfLogsInMem;
+ if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
+ needToCompactLog = true;
+ sizeToReserveForNew =
+ committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ }
- if (needToCompactLog) {
- int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
- int sizeToReserveForConfig = minNumOfLogsInMem;
- startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
- synchronized (this) {
- innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
- }
- Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ if (needToCompactLog) {
+ int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
+ int sizeToReserveForConfig = minNumOfLogsInMem;
+ startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
+ synchronized (this) {
+ innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
}
+ Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ }
- startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ try {
+ // As the operations here are so simple that we can think the execution
+ // are success or fail at same time approximately.
Review comment:
Thank you! Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] LebronAl commented on a change in pull request #3848: [discussion-3833] refine exception handling logic in commitTo in Raft…
Posted by GitBox <gi...@apache.org>.
LebronAl commented on a change in pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848#discussion_r700835648
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
##########
@@ -585,71 +585,87 @@ public void commitTo(long newCommitIndex) throws LogExecutionException {
.subList(0, (int) (getCommitLogIndex() - entries.get(0).getCurrLogIndex() + 1))
.clear();
}
- try {
- boolean needToCompactLog = false;
- int numToReserveForNew = minNumOfLogsInMem;
- if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
- needToCompactLog = true;
- numToReserveForNew = maxNumOfLogsInMem - entries.size();
- }
- long newEntryMemSize = 0;
- for (Log entry : entries) {
- if (entry.getByteSize() == 0) {
- logger.debug(
- "{} should not go here, must be send to the follower, "
- + "so the log has been serialized exclude single node mode",
- entry);
- entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
- }
- newEntryMemSize += entry.getByteSize();
- }
- int sizeToReserveForNew = minNumOfLogsInMem;
- if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
- needToCompactLog = true;
- sizeToReserveForNew =
- committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ boolean needToCompactLog = false;
+ int numToReserveForNew = minNumOfLogsInMem;
+ if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
+ needToCompactLog = true;
+ numToReserveForNew = maxNumOfLogsInMem - entries.size();
+ }
+
+ long newEntryMemSize = 0;
+ for (Log entry : entries) {
+ if (entry.getByteSize() == 0) {
+ logger.debug(
+ "{} should not go here, must be send to the follower, "
+ + "so the log has been serialized exclude single node mode",
+ entry);
+ entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
}
+ newEntryMemSize += entry.getByteSize();
+ }
+ int sizeToReserveForNew = minNumOfLogsInMem;
+ if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
+ needToCompactLog = true;
+ sizeToReserveForNew =
+ committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ }
- if (needToCompactLog) {
- int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
- int sizeToReserveForConfig = minNumOfLogsInMem;
- startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
- synchronized (this) {
- innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
- }
- Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ if (needToCompactLog) {
+ int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
+ int sizeToReserveForConfig = minNumOfLogsInMem;
+ startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
+ synchronized (this) {
+ innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
}
+ Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ }
- startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ try {
+ // As the operations here are so simple that we can think the execution
+ // are success or fail at same time approximately.
getCommittedEntryManager().append(entries);
- if (ClusterDescriptor.getInstance().getConfig().isEnableRaftLogPersistence()) {
- getStableEntryManager().append(entries, maxHaveAppliedCommitIndex);
- }
Log lastLog = entries.get(entries.size() - 1);
getUnCommittedEntryManager().stableTo(lastLog.getCurrLogIndex());
- Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.calOperationCostTimeFromStart(startTime);
-
commitIndex = lastLog.getCurrLogIndex();
- startTime = Statistic.RAFT_SENDER_COMMIT_APPLY_LOGS.getOperationStartTime();
- applyEntries(entries);
- Statistic.RAFT_SENDER_COMMIT_APPLY_LOGS.calOperationCostTimeFromStart(startTime);
- long unappliedLogSize = commitLogIndex - maxHaveAppliedCommitIndex;
- if (unappliedLogSize > ClusterDescriptor.getInstance().getConfig().getMaxNumOfLogsInMem()) {
- logger.debug(
- "There are too many unapplied logs [{}], wait for a while to avoid memory "
- + "overflow",
- unappliedLogSize);
- Thread.sleep(
- unappliedLogSize - ClusterDescriptor.getInstance().getConfig().getMaxNumOfLogsInMem());
+ if (ClusterDescriptor.getInstance().getConfig().isEnableRaftLogPersistence()) {
+ // Cluster could continue provide service when exception is thrown here
+ getStableEntryManager().append(entries, maxHaveAppliedCommitIndex);
}
} catch (TruncateCommittedEntryException e) {
+ // fatal error, node won't recover from the error anymore
+ // TODO: let node quit the raft group once encounter the error
logger.error("{}: Unexpected error:", name, e);
} catch (IOException e) {
+ // The exception happens because persistent entries fail which should block the service
+ // continue accept data. We need to know that since then, the persisted raft log for the
Review comment:
grammer error?
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
##########
@@ -585,71 +585,87 @@ public void commitTo(long newCommitIndex) throws LogExecutionException {
.subList(0, (int) (getCommitLogIndex() - entries.get(0).getCurrLogIndex() + 1))
.clear();
}
- try {
- boolean needToCompactLog = false;
- int numToReserveForNew = minNumOfLogsInMem;
- if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
- needToCompactLog = true;
- numToReserveForNew = maxNumOfLogsInMem - entries.size();
- }
- long newEntryMemSize = 0;
- for (Log entry : entries) {
- if (entry.getByteSize() == 0) {
- logger.debug(
- "{} should not go here, must be send to the follower, "
- + "so the log has been serialized exclude single node mode",
- entry);
- entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
- }
- newEntryMemSize += entry.getByteSize();
- }
- int sizeToReserveForNew = minNumOfLogsInMem;
- if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
- needToCompactLog = true;
- sizeToReserveForNew =
- committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ boolean needToCompactLog = false;
+ int numToReserveForNew = minNumOfLogsInMem;
+ if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
+ needToCompactLog = true;
+ numToReserveForNew = maxNumOfLogsInMem - entries.size();
+ }
+
+ long newEntryMemSize = 0;
+ for (Log entry : entries) {
+ if (entry.getByteSize() == 0) {
+ logger.debug(
+ "{} should not go here, must be send to the follower, "
+ + "so the log has been serialized exclude single node mode",
+ entry);
+ entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
}
+ newEntryMemSize += entry.getByteSize();
+ }
+ int sizeToReserveForNew = minNumOfLogsInMem;
+ if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
+ needToCompactLog = true;
+ sizeToReserveForNew =
+ committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ }
- if (needToCompactLog) {
- int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
- int sizeToReserveForConfig = minNumOfLogsInMem;
- startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
- synchronized (this) {
- innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
- }
- Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ if (needToCompactLog) {
+ int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
+ int sizeToReserveForConfig = minNumOfLogsInMem;
+ startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
+ synchronized (this) {
+ innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
}
+ Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ }
- startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ try {
+ // As the operations here are so simple that we can think the execution
+ // are success or fail at same time approximately.
Review comment:
grammer error?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] coveralls edited a comment on pull request #3848: [discussion-3833] refine exception handling logic in commitTo in Raft…
Posted by GitBox <gi...@apache.org>.
coveralls edited a comment on pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848#issuecomment-906324765
[](https://coveralls.io/builds/42636481)
Coverage decreased (-0.1%) to 67.228% when pulling **ce96475341afa7ebd04cade3150c0481d769c297 on chengjianyun:refine/error-handling** into **228d62d622eb772f0317e40cc341aa0b49d38af2 on apache:master**.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #3848: [discussion-3833] refine exception handling logic in commitTo in Raft…
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848#discussion_r701751614
##########
File path: cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
##########
@@ -585,71 +585,87 @@ public void commitTo(long newCommitIndex) throws LogExecutionException {
.subList(0, (int) (getCommitLogIndex() - entries.get(0).getCurrLogIndex() + 1))
.clear();
}
- try {
- boolean needToCompactLog = false;
- int numToReserveForNew = minNumOfLogsInMem;
- if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
- needToCompactLog = true;
- numToReserveForNew = maxNumOfLogsInMem - entries.size();
- }
- long newEntryMemSize = 0;
- for (Log entry : entries) {
- if (entry.getByteSize() == 0) {
- logger.debug(
- "{} should not go here, must be send to the follower, "
- + "so the log has been serialized exclude single node mode",
- entry);
- entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
- }
- newEntryMemSize += entry.getByteSize();
- }
- int sizeToReserveForNew = minNumOfLogsInMem;
- if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
- needToCompactLog = true;
- sizeToReserveForNew =
- committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ boolean needToCompactLog = false;
+ int numToReserveForNew = minNumOfLogsInMem;
+ if (committedEntryManager.getTotalSize() + entries.size() > maxNumOfLogsInMem) {
+ needToCompactLog = true;
+ numToReserveForNew = maxNumOfLogsInMem - entries.size();
+ }
+
+ long newEntryMemSize = 0;
+ for (Log entry : entries) {
+ if (entry.getByteSize() == 0) {
+ logger.debug(
+ "{} should not go here, must be send to the follower, "
+ + "so the log has been serialized exclude single node mode",
+ entry);
+ entry.setByteSize((int) RamUsageEstimator.sizeOf(entry));
}
+ newEntryMemSize += entry.getByteSize();
+ }
+ int sizeToReserveForNew = minNumOfLogsInMem;
+ if (newEntryMemSize + committedEntryManager.getEntryTotalMemSize() > maxLogMemSize) {
+ needToCompactLog = true;
+ sizeToReserveForNew =
+ committedEntryManager.maxLogNumShouldReserve(maxLogMemSize - newEntryMemSize);
+ }
- if (needToCompactLog) {
- int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
- int sizeToReserveForConfig = minNumOfLogsInMem;
- startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
- synchronized (this) {
- innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
- }
- Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ if (needToCompactLog) {
+ int numForNew = Math.min(numToReserveForNew, sizeToReserveForNew);
+ int sizeToReserveForConfig = minNumOfLogsInMem;
+ startTime = Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.getOperationStartTime();
+ synchronized (this) {
+ innerDeleteLog(Math.min(sizeToReserveForConfig, numForNew));
}
+ Statistic.RAFT_SENDER_COMMIT_DELETE_EXCEEDING_LOGS.calOperationCostTimeFromStart(startTime);
+ }
- startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ startTime = Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.getOperationStartTime();
+ try {
+ // As the operations here are so simple that we can think the execution
+ // are success or fail at same time approximately.
getCommittedEntryManager().append(entries);
- if (ClusterDescriptor.getInstance().getConfig().isEnableRaftLogPersistence()) {
- getStableEntryManager().append(entries, maxHaveAppliedCommitIndex);
- }
Log lastLog = entries.get(entries.size() - 1);
getUnCommittedEntryManager().stableTo(lastLog.getCurrLogIndex());
- Statistic.RAFT_SENDER_COMMIT_APPEND_AND_STABLE_LOGS.calOperationCostTimeFromStart(startTime);
-
commitIndex = lastLog.getCurrLogIndex();
- startTime = Statistic.RAFT_SENDER_COMMIT_APPLY_LOGS.getOperationStartTime();
- applyEntries(entries);
- Statistic.RAFT_SENDER_COMMIT_APPLY_LOGS.calOperationCostTimeFromStart(startTime);
- long unappliedLogSize = commitLogIndex - maxHaveAppliedCommitIndex;
- if (unappliedLogSize > ClusterDescriptor.getInstance().getConfig().getMaxNumOfLogsInMem()) {
- logger.debug(
- "There are too many unapplied logs [{}], wait for a while to avoid memory "
- + "overflow",
- unappliedLogSize);
- Thread.sleep(
- unappliedLogSize - ClusterDescriptor.getInstance().getConfig().getMaxNumOfLogsInMem());
+ if (ClusterDescriptor.getInstance().getConfig().isEnableRaftLogPersistence()) {
+ // Cluster could continue provide service when exception is thrown here
+ getStableEntryManager().append(entries, maxHaveAppliedCommitIndex);
}
} catch (TruncateCommittedEntryException e) {
+ // fatal error, node won't recover from the error anymore
+ // TODO: let node quit the raft group once encounter the error
logger.error("{}: Unexpected error:", name, e);
} catch (IOException e) {
+ // The exception happens because persistent entries fail which should block the service
+ // continue accept data. We need to know that since then, the persisted raft log for the
Review comment:
Thanks for pointing it out. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] coveralls commented on pull request #3848: [discussion-3833] refine exception handling logic in commitTo in Raft…
Posted by GitBox <gi...@apache.org>.
coveralls commented on pull request #3848:
URL: https://github.com/apache/iotdb/pull/3848#issuecomment-906324765
[](https://coveralls.io/builds/42433690)
Coverage decreased (-0.003%) to 67.357% when pulling **8c439fee1e12bd5962c05b0a8696bd1c69619b0e on chengjianyun:refine/error-handling** into **228d62d622eb772f0317e40cc341aa0b49d38af2 on apache:master**.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org