Skip to content

Commit cddb4b1

Browse files
CaideyipiCaideyipi
committed
Address ConfigNode review feedback
1 parent 2e4c652 commit cddb4b1

7 files changed

Lines changed: 70 additions & 9 deletions

File tree

‎iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,9 +56,9 @@ public final class ConfigNodeMessages {
5656
public static final String PERSIST_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
5757
"Persist current ConfigPhysicalPlan for ConfigNode SimpleConsensus mode failed";
5858
public static final String FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG =
59-
"Failed to rollback persisted ConfigNode SimpleConsensus log.";
59+
"Failed to rollback persisted ConfigNode SimpleConsensus log. The persisted plan may be replayed after restart.";
6060
public static final String ROLLBACK_FAILED_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
61-
"Rollback failed ConfigPhysicalPlan for ConfigNode SimpleConsensus mode failed";
61+
"Rollback failed ConfigPhysicalPlan for ConfigNode SimpleConsensus mode failed, planType: {}, logFile: {}, truncateOffset: {}, endIndexBeforeWrite: {}";
6262
public static final String SEAL_RECOVERED_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED =
6363
"Seal recovered ConfigNode SimpleConsensus log failed: {}";
6464
public static final String SIMPLECONSENSUS_LOG_WRITER_IS_NOT_INITIALIZED =

‎iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,9 @@ public final class ConfigNodeMessages {
5555
public static final String PERSIST_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
5656
"Persist current ConfigPhysicalPlan for ConfigNode SimpleConsensus mode failed";
5757
public static final String FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG =
58-
"Failed to rollback persisted ConfigNode SimpleConsensus log.";
58+
"Failed to rollback persisted ConfigNode SimpleConsensus log. The persisted plan may be replayed after restart.";
5959
public static final String ROLLBACK_FAILED_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
60-
"Rollback failed ConfigPhysicalPlan for ConfigNode SimpleConsensus mode failed";
60+
"Rollback failed ConfigPhysicalPlan for ConfigNode SimpleConsensus mode failed, planType: {}, logFile: {}, truncateOffset: {}, endIndexBeforeWrite: {}";
6161
public static final String SEAL_RECOVERED_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED =
6262
"Seal recovered ConfigNode SimpleConsensus log failed: {}";
6363
public static final String SIMPLECONSENSUS_LOG_WRITER_IS_NOT_INITIALIZED =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,7 @@ protected TSStatus write(ConfigPhysicalPlan plan) {
141141

142142
if (persistResult != null
143143
&& result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
144-
&& !rollbackFailedPlanForSimpleConsensus(persistResult)) {
144+
&& !rollbackFailedPlanForSimpleConsensus(plan, persistResult)) {
145145
final String rollbackFailureMessage =
146146
ConfigNodeMessages.FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG;
147147
result.setMessage(
@@ -418,7 +418,8 @@ private SimpleConsensusPersistResult persistPlanForSimpleConsensus(ConfigPhysica
418418
persistedLogFile, logFileSizeBeforeWrite, endIndexBeforeWrite);
419419
}
420420

421-
private boolean rollbackFailedPlanForSimpleConsensus(SimpleConsensusPersistResult persistResult) {
421+
private boolean rollbackFailedPlanForSimpleConsensus(
422+
ConfigPhysicalPlan plan, SimpleConsensusPersistResult persistResult) {
422423
closeSimpleLogWriter();
423424
try (FileOutputStream outputStream = new FileOutputStream(persistResult.logFile, true);
424425
FileChannel channel = outputStream.getChannel()) {
@@ -432,6 +433,10 @@ private boolean rollbackFailedPlanForSimpleConsensus(SimpleConsensusPersistResul
432433
LOGGER.error(
433434
ConfigNodeMessages
434435
.ROLLBACK_FAILED_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED,
436+
plan.getType(),
437+
persistResult.logFile,
438+
persistResult.logFileSizeBeforeWrite,
439+
persistResult.endIndexBeforeWrite,
435440
e);
436441
return false;
437442
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -348,9 +348,8 @@ public DataSet registerDataNode(TDataNodeRegisterReq req) {
348348
return resp;
349349
}
350350

351-
// Create a new DataNodeHeartbeatCache and force update NodeStatus
351+
// Create a new DataNodeHeartbeatCache. The heartbeat service will refresh NodeStatus shortly.
352352
getLoadManager().getLoadCache().createNodeHeartbeatCache(NodeType.DataNode, dataNodeId);
353-
// TODO: invoke a force heartbeat to update new DataNode's status immediately
354353

355354
// Bind DataNode metrics
356355
PartitionMetrics.bindDataNodePartitionMetricsWhenUpdate(

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachineTest.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
2626
import org.apache.iotdb.confignode.consensus.request.TestOnlyPlan;
2727
import org.apache.iotdb.confignode.persistence.executor.ConfigPlanExecutor;
28+
import org.apache.iotdb.confignode.writelog.io.SingleFileLogReader;
2829
import org.apache.iotdb.consensus.ConsensusFactory;
2930
import org.apache.iotdb.db.utils.writelog.LogWriter;
3031
import org.apache.iotdb.rpc.TSStatusCode;
@@ -81,6 +82,8 @@ public void testFailedSimpleConsensusWriteRollsBackPersistedPlan() throws Except
8182
Assert.assertEquals(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), status.getCode());
8283
Assert.assertEquals(0, Files.size(tempLogFile));
8384
Assert.assertEquals(0, getField(stateMachine, "endIndex"));
85+
closeSimpleLogWriter(stateMachine);
86+
assertNoReplayablePlans(tempLogFile);
8487
} finally {
8588
closeSimpleLogWriter(stateMachine);
8689
Files.deleteIfExists(tempLogFile);
@@ -100,6 +103,15 @@ private static Object getField(Object target, String fieldName) throws Exception
100103
return field.get(target);
101104
}
102105

106+
private static void assertNoReplayablePlans(Path tempLogFile) throws Exception {
107+
SingleFileLogReader logReader = new SingleFileLogReader(tempLogFile.toFile());
108+
try {
109+
Assert.assertFalse(logReader.hasNext());
110+
} finally {
111+
logReader.close();
112+
}
113+
}
114+
103115
private static void closeSimpleLogWriter(ConfigRegionStateMachine stateMachine) throws Exception {
104116
LogWriter logWriter = (LogWriter) getField(stateMachine, "simpleLogWriter");
105117
if (logWriter != null) {

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,11 @@
2626
import org.apache.iotdb.common.rpc.thrift.TNodeResource;
2727
import org.apache.iotdb.commons.exception.StartupException;
2828
import org.apache.iotdb.confignode.consensus.request.write.confignode.ApplyConfigNodePlan;
29+
import org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateVersionInfoPlan;
2930
import org.apache.iotdb.confignode.consensus.request.write.datanode.RegisterDataNodePlan;
3031
import org.apache.iotdb.confignode.persistence.node.NodeInfo;
32+
import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
33+
import org.apache.iotdb.rpc.TSStatusCode;
3134

3235
import org.apache.thrift.TException;
3336
import org.apache.tsfile.external.commons.io.FileUtils;
@@ -73,6 +76,49 @@ public void testSnapshot() throws TException, IOException {
7376
Assert.assertEquals(nodeInfo, nodeInfo1);
7477
}
7578

79+
@Test
80+
public void testRegistrationPlansAreIdempotentForWalReplay() {
81+
NodeInfo replayNodeInfo = new NodeInfo();
82+
83+
TDataNodeConfiguration dataNodeConfiguration = generateTDataNodeConfiguration(100);
84+
RegisterDataNodePlan registerDataNodePlan = new RegisterDataNodePlan(dataNodeConfiguration);
85+
Assert.assertEquals(
86+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
87+
replayNodeInfo.registerDataNode(registerDataNodePlan).getCode());
88+
Assert.assertEquals(
89+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
90+
replayNodeInfo.registerDataNode(registerDataNodePlan).getCode());
91+
Assert.assertEquals(1, replayNodeInfo.getRegisteredDataNodeCount());
92+
Assert.assertEquals(
93+
dataNodeConfiguration,
94+
replayNodeInfo.getRegisteredDataNode(dataNodeConfiguration.getLocation().getDataNodeId()));
95+
96+
TConfigNodeLocation configNodeLocation =
97+
new TConfigNodeLocation(
98+
20000, new TEndPoint("127.0.0.1", 22200), new TEndPoint("127.0.0.1", 22300));
99+
ApplyConfigNodePlan applyConfigNodePlan = new ApplyConfigNodePlan(configNodeLocation);
100+
Assert.assertEquals(
101+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
102+
replayNodeInfo.applyConfigNode(applyConfigNodePlan).getCode());
103+
Assert.assertEquals(
104+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
105+
replayNodeInfo.applyConfigNode(applyConfigNodePlan).getCode());
106+
Assert.assertEquals(1, replayNodeInfo.getRegisteredConfigNodes().size());
107+
Assert.assertEquals(configNodeLocation, replayNodeInfo.getRegisteredConfigNodes().get(0));
108+
109+
TNodeVersionInfo versionInfo = new TNodeVersionInfo("version", "build");
110+
UpdateVersionInfoPlan updateVersionInfoPlan =
111+
new UpdateVersionInfoPlan(versionInfo, configNodeLocation.getConfigNodeId());
112+
Assert.assertEquals(
113+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
114+
replayNodeInfo.updateVersionInfo(updateVersionInfoPlan).getCode());
115+
Assert.assertEquals(
116+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
117+
replayNodeInfo.updateVersionInfo(updateVersionInfoPlan).getCode());
118+
Assert.assertEquals(
119+
versionInfo, replayNodeInfo.getVersionInfo(configNodeLocation.getConfigNodeId()));
120+
}
121+
76122
private void registerConfigNodes() {
77123
for (int i = 0; i < 3; i++) {
78124
ApplyConfigNodePlan applyConfigNodePlan =

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/path/PartialPath.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -427,7 +427,6 @@ public boolean matchFullPath(PartialPath rPath) {
427427
}
428428

429429
public boolean matchFullPath(IDeviceID deviceID, String measurement) {
430-
// TODO change this way
431430
PartialPath devicePath;
432431
try {
433432
devicePath = new PartialPath(deviceID.toString());

0 commit comments

Comments
 (0)