Skip to content

Commit 532d6c1

Browse files
CaideyipiCaideyipi
authored
Fence IoTConsensus applies before storage shutdown (#18841)
* Fix queued IoTConsensus writes during shutdown * Test queued IoTConsensus writes across storage shutdown * Reproduce WAL checkpoint loss from writes after shutdown * Fence IoTConsensus applies before storage shutdown
1 parent 264053f commit 532d6c1

10 files changed

Lines changed: 802 additions & 34 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java‎

Lines changed: 25 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -348,6 +348,9 @@ public void testPipeAfterDataRegionLeaderStop() {
348348

349349
try (final SyncConfigNodeIServiceClient client =
350350
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
351+
// Distinguish a source-side write loss during the restart from a p2 transfer failure.
352+
TableModelUtils.assertData("test1", "test1", 0, 300, senderEnv);
353+
351354
// Create a new pipe and write new data
352355
final Map<String, String> sourceAttributes = new HashMap<>();
353356
final Map<String, String> processorAttributes = new HashMap<>();
@@ -449,31 +452,32 @@ private void flushTableDataRegionReplicasAfterReplicationComplete(
449452

450453
private void waitForTableDataRegionReplicationComplete(
451454
final BaseEnv env, final List<String> databases) {
455+
// IoTConsensus accepts writes on any replica, including a previous leader during failover.
456+
// Check every replica's outgoing replication before flushing or restarting the cluster.
452457
await()
453458
.pollInterval(500, TimeUnit.MILLISECONDS)
454459
.atMost(2, TimeUnit.MINUTES)
455460
.untilAsserted(
456461
() -> {
457462
try (final SyncConfigNodeIServiceClient client =
458463
(SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) {
459-
final List<TRegionInfo> leaderRegionInfoList =
460-
showTableDataRegionLeaders(databases, client);
464+
final List<TRegionInfo> regionInfoList = showTableDataRegions(databases, client);
461465
Assert.assertFalse(
462-
"No table DataRegion leader found for databases " + databases,
463-
leaderRegionInfoList.isEmpty());
466+
"No table DataRegion found for databases " + databases,
467+
regionInfoList.isEmpty());
464468

465-
for (final TRegionInfo regionInfo : leaderRegionInfoList) {
466-
final DataNodeWrapper leaderNode =
469+
for (final TRegionInfo regionInfo : regionInfoList) {
470+
final DataNodeWrapper replicaNode =
467471
findDataNodeWrapperByPort(env, regionInfo.getClientRpcPort());
468472
final String metricsUrl =
469473
"http://"
470-
+ leaderNode.getIp()
474+
+ replicaNode.getIp()
471475
+ ":"
472-
+ leaderNode.getMetricPort()
476+
+ replicaNode.getMetricPort()
473477
+ "/metrics";
474478
final String metricsContent = env.getUrlContent(metricsUrl, null);
475479
Assert.assertNotNull(
476-
"Failed to fetch metrics from leader DataNode at " + metricsUrl,
480+
"Failed to fetch metrics from replica DataNode at " + metricsUrl,
477481
metricsContent);
478482
assertSyncLagIsZero(metricsContent, buildDataRegionTag(regionInfo), metricsUrl);
479483
}
@@ -483,6 +487,17 @@ private void waitForTableDataRegionReplicationComplete(
483487

484488
private List<TRegionInfo> showTableDataRegionLeaders(
485489
final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException {
490+
final List<TRegionInfo> result = new ArrayList<>();
491+
for (final TRegionInfo regionInfo : showTableDataRegions(databases, client)) {
492+
if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
493+
result.add(regionInfo);
494+
}
495+
}
496+
return result;
497+
}
498+
499+
private List<TRegionInfo> showTableDataRegions(
500+
final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException {
486501
final TShowRegionResp showRegionResp =
487502
client.showRegion(
488503
new TShowRegionReq()
@@ -491,13 +506,7 @@ private List<TRegionInfo> showTableDataRegionLeaders(
491506
.setIsTableModel(true));
492507
Assert.assertEquals(
493508
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
494-
final List<TRegionInfo> result = new ArrayList<>();
495-
for (final TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
496-
if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
497-
result.add(regionInfo);
498-
}
499-
}
500-
return result;
509+
return showRegionResp.getRegionInfoList();
501510
}
502511

503512
private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final int port) {

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java‎

Lines changed: 22 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -458,31 +458,32 @@ private void flushTreeDataRegionReplicasAfterReplicationComplete(
458458

459459
private void waitForTreeDataRegionReplicationComplete(
460460
final BaseEnv env, final List<String> databases) {
461+
// IoTConsensus accepts writes on any replica, including a previous leader during failover.
462+
// Check every replica's outgoing replication before flushing or restarting the cluster.
461463
await()
462464
.pollInterval(500, TimeUnit.MILLISECONDS)
463465
.atMost(2, TimeUnit.MINUTES)
464466
.untilAsserted(
465467
() -> {
466468
try (final SyncConfigNodeIServiceClient client =
467469
(SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) {
468-
final List<TRegionInfo> leaderRegionInfoList =
469-
showTreeDataRegionLeaders(databases, client);
470+
final List<TRegionInfo> regionInfoList = showTreeDataRegions(databases, client);
470471
Assert.assertFalse(
471-
"No tree DataRegion leader found for databases " + databases,
472-
leaderRegionInfoList.isEmpty());
472+
"No tree DataRegion found for databases " + databases,
473+
regionInfoList.isEmpty());
473474

474-
for (final TRegionInfo regionInfo : leaderRegionInfoList) {
475-
final DataNodeWrapper leaderNode =
475+
for (final TRegionInfo regionInfo : regionInfoList) {
476+
final DataNodeWrapper replicaNode =
476477
findDataNodeWrapperByPort(env, regionInfo.getClientRpcPort());
477478
final String metricsUrl =
478479
"http://"
479-
+ leaderNode.getIp()
480+
+ replicaNode.getIp()
480481
+ ":"
481-
+ leaderNode.getMetricPort()
482+
+ replicaNode.getMetricPort()
482483
+ "/metrics";
483484
final String metricsContent = env.getUrlContent(metricsUrl, null);
484485
Assert.assertNotNull(
485-
"Failed to fetch metrics from leader DataNode at " + metricsUrl,
486+
"Failed to fetch metrics from replica DataNode at " + metricsUrl,
486487
metricsContent);
487488
assertSyncLagIsZero(metricsContent, buildDataRegionTag(regionInfo), metricsUrl);
488489
}
@@ -492,20 +493,25 @@ private void waitForTreeDataRegionReplicationComplete(
492493

493494
private List<TRegionInfo> showTreeDataRegionLeaders(
494495
final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException {
496+
final List<TRegionInfo> result = new ArrayList<>();
497+
for (final TRegionInfo regionInfo : showTreeDataRegions(databases, client)) {
498+
if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
499+
result.add(regionInfo);
500+
}
501+
}
502+
return result;
503+
}
504+
505+
private List<TRegionInfo> showTreeDataRegions(
506+
final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException {
495507
final TShowRegionResp showRegionResp =
496508
client.showRegion(
497509
new TShowRegionReq()
498510
.setConsensusGroupType(TConsensusGroupType.DataRegion)
499511
.setDatabases(databases));
500512
Assert.assertEquals(
501513
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
502-
final List<TRegionInfo> result = new ArrayList<>();
503-
for (final TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
504-
if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
505-
result.add(regionInfo);
506-
}
507-
}
508-
return result;
514+
return showRegionResp.getRegionInfoList();
509515
}
510516

511517
private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final int port) {

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1617,7 +1617,14 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque
16171617
for (int i = 0; i < insertNodes.size(); i++) {
16181618
IConsensusRequest insertNode = insertNodes.get(i);
16191619
insertNode.markAsGeneratedByRemoteConsensusLeader();
1620-
TSStatus status = stateMachine.write(insertNode);
1620+
// Reject requests that became read-only while queued. This is an early rejection only:
1621+
// the state machine must also fence the full apply against storage shutdown, since the
1622+
// replica can become read-only after this check.
1623+
TSStatus status =
1624+
isReadOnly()
1625+
? RpcUtils.getStatus(
1626+
TSStatusCode.SYSTEM_READ_ONLY, ConsensusMessages.SYNC_LOG_SYSTEM_READ_ONLY)
1627+
: stateMachine.write(insertNode);
16211628
subStatus.add(status);
16221629
if (RetryUtils.needRetryForWrite(status.getCode())) {
16231630
for (int j = i + 1; j < insertNodes.size(); j++) {

‎iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,21 @@
2020
package org.apache.iotdb.consensus.iot;
2121

2222
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
23+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2324
import org.apache.iotdb.commons.audit.UserDataTransferAuditHandler;
2425
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
2526
import org.apache.iotdb.commons.consensus.DataRegionId;
2627
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
2728
import org.apache.iotdb.commons.request.IConsensusRequest;
2829
import org.apache.iotdb.consensus.common.Peer;
30+
import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest;
31+
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
2932
import org.apache.iotdb.consensus.config.IoTConsensusConfig;
3033
import org.apache.iotdb.consensus.config.UserDataTransferAuditClassifier;
34+
import org.apache.iotdb.consensus.i18n.ConsensusMessages;
35+
import org.apache.iotdb.consensus.iot.util.TestEntry;
3136
import org.apache.iotdb.consensus.iot.util.TestStateMachine;
37+
import org.apache.iotdb.rpc.TSStatusCode;
3238

3339
import org.junit.Rule;
3440
import org.junit.Test;
@@ -46,6 +52,8 @@
4652
import java.util.concurrent.Future;
4753
import java.util.concurrent.ScheduledExecutorService;
4854
import java.util.concurrent.TimeUnit;
55+
import java.util.concurrent.atomic.AtomicBoolean;
56+
import java.util.concurrent.atomic.AtomicReference;
4957

5058
import static org.junit.Assert.assertEquals;
5159
import static org.junit.Assert.assertFalse;
@@ -59,6 +67,92 @@ public class IoTConsensusServerImplTest {
5967

6068
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
6169

70+
@Test
71+
public void testSyncLogBecomesReadOnlyWhileWaiting() throws Exception {
72+
final ScheduledExecutorService backgroundTaskService =
73+
Executors.newSingleThreadScheduledExecutor();
74+
final AtomicBoolean readOnly = new AtomicBoolean();
75+
final TestStateMachine stateMachine =
76+
new TestStateMachine() {
77+
@Override
78+
public boolean isReadOnly() {
79+
return readOnly.get();
80+
}
81+
};
82+
final Peer thisNode = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667));
83+
final IoTConsensusServerImpl server =
84+
new IoTConsensusServerImpl(
85+
temporaryFolder.getRoot().getAbsolutePath(),
86+
null,
87+
DirectoryStrategyType.SEQUENCE_STRATEGY,
88+
thisNode,
89+
Collections.singletonList(thisNode),
90+
stateMachine,
91+
backgroundTaskService,
92+
null,
93+
null,
94+
IoTConsensusConfig.newBuilder()
95+
.setReplication(
96+
IoTConsensusConfig.Replication.newBuilder()
97+
.setMaxWaitingTimeForWaitBatchInMs(100)
98+
.build())
99+
.build());
100+
final DeserializedBatchIndexedConsensusRequest request =
101+
new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2);
102+
for (int index = 1; index <= 2; index++) {
103+
request.add(
104+
new IndexedConsensusRequest(
105+
index, index, Collections.singletonList(new TestEntry(index, thisNode))));
106+
}
107+
final AtomicReference<TSStatus> status = new AtomicReference<>();
108+
final Thread syncLogThread = new Thread(() -> status.set(server.syncLog(1, request)));
109+
try {
110+
assertFalse(server.isReadOnly());
111+
syncLogThread.start();
112+
final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
113+
while (syncLogThread.getState() != Thread.State.TIMED_WAITING
114+
&& System.nanoTime() < deadline) {
115+
Thread.sleep(10);
116+
}
117+
assertEquals(Thread.State.TIMED_WAITING, syncLogThread.getState());
118+
readOnly.set(true);
119+
syncLogThread.join(TimeUnit.SECONDS.toMillis(15));
120+
assertFalse(syncLogThread.isAlive());
121+
assertEquals(2, status.get().getSubStatusSize());
122+
assertEquals(
123+
TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(),
124+
status.get().getSubStatus().get(0).getCode());
125+
assertEquals(
126+
ConsensusMessages.SYNC_LOG_SYSTEM_READ_ONLY,
127+
status.get().getSubStatus().get(0).getMessage());
128+
assertEquals(
129+
TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(),
130+
status.get().getSubStatus().get(1).getCode());
131+
assertTrue(stateMachine.getRequestSet().isEmpty());
132+
assertEquals(0, server.getEffectiveSafePhysicalTime(1));
133+
134+
// The sender must be able to retry the complete batch after the replica becomes writable.
135+
readOnly.set(false);
136+
server.reloadConsensusConfig(
137+
IoTConsensusConfig.newBuilder()
138+
.setReplication(
139+
IoTConsensusConfig.Replication.newBuilder()
140+
.setMaxPendingBatchesNum(1)
141+
.setMaxWaitingTimeForWaitBatchInMs(100)
142+
.build())
143+
.build());
144+
final TSStatus retryStatus = server.syncLog(1, request);
145+
assertTrue(
146+
retryStatus.getSubStatus().stream()
147+
.allMatch(s -> s.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()));
148+
assertEquals(2, stateMachine.getRequestSet().size());
149+
} finally {
150+
syncLogThread.interrupt();
151+
syncLogThread.join(TimeUnit.SECONDS.toMillis(5));
152+
backgroundTaskService.shutdownNow();
153+
}
154+
}
155+
62156
@Test
63157
public void testSnapshotTransferAppliesGroupClassifier() {
64158
final DataRegionId groupId = new DataRegionId(1);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -251,7 +251,7 @@ protected TSStatus write(PlanNode planNode) {
251251
TSStatus result = null;
252252
int retryTime = 0;
253253
while (retryTime < MAX_WRITE_RETRY_TIMES) {
254-
result = planNode.accept(new DataExecutionVisitor(), region);
254+
result = executeWrite(planNode);
255255
// Let pipe retry with the original event instead of retrying a possibly mutated node here.
256256
if (needRetryForSpecificCases(result.getCode(), planNode)) {
257257
retryTime++;
@@ -284,6 +284,10 @@ protected TSStatus write(PlanNode planNode) {
284284
return result;
285285
}
286286

287+
protected TSStatus executeWrite(PlanNode planNode) {
288+
return planNode.accept(new DataExecutionVisitor(), region);
289+
}
290+
287291
protected void waitBeforeNextWriteRetry() throws InterruptedException {
288292
Thread.sleep(WRITE_RETRY_WAIT_TIME_IN_MS);
289293
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,13 @@ public IoTConsensusDataRegionStateMachine(DataRegion region) {
5050
super(region);
5151
}
5252

53+
@Override
54+
protected TSStatus executeWrite(PlanNode planNode) {
55+
final DataRegion targetRegion = region;
56+
return targetRegion.executeIoTConsensusWrite(
57+
() -> planNode.accept(new DataExecutionVisitor(), targetRegion));
58+
}
59+
5360
@Override
5461
public TSStatus write(IConsensusRequest request) {
5562
try {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,9 @@ public void run() {
120120
CommonDescriptor.getInstance()
121121
.getConfig()
122122
.setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.STOPPING);
123+
// A pre-apply read-only check can race with this transition. Drain admitted IoTConsensus
124+
// applies, including their WAL separators, and permanently reject new ones before flushing.
125+
StorageEngine.getInstance().stopIoTConsensusWrites();
123126
// Wait all wal are flushed
124127
WALManager.getInstance().waitAllWALFlushed();
125128

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -435,6 +435,7 @@ private void asyncRecoverTsFileResource() {
435435

436436
@Override
437437
public void stop() {
438+
stopIoTConsensusWrites();
438439
loadTsFileManager.stop();
439440
for (DataRegion dataRegion : dataRegionMap.values()) {
440441
if (dataRegion != null) {
@@ -455,6 +456,7 @@ public void stop() {
455456

456457
@Override
457458
public void shutdown(long milliseconds) throws ShutdownException {
459+
stopIoTConsensusWrites();
458460
loadTsFileManager.stop();
459461
try {
460462
for (DataRegion dataRegion : dataRegionMap.values()) {
@@ -526,6 +528,17 @@ public synchronized void reset() {
526528
dataRegionMap.clear();
527529
}
528530

531+
public void stopIoTConsensusWrites() {
532+
if (!CONFIG.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) {
533+
return;
534+
}
535+
for (DataRegion dataRegion : dataRegionMap.values()) {
536+
if (dataRegion != null) {
537+
dataRegion.stopIoTConsensusWrites();
538+
}
539+
}
540+
}
541+
529542
/** flush command Sync asyncCloseOneProcessor all file node processors. */
530543
public void syncCloseAllProcessor() {
531544
LOGGER.info(StorageEngineMessages.START_CLOSING_ALL_DB_PROCESSOR);

0 commit comments

Comments
 (0)