You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Pengzna (via GitHub)" <gi...@apache.org> on 2026/03/21 06:03:58 UTC
[PR] Add IT for DELETE TIMESERIES replica consistency under IoTConsensusV2 [iotdb]
Pengzna opened a new pull request, #17332:
URL: https://github.com/apache/iotdb/pull/17332
## Summary
- Add `testDeleteTimeSeriesReplicaConsistency()` integration test to verify that `DELETE TIMESERIES` operations are properly replicated across all DataNode replicas in a 3C3D IoTConsensusV2 cluster
- The test reproduces the scenario from the historical deletion replication bug (fixed in #17329): deletion events missing `replicateIndex` were silently dropped by `IoTConsensusV2Processor`, causing schema inconsistency across replicas
- Unify `INSERTION` constants across all test methods to use 3 columns (`speed`, `temperature`, `power`), replacing the prior 2-column variants
## Test Scenario
1. Insert data with 3 measurements (`speed`, `temperature`, `power`) and flush
2. Insert more data **without flush** (WAL-only entries, simulating in-flight data)
3. `DELETE TIMESERIES root.sg.d1.speed`
4. Flush to persist the deletion
5. Wait for replication to complete (syncLag == 0) on all data region leaders
6. Verify schema consistency on each DataNode independently via `SHOW TIMESERIES`
7. Stop each DataNode one by one (triggers consensus pipe reconstruction + historical replay), verify all surviving nodes still show consistent schema — the deleted timeseries must be absent everywhere
## Test Plan
- [x] `testDeleteTimeSeriesReplicaConsistency` added in both **stream** and **batch** mode subclasses
- [ ] Run `IoTDBIoTConsensusV2Stream3C3DBasicIT.testDeleteTimeSeriesReplicaConsistency`
- [ ] Run `IoTDBIoTConsensusV2Batch3C3DBasicIT.testDeleteTimeSeriesReplicaConsistency`
- [ ] Verify existing tests `test3C3DWriteFlushAndQuery` and `testReplicaConsistencyAfterLeaderStop` still pass with unified 3-column insertions
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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
Re: [PR] Add IT for DELETE TIMESERIES replica consistency under IoTConsensusV2 [iotdb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17332:
URL: https://github.com/apache/iotdb/pull/17332#discussion_r2969190222
##########
integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java:
##########
@@ -210,6 +217,187 @@ public void testReplicaConsistencyAfterLeaderStop() throws Exception {
}
}
+ /**
+ * Test that DELETE TIMESERIES is properly replicated to all DataNode replicas via IoTConsensusV2.
+ *
+ * <p>This test reproduces the scenario from the historical deletion replication bug: when a
+ * timeseries is deleted after data insertion (with some unflushed data), the deletion event must
+ * be consistently replicated to all replicas. After waiting for replication to complete, stopping
+ * each DataNode in turn should show the same schema on all surviving nodes.
+ *
+ * <p>Scenario:
+ *
+ * <ol>
+ * <li>Insert data into root.sg.d1 with 3 measurements (speed, temperature, power), flush
+ * <li>Insert more data (unflushed to create WAL-only entries)
+ * <li>DELETE TIMESERIES root.sg.d1.speed
+ * <li>Flush again to persist deletion
+ * <li>Wait for replication to complete on all DataNodes
+ * <li>Verify that every DataNode independently shows the same timeseries (speed is gone)
+ * </ol>
+ */
+ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
+ try (Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection());
+ Statement statement = makeItCloseQuietly(connection.createStatement())) {
+
+ // Step 1: Insert data with 3 measurements and flush
+ LOGGER.info(
+ "Step 1: Inserting data with 3 measurements and flushing (mode: {})...",
+ getIoTConsensusV2Mode());
+ statement.execute(INSERTION1);
+ statement.execute(INSERTION2);
+ statement.execute(FLUSH_COMMAND);
+
+ // Step 2: Insert more data without flush (creates WAL-only entries)
+ LOGGER.info("Step 2: Inserting more data without flush (WAL-only entries)...");
+ statement.execute(INSERTION3);
+
+ // Step 3: Delete one timeseries
+ LOGGER.info("Step 3: Deleting timeseries root.sg.d1.speed...");
+ statement.execute(DELETE_TIMESERIES_SPEED);
+
+ // Step 4: Flush again to persist the deletion
+ LOGGER.info("Step 4: Flushing to persist deletion...");
+ statement.execute(FLUSH_COMMAND);
+
+ // Verify on the current connection: speed should be gone, 2 timeseries remain
+ verifyTimeSeriesAfterDelete(statement, "via initial connection");
+
+ // Step 5: Wait for replication to complete on data region leaders
+ LOGGER.info("Step 5: Waiting for replication to complete on data region leaders...");
+ Map<Integer, Pair<Integer, Set<Integer>>> dataRegionMap =
+ getDataRegionMapWithLeader(statement);
+ Set<Integer> leaderNodeIds = new HashSet<>();
+ for (Pair<Integer, Set<Integer>> leaderAndReplicas : dataRegionMap.values()) {
+ if (leaderAndReplicas.getLeft() > 0) {
+ leaderNodeIds.add(leaderAndReplicas.getLeft());
+ }
+ }
+ for (int leaderNodeId : leaderNodeIds) {
+ EnvFactory.getEnv()
+ .dataNodeIdToWrapper(leaderNodeId)
+ .ifPresent(this::waitForReplicationComplete);
+ }
+
+ // Step 6: Verify schema consistency on each DataNode independently
+ LOGGER.info("Step 6: Verifying schema consistency on each DataNode independently...");
+ List<DataNodeWrapper> dataNodeWrappers = EnvFactory.getEnv().getDataNodeWrapperList();
+ for (DataNodeWrapper wrapper : dataNodeWrappers) {
+ String nodeDescription = "DataNode " + wrapper.getIp() + ":" + wrapper.getPort();
+ LOGGER.info("Verifying schema on {}", nodeDescription);
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ try (Connection nodeConn =
+ makeItCloseQuietly(
+ EnvFactory.getEnv()
+ .getConnection(
+ wrapper,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TREE_SQL_DIALECT));
+ Statement nodeStmt = makeItCloseQuietly(nodeConn.createStatement())) {
+ verifyTimeSeriesAfterDelete(nodeStmt, nodeDescription);
+ }
+ });
+ }
+
+ // Step 7: Stop each DataNode one by one and verify remaining nodes still consistent
+ LOGGER.info(
+ "Step 7: Stopping each DataNode in turn and verifying remaining nodes show consistent schema...");
+ for (DataNodeWrapper stoppedNode : dataNodeWrappers) {
+ String stoppedDesc = "DataNode " + stoppedNode.getIp() + ":" + stoppedNode.getPort();
+ LOGGER.info("Stopping {}", stoppedDesc);
+ stoppedNode.stopForcibly();
+ Assert.assertFalse(stoppedDesc + " should be stopped", stoppedNode.isAlive());
Review Comment:
After calling stopForcibly(), the test immediately asserts !isAlive(). AbstractNodeWrapper.stopForcibly() only waits up to 10s and ignores the return value, so the process may still be alive and this assertion can be flaky. Prefer awaiting the node to actually stop (e.g., Awaitility.until(() -> !stoppedNode.isAlive())) before proceeding.
```suggestion
Awaitility.await()
.atMost(60, TimeUnit.SECONDS)
.untilAsserted(
() ->
Assert.assertFalse(
stoppedDesc + " should be stopped", stoppedNode.isAlive()));
```
##########
integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java:
##########
@@ -210,6 +217,187 @@ public void testReplicaConsistencyAfterLeaderStop() throws Exception {
}
}
+ /**
+ * Test that DELETE TIMESERIES is properly replicated to all DataNode replicas via IoTConsensusV2.
+ *
+ * <p>This test reproduces the scenario from the historical deletion replication bug: when a
+ * timeseries is deleted after data insertion (with some unflushed data), the deletion event must
+ * be consistently replicated to all replicas. After waiting for replication to complete, stopping
+ * each DataNode in turn should show the same schema on all surviving nodes.
+ *
+ * <p>Scenario:
+ *
+ * <ol>
+ * <li>Insert data into root.sg.d1 with 3 measurements (speed, temperature, power), flush
+ * <li>Insert more data (unflushed to create WAL-only entries)
+ * <li>DELETE TIMESERIES root.sg.d1.speed
+ * <li>Flush again to persist deletion
+ * <li>Wait for replication to complete on all DataNodes
+ * <li>Verify that every DataNode independently shows the same timeseries (speed is gone)
+ * </ol>
+ */
+ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
+ try (Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection());
+ Statement statement = makeItCloseQuietly(connection.createStatement())) {
+
+ // Step 1: Insert data with 3 measurements and flush
+ LOGGER.info(
+ "Step 1: Inserting data with 3 measurements and flushing (mode: {})...",
+ getIoTConsensusV2Mode());
+ statement.execute(INSERTION1);
+ statement.execute(INSERTION2);
+ statement.execute(FLUSH_COMMAND);
+
+ // Step 2: Insert more data without flush (creates WAL-only entries)
+ LOGGER.info("Step 2: Inserting more data without flush (WAL-only entries)...");
+ statement.execute(INSERTION3);
+
+ // Step 3: Delete one timeseries
+ LOGGER.info("Step 3: Deleting timeseries root.sg.d1.speed...");
+ statement.execute(DELETE_TIMESERIES_SPEED);
+
+ // Step 4: Flush again to persist the deletion
+ LOGGER.info("Step 4: Flushing to persist deletion...");
+ statement.execute(FLUSH_COMMAND);
+
+ // Verify on the current connection: speed should be gone, 2 timeseries remain
+ verifyTimeSeriesAfterDelete(statement, "via initial connection");
+
+ // Step 5: Wait for replication to complete on data region leaders
+ LOGGER.info("Step 5: Waiting for replication to complete on data region leaders...");
+ Map<Integer, Pair<Integer, Set<Integer>>> dataRegionMap =
+ getDataRegionMapWithLeader(statement);
+ Set<Integer> leaderNodeIds = new HashSet<>();
+ for (Pair<Integer, Set<Integer>> leaderAndReplicas : dataRegionMap.values()) {
+ if (leaderAndReplicas.getLeft() > 0) {
+ leaderNodeIds.add(leaderAndReplicas.getLeft());
+ }
+ }
+ for (int leaderNodeId : leaderNodeIds) {
+ EnvFactory.getEnv()
+ .dataNodeIdToWrapper(leaderNodeId)
+ .ifPresent(this::waitForReplicationComplete);
+ }
+
+ // Step 6: Verify schema consistency on each DataNode independently
+ LOGGER.info("Step 6: Verifying schema consistency on each DataNode independently...");
+ List<DataNodeWrapper> dataNodeWrappers = EnvFactory.getEnv().getDataNodeWrapperList();
+ for (DataNodeWrapper wrapper : dataNodeWrappers) {
+ String nodeDescription = "DataNode " + wrapper.getIp() + ":" + wrapper.getPort();
+ LOGGER.info("Verifying schema on {}", nodeDescription);
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ try (Connection nodeConn =
+ makeItCloseQuietly(
+ EnvFactory.getEnv()
+ .getConnection(
+ wrapper,
+ SessionConfig.DEFAULT_USER,
+ SessionConfig.DEFAULT_PASSWORD,
+ BaseEnv.TREE_SQL_DIALECT));
+ Statement nodeStmt = makeItCloseQuietly(nodeConn.createStatement())) {
+ verifyTimeSeriesAfterDelete(nodeStmt, nodeDescription);
+ }
+ });
+ }
+
+ // Step 7: Stop each DataNode one by one and verify remaining nodes still consistent
+ LOGGER.info(
+ "Step 7: Stopping each DataNode in turn and verifying remaining nodes show consistent schema...");
Review Comment:
This log message line is likely to exceed the project's 100-character Checkstyle limit once indentation is included. Please wrap/split the string (or use multiple LOGGER.info calls) to keep each line within the limit.
```suggestion
"Step 7: Stopping each DataNode in turn and verifying remaining nodes "
+ "show consistent schema...");
```
--
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
Re: [PR] Add IT for DELETE TIMESERIES replica consistency under IoTConsensusV2 [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #17332:
URL: https://github.com/apache/iotdb/pull/17332
--
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