CRZbulabula commented on code in PR #17637:
URL: https://github.com/apache/iotdb/pull/17637#discussion_r3433808772
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/partition/DataPartitionTableIntegrityCheckProcedure.java:
##########
@@ -1114,4 +1117,86 @@ public void setSkipDataNodes(Set<TDataNodeConfiguration>
skipDataNodes) {
public void setFailedDataNodes(Set<TDataNodeConfiguration> failedDataNodes) {
this.failedDataNodes = failedDataNodes;
}
+
+ public TShowRepairDataPartitionTableProgressResp getProgress(final
ConfigNodeProcedureEnv env) {
+ final DataPartitionTableIntegrityCheckProcedureState currentState =
getCurrentState();
+ final String state = currentState == null ? "UNKNOWN" :
currentState.name();
+ final double progress =
+ currentState == null ? 0.0 : calculateProgressByState(env,
currentState) * 100;
+
+ return new TShowRepairDataPartitionTableProgressResp(
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS), state, progress)
+ .setMessage(String.format("DataPartitionTable integrity check
progress: %.1f%%", progress));
+ }
+
+ private double calculateProgressByState(
+ final ConfigNodeProcedureEnv env,
+ final DataPartitionTableIntegrityCheckProcedureState currentState) {
+ switch (currentState) {
+ case COLLECT_EARLIEST_TIMESLOTS:
+ return 0.0;
+ case ANALYZE_MISSING_PARTITIONS:
+ return 0.05;
+ case REQUEST_PARTITION_TABLES:
+ return 0.1;
+ case REQUEST_PARTITION_TABLES_HEART_BEAT:
+ return 0.1 + 0.8 * calculateDataNodeGeneratorProgress(env);
+ case MERGE_PARTITION_TABLES:
+ return 0.95;
+ case WRITE_PARTITION_TABLE_TO_CONSENSUS:
+ return 0.99;
+ default:
+ return 0.0;
+ }
+ }
+
+ private double calculateDataNodeGeneratorProgress(final
ConfigNodeProcedureEnv env) {
+ final LoadManager currentLoadManager =
+ loadManager == null ? env.getConfigManager().getLoadManager() :
loadManager;
+
+ final Set<TDataNodeConfiguration> targetDataNodes = new
HashSet<>(allDataNodes);
Review Comment:
**[Thread-safety] Unsynchronized cross-thread read of procedure fields.**
This method runs on the RPC handler thread, but `allDataNodes` is a
non-`volatile` `ArrayList` that the procedure-executor thread reassigns on
every step (`allDataNodes = dataNodeManager.getRegisteredDataNodes()`) and
clears in `rollbackState` (`allDataNodes.clear()`). Copying it via `new
HashSet<>(allDataNodes)` while another thread structurally modifies or
reassigns it can throw `ConcurrentModificationException` or observe a
half-published reference. The same concern applies to
`skipDataNodes`/`failedDataNodes`, which are reassigned with `= new
HashSet<>()` in `executeFromState`/`getInitialState`. A transient exception
here would fail the user's `SHOW ... PROGRESS` even though the repair itself is
healthy.
Please snapshot these fields under a lock, make them `volatile`, or wrap the
read so `getProgress` degrades to `UNKNOWN` instead of throwing.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/partition/DataPartitionTableIntegrityCheckProcedure.java:
##########
@@ -1114,4 +1117,86 @@ public void setSkipDataNodes(Set<TDataNodeConfiguration>
skipDataNodes) {
public void setFailedDataNodes(Set<TDataNodeConfiguration> failedDataNodes) {
this.failedDataNodes = failedDataNodes;
}
+
+ public TShowRepairDataPartitionTableProgressResp getProgress(final
ConfigNodeProcedureEnv env) {
+ final DataPartitionTableIntegrityCheckProcedureState currentState =
getCurrentState();
+ final String state = currentState == null ? "UNKNOWN" :
currentState.name();
+ final double progress =
+ currentState == null ? 0.0 : calculateProgressByState(env,
currentState) * 100;
+
+ return new TShowRepairDataPartitionTableProgressResp(
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS), state, progress)
+ .setMessage(String.format("DataPartitionTable integrity check
progress: %.1f%%", progress));
+ }
+
+ private double calculateProgressByState(
+ final ConfigNodeProcedureEnv env,
+ final DataPartitionTableIntegrityCheckProcedureState currentState) {
+ switch (currentState) {
+ case COLLECT_EARLIEST_TIMESLOTS:
+ return 0.0;
+ case ANALYZE_MISSING_PARTITIONS:
+ return 0.05;
+ case REQUEST_PARTITION_TABLES:
+ return 0.1;
+ case REQUEST_PARTITION_TABLES_HEART_BEAT:
+ return 0.1 + 0.8 * calculateDataNodeGeneratorProgress(env);
+ case MERGE_PARTITION_TABLES:
+ return 0.95;
+ case WRITE_PARTITION_TABLE_TO_CONSENSUS:
+ return 0.99;
+ default:
+ return 0.0;
+ }
+ }
+
+ private double calculateDataNodeGeneratorProgress(final
ConfigNodeProcedureEnv env) {
+ final LoadManager currentLoadManager =
+ loadManager == null ? env.getConfigManager().getLoadManager() :
loadManager;
+
+ final Set<TDataNodeConfiguration> targetDataNodes = new
HashSet<>(allDataNodes);
+ targetDataNodes.removeAll(skipDataNodes);
+ if (targetDataNodes.isEmpty()) {
+ return dataPartitionTables.isEmpty() ? 0.0 : 1.0;
+ }
+
+ double progressSum = 0.0;
+ for (TDataNodeConfiguration dataNode : targetDataNodes) {
+ final int dataNodeId = dataNode.getLocation().getDataNodeId();
+ if (dataPartitionTables.containsKey(dataNodeId)
+ || failedDataNodes.contains(dataNode)
+ ||
!NodeStatus.Running.equals(currentLoadManager.getNodeStatus(dataNodeId))) {
+ progressSum += 1.0;
+ continue;
+ }
+
+ try {
+ Object response =
+ SyncDataNodeClientPool.getInstance()
+ .sendSyncRequestToDataNodeWithGivenRetry(
Review Comment:
**[Performance] Synchronous fan-out RPC inside a status query.**
This issues a synchronous RPC with up to `MAX_RETRY_COUNT` (3) retries to
every target DataNode, sequentially, on the RPC handler thread. On a large
cluster with one slow/unreachable DataNode, a `SHOW ... PROGRESS` query can
block for roughly `N × retries × timeout`.
The `REQUEST_PARTITION_TABLES_HEART_BEAT` state already polls per-DataNode
generation progress — consider caching the latest per-DataNode progress on the
procedure and reading it here, which avoids the extra fan-out entirely and is
both faster and race-free. If the synchronous fan-out is kept, please bound or
parallelize it.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/partition/DataPartitionTableIntegrityCheckProcedure.java:
##########
@@ -1114,4 +1117,86 @@ public void setSkipDataNodes(Set<TDataNodeConfiguration>
skipDataNodes) {
public void setFailedDataNodes(Set<TDataNodeConfiguration> failedDataNodes) {
this.failedDataNodes = failedDataNodes;
}
+
+ public TShowRepairDataPartitionTableProgressResp getProgress(final
ConfigNodeProcedureEnv env) {
+ final DataPartitionTableIntegrityCheckProcedureState currentState =
getCurrentState();
+ final String state = currentState == null ? "UNKNOWN" :
currentState.name();
+ final double progress =
+ currentState == null ? 0.0 : calculateProgressByState(env,
currentState) * 100;
+
+ return new TShowRepairDataPartitionTableProgressResp(
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS), state, progress)
+ .setMessage(String.format("DataPartitionTable integrity check
progress: %.1f%%", progress));
+ }
+
+ private double calculateProgressByState(
+ final ConfigNodeProcedureEnv env,
+ final DataPartitionTableIntegrityCheckProcedureState currentState) {
+ switch (currentState) {
+ case COLLECT_EARLIEST_TIMESLOTS:
+ return 0.0;
+ case ANALYZE_MISSING_PARTITIONS:
+ return 0.05;
+ case REQUEST_PARTITION_TABLES:
+ return 0.1;
+ case REQUEST_PARTITION_TABLES_HEART_BEAT:
+ return 0.1 + 0.8 * calculateDataNodeGeneratorProgress(env);
+ case MERGE_PARTITION_TABLES:
+ return 0.95;
+ case WRITE_PARTITION_TABLE_TO_CONSENSUS:
+ return 0.99;
+ default:
Review Comment:
**[Maintainability] Silent `default` branch.**
The `default` branch maps any unhandled state to `0.0`. If a new
`DataPartitionTableIntegrityCheckProcedureState` is added later, progress will
silently report 0% for it. Please add a `LOG.warn` here to catch enum drift.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java:
##########
@@ -3522,6 +3523,52 @@ public TGenerateDataPartitionTableHeartbeatResp
generateDataPartitionTableHeartb
return resp;
}
+ @Override
+ public TGetDataPartitionTableGeneratorProgressResp
getDataPartitionTableGeneratorProgress() {
+ TGetDataPartitionTableGeneratorProgressResp resp =
+ new TGetDataPartitionTableGeneratorProgressResp();
+
+ if (currentGenerator == null) {
+ resp.setErrorCode(DataPartitionTableGeneratorState.UNKNOWN.getCode());
+ resp.setProgress(0.0);
+ resp.setMessage("No DataPartitionTable generation task found");
+ resp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
+ return resp;
+ }
+
+ switch (currentGenerator.getStatus()) {
Review Comment:
**[Possible NPE] Re-reading `volatile currentGenerator` without a local
copy.**
`currentGenerator` is `volatile`, but this method reads it multiple times
(`getStatus()`, `getProgress()`, `getErrorMessage()`) after the null-check
above. The generation-with-heartbeat path sets `currentGenerator = null` on
completion, so a concurrent null-assignment between the null-check and this
`switch` would NPE. Please capture `final DataPartitionTableGenerator generator
= currentGenerator;` once at the top and operate on the local.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java:
##########
@@ -543,6 +544,18 @@ public void
markDataPartitionTableIntegrityCheckProcedureFinished() {
dataPartitionTableIntegrityCheckProcedureRunning.set(false);
}
+ public TShowRepairDataPartitionTableProgressResp
showRepairDataPartitionTableProgress() {
+ return configManager
+ .getProcedureManager()
+ .getUnfinishedDataPartitionTableIntegrityCheckProcedure()
+ .map(procedure ->
procedure.getProgress(configManager.getProcedureManager().getEnv()))
+ .orElseGet(
+ () ->
+ new TShowRepairDataPartitionTableProgressResp(
+ RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS),
"IDLE", 100.0)
Review Comment:
**[Semantics] `IDLE` returns `progress = 100.0`.**
When no procedure is running, this returns `IDLE` with `progress = 100.0`.
"Idle at 100%" is ambiguous — a user who has never triggered a repair will see
100% progress. Suggest returning `0.0` (or a distinct sentinel) for the idle
case, or at minimum a comment clarifying the semantics.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java:
##########
@@ -1180,6 +1181,17 @@ public TSStatus dataPartitionTableIntegrityCheck() {
return partitionManager.dataPartitionTableIntegrityCheck();
}
+ @Override
+ public TShowRepairDataPartitionTableProgressResp
showRepairDataPartitionTableProgress() {
+ TSStatus status = confirmLeader();
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return new TShowRepairDataPartitionTableProgressResp(status, "UNKNOWN",
0.0)
Review Comment:
**[Maintainability] Magic state strings duplicated across modules.**
The `state` string is produced ad hoc in three different places: `"UNKNOWN"`
here, `"IDLE"` in `PartitionManager.showRepairDataPartitionTableProgress`, and
the raw enum `name()` in
`DataPartitionTableIntegrityCheckProcedure.getProgress`. Please define these as
shared constants or a small enum so the `state` contract is single-sourced and
testable, rather than relying on matching string literals across modules.
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java:
##########
@@ -1055,6 +1055,14 @@ public void testCreateView() throws IllegalPathException
{
assertEquals(null, stmt.getQueryStatement());
}
+ @Test
+ public void testShowRepairDataPartitionTableProgress() {
Review Comment:
**[Test coverage] Add an integration test for progress reporting.**
This parser-level test is good, but it's the only coverage. Please add an
integration test that exercises the feature end-to-end, covering: (a) `SHOW
REPAIR DATA PARTITION TABLE PROGRESS` while a `REPAIR DATA PARTITION TABLE`
procedure is running (asserting a 0–100 `Progress(%)` and a non-terminal
`Status`); (b) the idle case when no procedure exists; and (c) the non-leader /
`UNKNOWN` branch. This would also guard the concurrency and DataNode-RPC
fan-out paths flagged in the other comments. Per the repo conventions, scope
the new IT to only this feature (`-Dit.test=ClassName`) and place it with the
existing repair-partition ITs.
--
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]
