Caideyipi commented on code in PR #17609:
URL: https://github.com/apache/iotdb/pull/17609#discussion_r3680780440
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java:
##########
@@ -121,6 +121,13 @@ public TSStatus write(IConsensusRequest request) {
/** Transmit {@link ConfigPhysicalPlan} to {@link ConfigPlanExecutor} */
protected TSStatus write(ConfigPhysicalPlan plan) {
+ if
(ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass()))
{
+ final TSStatus persistStatus = persistPlanForSimpleConsensus(plan);
+ if (persistStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return persistStatus;
+ }
+ }
Review Comment:
Addressed in cddb4b1c962. A plan whose execution fails is now removed from
the SimpleConsensus WAL by truncating and force-syncing back to the pre-write
offset, so normal rollback leaves nothing to replay.
`ConfigRegionStateMachineTest.testFailedSimpleConsensusWriteRollsBackPersistedPlan`
verifies the WAL contains no replayable plan. For the registration-related
plans touched here,
`NodeInfoTest.testRegistrationPlansAreIdempotentForWalReplay` also executes
RegisterDataNodePlan, ApplyConfigNodePlan, and UpdateVersionInfoPlan twice and
verifies stable state. If truncation itself fails, the status and operator log
explicitly warn that replay remains possible.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java:
##########
@@ -351,56 +360,48 @@ public boolean isReadOnly() {
return CommonDescriptor.getInstance().getConfig().isReadOnly();
}
- private void writeLogForSimpleConsensus(ConfigPhysicalPlan plan) {
- if (simpleLogFile.length() > LOG_FILE_MAX_SIZE) {
- try {
- simpleLogWriter.force();
- File completedFilePath = new File(FILE_PATH + startIndex + "_" +
endIndex);
- Files.move(
- simpleLogFile.toPath(), completedFilePath.toPath(),
StandardCopyOption.ATOMIC_MOVE);
- } catch (IOException e) {
- LOGGER.error("Can't force logWriter for ConfigNode SimpleConsensus
mode", e);
+ private TSStatus persistPlanForSimpleConsensus(ConfigPhysicalPlan plan) {
+ try {
+ if (simpleLogWriter == null || simpleLogFile == null) {
+ throw new IOException("SimpleConsensus log writer is not
initialized.");
}
- for (int retry = 0; retry < 5; retry++) {
- try {
- simpleLogWriter.close();
- } catch (IOException e) {
- LOGGER.warn(
- "Can't close StandAloneLog for ConfigNode SimpleConsensus mode, "
- + "filePath: {}, retry: {}",
- simpleLogFile.getAbsolutePath(),
- retry);
- try {
- // Sleep 1s and retry
- TimeUnit.SECONDS.sleep(1);
- } catch (InterruptedException e2) {
- Thread.currentThread().interrupt();
- LOGGER.warn("Unexpected interruption during the close method of
logWriter");
- }
- continue;
- }
- break;
+
+ if (simpleLogFile.length() > LOG_FILE_MAX_SIZE) {
+ rollSimpleConsensusLogFile();
}
- startIndex = endIndex + 1;
- createLogFile(startIndex);
- }
- try {
ByteBuffer buffer = plan.serializeToByteBuffer();
buffer.position(buffer.limit());
simpleLogWriter.write(buffer);
+ simpleLogWriter.force();
Review Comment:
Addressed. The scheduled flush executor and `flushWALForSimpleConsensus`
method were removed; each persisted write is synchronously forced, and the
writer is now closed from `stop()`.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java:
##########
@@ -351,56 +360,48 @@ public boolean isReadOnly() {
return CommonDescriptor.getInstance().getConfig().isReadOnly();
}
- private void writeLogForSimpleConsensus(ConfigPhysicalPlan plan) {
- if (simpleLogFile.length() > LOG_FILE_MAX_SIZE) {
- try {
- simpleLogWriter.force();
- File completedFilePath = new File(FILE_PATH + startIndex + "_" +
endIndex);
- Files.move(
- simpleLogFile.toPath(), completedFilePath.toPath(),
StandardCopyOption.ATOMIC_MOVE);
- } catch (IOException e) {
- LOGGER.error("Can't force logWriter for ConfigNode SimpleConsensus
mode", e);
+ private TSStatus persistPlanForSimpleConsensus(ConfigPhysicalPlan plan) {
+ try {
+ if (simpleLogWriter == null || simpleLogFile == null) {
+ throw new IOException("SimpleConsensus log writer is not
initialized.");
}
- for (int retry = 0; retry < 5; retry++) {
- try {
- simpleLogWriter.close();
- } catch (IOException e) {
- LOGGER.warn(
- "Can't close StandAloneLog for ConfigNode SimpleConsensus mode, "
- + "filePath: {}, retry: {}",
- simpleLogFile.getAbsolutePath(),
- retry);
- try {
- // Sleep 1s and retry
- TimeUnit.SECONDS.sleep(1);
- } catch (InterruptedException e2) {
- Thread.currentThread().interrupt();
- LOGGER.warn("Unexpected interruption during the close method of
logWriter");
- }
- continue;
- }
- break;
+
+ if (simpleLogFile.length() > LOG_FILE_MAX_SIZE) {
+ rollSimpleConsensusLogFile();
}
- startIndex = endIndex + 1;
- createLogFile(startIndex);
- }
- try {
ByteBuffer buffer = plan.serializeToByteBuffer();
buffer.position(buffer.limit());
simpleLogWriter.write(buffer);
+ simpleLogWriter.force();
endIndex = endIndex + 1;
} catch (Exception e) {
LOGGER.error(
- "Can't serialize current ConfigPhysicalPlan for ConfigNode
SimpleConsensus mode", e);
+ "Persist current ConfigPhysicalPlan for ConfigNode SimpleConsensus
mode failed", e);
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(
+ "Persist ConfigNode SimpleConsensus log failed: " +
String.valueOf(e.getMessage()));
Review Comment:
Addressed in the rewritten persistence error path: the message now directly
concatenates `e.getMessage()`.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java:
##########
@@ -1008,9 +1010,14 @@ public boolean processTakeSnapshot(File snapshotDir)
throws TException, IOExcept
databasePartitionTableEntry.getValue().serialize(bufferedOutputStream,
protocol);
}
+ final List<RegionMaintainTask> copiedRegionMaintainTaskList;
+ synchronized (regionMaintainTaskList) {
+ copiedRegionMaintainTaskList = new ArrayList<>(regionMaintainTaskList);
+ }
Review Comment:
Thanks for noting this. The snapshot path intentionally copies under the
lock to obtain a consistent list, then performs serialization after releasing
the lock. As noted, this list is expected to remain small for ConfigNode.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java:
##########
@@ -243,7 +243,7 @@ public int getRegionGroupCount(TConsensusGroupType type) {
result.getAndIncrement();
}
});
- return result.getAndIncrement();
+ return result.get();
Review Comment:
Yes, this was an off-by-one side effect: `getAndIncrement()` returned the
correct count but unnecessarily advanced the local counter before returning. It
now uses `get()`, and `PartitionInfoTest.testRegionGroupCount` covers both
schema and data region counts.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java:
##########
@@ -122,16 +122,32 @@ public TSStatus write(IConsensusRequest request) {
/** Transmit {@link ConfigPhysicalPlan} to {@link ConfigPlanExecutor} */
protected TSStatus write(ConfigPhysicalPlan plan) {
+ SimpleConsensusPersistResult persistResult = null;
+ if
(ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass()))
{
+ persistResult = persistPlanForSimpleConsensus(plan);
Review Comment:
Addressed. The failed-write test audits the persist-success/execute-failure
path and verifies rollback leaves no replayable WAL entry. I also added replay
idempotency coverage for the plan types directly touched by the registration
changes: RegisterDataNodePlan, ApplyConfigNodePlan, and UpdateVersionInfoPlan.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java:
##########
@@ -122,16 +122,32 @@ public TSStatus write(IConsensusRequest request) {
/** Transmit {@link ConfigPhysicalPlan} to {@link ConfigPlanExecutor} */
protected TSStatus write(ConfigPhysicalPlan plan) {
+ SimpleConsensusPersistResult persistResult = null;
+ if
(ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass()))
{
+ persistResult = persistPlanForSimpleConsensus(plan);
+ final TSStatus persistStatus = persistResult.status;
+ if (persistStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return persistStatus;
+ }
+ }
+
TSStatus result;
try {
result = executor.executeNonQueryPlan(plan);
Review Comment:
Addressed. Rollback failure logging now includes the plan type, WAL file,
truncate offset, and pre-write end index. The returned status also explicitly
says that the persisted plan may replay after restart, so both operators and
callers can see the possible WAL/in-memory divergence.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
