Skip to content

Commit 2fd1fe3

Browse files
alpass163gmailalpass163gmail
authored
support the retry for the execution the insert path in statemachine with DN metadata fenced and retry for coordinator (#18301)
1 parent 892dbd8 commit 2fd1fe3

17 files changed

Lines changed: 349 additions & 44 deletions

File tree

‎iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -90,7 +90,9 @@ public enum TSStatusCode {
9090
TYPE_NOT_FOUND(528),
9191
DATABASE_CONFLICT(529),
9292
DATABASE_MODEL(530),
93-
METADATA_LEASE_FENCED(531),
93+
// the range [531, 534] has been occupied
94+
METADATA_LEASE_FENCED(535),
95+
METADATA_LEASE_FENCED_RETRY_REQUIRED(536),
9496

9597
TABLE_NOT_EXISTS(550),
9698
TABLE_ALREADY_EXISTS(551),

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@
5757
import java.nio.file.StandardCopyOption;
5858
import java.util.Collection;
5959
import java.util.concurrent.CompletableFuture;
60+
import java.util.concurrent.TimeUnit;
6061
import java.util.function.BiConsumer;
6162

6263
public class ApplicationStateMachineProxy extends BaseStateMachine {
@@ -152,6 +153,10 @@ public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
152153
deserializedRequest.markAsGeneratedByRemoteConsensusLeader();
153154
}
154155
final TSStatus result = applicationStateMachine.write(deserializedRequest);
156+
if (result.getCode() == TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()
157+
&& waitBeforeRetry()) {
158+
continue;
159+
}
155160
ret = new ResponseMessage(result);
156161
break;
157162
} catch (Throwable rte) {
@@ -160,13 +165,11 @@ public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
160165
new ResponseMessage(
161166
new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())
162167
.setMessage(RatisMessages.INTERNAL_ERROR_STATEMACHINE_RUNTIME_EXCEPTION + rte));
163-
if (Utils.stallApply(consensusGroupType)) {
164-
waitUntilSystemAllowApply();
165-
} else {
168+
if (!Utils.stallApply(consensusGroupType) || !waitUntilSystemAllowApply()) {
166169
break;
167170
}
168171
}
169-
} while (Utils.stallApply(consensusGroupType));
172+
} while (true);
170173

171174
if (isLeader) {
172175
// only record time cost for data region in Performance Overview Dashboard
@@ -182,16 +185,35 @@ public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
182185
return CompletableFuture.completedFuture(ret);
183186
}
184187

185-
private void waitUntilSystemAllowApply() {
188+
/**
189+
* @return true if the wait completed normally, false if interrupted
190+
*/
191+
private boolean waitUntilSystemAllowApply() {
186192
try {
187193
Retriable.attemptUntilTrue(
188194
() -> !Utils.stallApply(consensusGroupType),
189195
TimeDuration.ONE_MINUTE,
190196
"waitUntilSystemAllowApply",
191197
logger);
198+
return true;
199+
} catch (InterruptedException e) {
200+
logger.warn(RatisMessages.INTERRUPTED_WAITING_SYSTEM_READY, this, e);
201+
Thread.currentThread().interrupt();
202+
return false;
203+
}
204+
}
205+
206+
/**
207+
* @return true if the sleep completed normally, false if interrupted
208+
*/
209+
private boolean waitBeforeRetry() {
210+
try {
211+
TimeUnit.MINUTES.sleep(1);
212+
return true;
192213
} catch (InterruptedException e) {
193214
logger.warn(RatisMessages.INTERRUPTED_WAITING_SYSTEM_READY, this, e);
194215
Thread.currentThread().interrupt();
216+
return false;
195217
}
196218
}
197219

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2323
import org.apache.iotdb.commons.exception.IllegalPathException;
24+
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
2425
import org.apache.iotdb.commons.exception.SemanticException;
2526
import org.apache.iotdb.commons.path.MeasurementPath;
2627
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
@@ -88,6 +89,8 @@ public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) {
8889
} catch (WriteProcessException e) {
8990
LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE, node, e);
9091
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
92+
} catch (MetadataLeaseFencedException e) {
93+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
9194
}
9295
}
9396

@@ -132,6 +135,8 @@ public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion
132135
}
133136
}
134137
return firstStatus;
138+
} catch (final MetadataLeaseFencedException e) {
139+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
135140
}
136141
}
137142

@@ -170,6 +175,8 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
170175
} catch (SemanticException | TableLostRuntimeException e) {
171176
LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage());
172177
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
178+
} catch (MetadataLeaseFencedException e) {
179+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
173180
}
174181
}
175182

@@ -205,6 +212,8 @@ public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion
205212
}
206213
}
207214
return firstStatus;
215+
} catch (MetadataLeaseFencedException e) {
216+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
208217
}
209218
}
210219

@@ -243,6 +252,8 @@ public TSStatus visitInsertRowsOfOneDevice(
243252
}
244253
}
245254
return firstStatus;
255+
} catch (MetadataLeaseFencedException e) {
256+
return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
246257
}
247258
}
248259

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

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -253,7 +253,7 @@ protected TSStatus write(PlanNode planNode) {
253253
while (retryTime < MAX_WRITE_RETRY_TIMES) {
254254
result = planNode.accept(new DataExecutionVisitor(), region);
255255
// Let pipe retry with the original event instead of retrying a possibly mutated node here.
256-
if (needRetry(result.getCode()) && !planNode.isGeneratedByPipe()) {
256+
if (needRetryForSpecificCases(result.getCode(), planNode)) {
257257
retryTime++;
258258
logger.debug(
259259
DataNodePipeMessages.PIPE_LOG_WRITE_OPERATION_FAILED_BECAUSE_RETRYTIME_34EFBE99,
@@ -330,10 +330,16 @@ public File getSnapshotRoot() {
330330
}
331331
}
332332

333-
public static boolean needRetry(int statusCode) {
334-
// To fix the atomicity problem, we only need to add retry for system reject.
333+
public static boolean needRetryForSpecificCases(int statusCode, PlanNode planNode) {
334+
// To fix the atomicity problem, retry system rejection and a fenced metadata lease that
335+
// explicitly requires retry.
335336
// In other cases, such as readonly, we can return directly because there are retries at the
336337
// consensus layer.
337-
return statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode();
338+
if (statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()
339+
&& !planNode.isGeneratedByPipe()) {
340+
return true;
341+
}
342+
return statusCode == TSStatusCode.METADATA_LEASE_FENCED_RETRY_REQUIRED.getStatusCode()
343+
&& !planNode.isGeneratedByPipe();
338344
}
339345
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/partition/PartitionCache.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.iotdb.commons.consensus.ConfigRegionId;
3131
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
3232
import org.apache.iotdb.commons.exception.MetadataException;
33+
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
3334
import org.apache.iotdb.commons.memory.IMemoryBlock;
3435
import org.apache.iotdb.commons.memory.MemoryBlockType;
3536
import org.apache.iotdb.commons.partition.DataPartition;
@@ -145,7 +146,8 @@ public PartitionCache() {
145146
}
146147

147148
protected void failIfMetadataLeaseFenced() {
148-
MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced();
149+
MetadataLeaseManager.getInstance()
150+
.failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
149151
}
150152

151153
// region database cache

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TreeDeviceSchemaCacheManager.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache;
2121

2222
import org.apache.iotdb.commons.conf.CommonDescriptor;
23+
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
2324
import org.apache.iotdb.commons.path.MeasurementPath;
2425
import org.apache.iotdb.commons.path.PartialPath;
2526
import org.apache.iotdb.commons.path.PathPatternUtil;
@@ -74,7 +75,8 @@ public static TreeDeviceSchemaCacheManager getInstance() {
7475
}
7576

7677
void failIfMetadataLeaseFenced() {
77-
MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced();
78+
MetadataLeaseManager.getInstance()
79+
.failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
7880
}
7981

8082
/** singleton pattern. */

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
2323
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
2424
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
25+
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
2526
import org.apache.iotdb.commons.utils.TestOnly;
2627
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2728
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
@@ -278,9 +279,10 @@ void checkLeaseStatus() {
278279
* refuse to serve it rather than risk validating writes/queries against stale schema and
279280
* producing dirty data.
280281
*/
281-
public void failIfMetadataLeaseFenced() {
282+
public void failIfMetadataLeaseFenced(final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
282283
if (isFenced()) {
283-
throw new MetadataLeaseFencedException(DataNodeSchemaMessages.METADATA_LEASE_IS_FENCED);
284+
throw new MetadataLeaseFencedException(
285+
DataNodeSchemaMessages.METADATA_LEASE_IS_FENCED, leaseFencedRetryPolicy);
284286
}
285287
}
286288

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java‎

Lines changed: 38 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.commons.client.exception.ClientManagerException;
2525
import org.apache.iotdb.commons.consensus.ConfigRegionId;
2626
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
27+
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
2728
import org.apache.iotdb.commons.exception.SemanticException;
2829
import org.apache.iotdb.commons.schema.table.NonCommittableTsTable;
2930
import org.apache.iotdb.commons.schema.table.PreDeleteTsTable;
@@ -102,8 +103,8 @@ public static ITableCache getInstance() {
102103
return DataNodeTableCacheHolder.INSTANCE;
103104
}
104105

105-
void failIfMetadataLeaseFenced() {
106-
MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced();
106+
void failIfMetadataLeaseFenced(final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
107+
MetadataLeaseManager.getInstance().failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
107108
}
108109

109110
@Override
@@ -180,7 +181,7 @@ public void preUpdateTable(String database, final TsTable table, final String ol
180181
database = PathUtils.unQualifyDatabaseName(database);
181182
readWriteLock.writeLock().lock();
182183
try {
183-
failIfMetadataLeaseFenced();
184+
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
184185
specialStatusMap
185186
.computeIfAbsent(database, k -> new ConcurrentHashMap<>())
186187
.compute(
@@ -228,7 +229,7 @@ public void rollbackUpdateTable(String database, final String tableName, final S
228229
database = PathUtils.unQualifyDatabaseName(database);
229230
readWriteLock.writeLock().lock();
230231
try {
231-
failIfMetadataLeaseFenced();
232+
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
232233
// if rollback the drop table procedure, do nothing,
233234
// wait for triggering the action of pull table from CN
234235
final TsTable table = getTableFromSpecialStatusMap(database, tableName);
@@ -299,7 +300,7 @@ public void commitUpdateTable(
299300
database = PathUtils.unQualifyDatabaseName(database);
300301
readWriteLock.writeLock().lock();
301302
try {
302-
failIfMetadataLeaseFenced();
303+
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
303304
final TsTable newTable = getTableFromSpecialStatusMap(database, tableName);
304305
if (Objects.isNull(newTable)) {
305306
LOGGER.info(
@@ -422,7 +423,7 @@ public long getInstanceVersion() {
422423
public Map<String, Map<String, TsTable>> getTableSnapshot() {
423424
readWriteLock.readLock().lock();
424425
try {
425-
failIfMetadataLeaseFenced();
426+
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
426427
return databaseTableMap.entrySet().stream()
427428
.collect(
428429
Collectors.toMap(
@@ -444,7 +445,8 @@ public Map<String, Map<String, TsTable>> getTableSnapshot() {
444445

445446
@Override
446447
public TsTable getTableInWrite(final String database, final String tableName) {
447-
final TsTable result = getTableInCache(database, tableName);
448+
final TsTable result =
449+
getTableInCache(database, tableName, LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
448450
return Objects.nonNull(result) ? result : getTable(database, tableName, false);
449451
}
450452

@@ -459,21 +461,30 @@ public TsTable getTable(final String database, final String tableName) {
459461
*/
460462
@Override
461463
public TsTable getTable(String database, final String tableName, final boolean force) {
464+
return getTable(database, tableName, force, LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
465+
}
466+
467+
@Override
468+
public TsTable getTable(
469+
String database,
470+
final String tableName,
471+
final boolean force,
472+
final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
462473
database = PathUtils.unQualifyDatabaseName(database);
463474
final AtomicReference<TableNodeStatus> tableStatusRef = new AtomicReference<>();
464475
final Map<String, Map<String, Long>> specialStatusMap =
465-
mayGetTableInSpecialStatusMap(database, tableName, tableStatusRef);
476+
mayGetTableInSpecialStatusMap(database, tableName, tableStatusRef, leaseFencedRetryPolicy);
466477

467478
if (Objects.nonNull(specialStatusMap) && !specialStatusMap.isEmpty()) {
468479
Map<String, Map<String, TsTable>> fetchedTables =
469480
getTablesInConfigNode(specialStatusMap, tableStatusRef.get());
470481
if (tableStatusRef.get() == TableNodeStatus.USING) {
471-
updateUsingTable(fetchedTables, specialStatusMap);
482+
updateUsingTable(fetchedTables, specialStatusMap, leaseFencedRetryPolicy);
472483
} else {
473-
updateDeleteTable(fetchedTables, database, tableName);
484+
updateDeleteTable(fetchedTables, database, tableName, leaseFencedRetryPolicy);
474485
}
475486
}
476-
final TsTable table = getTableInCache(database, tableName);
487+
final TsTable table = getTableInCache(database, tableName, leaseFencedRetryPolicy);
477488
if (Objects.isNull(table) && force) {
478489
CommonMetadataUtils.throwTableNotExistsException(database, tableName);
479490
}
@@ -483,10 +494,11 @@ public TsTable getTable(String database, final String tableName, final boolean f
483494
private Map<String, Map<String, Long>> mayGetTableInSpecialStatusMap(
484495
final String database,
485496
final String tableName,
486-
final AtomicReference<TableNodeStatus> tableNodeStatus) {
497+
final AtomicReference<TableNodeStatus> tableNodeStatus,
498+
final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
487499
readWriteLock.readLock().lock();
488500
try {
489-
failIfMetadataLeaseFenced();
501+
failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
490502
final Map<String, Pair<TsTable, Long>> targetDatabaseMap = specialStatusMap.get(database);
491503
if (Objects.isNull(targetDatabaseMap)) {
492504
return null;
@@ -560,10 +572,11 @@ private Map<String, Map<String, TsTable>> getTablesInConfigNode(
560572

561573
private void updateUsingTable(
562574
final Map<String, Map<String, TsTable>> fetchedTables,
563-
final Map<String, Map<String, Long>> previousVersions) {
575+
final Map<String, Map<String, Long>> previousVersions,
576+
final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
564577
readWriteLock.writeLock().lock();
565578
try {
566-
failIfMetadataLeaseFenced();
579+
failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
567580
final AtomicBoolean isUpdated = new AtomicBoolean(false);
568581
fetchedTables.forEach(
569582
(qualifiedDatabase, tableInfoMap) -> {
@@ -618,10 +631,11 @@ private void updateUsingTable(
618631
private void updateDeleteTable(
619632
Map<String, Map<String, TsTable>> fetchedTables,
620633
String targetDatabase,
621-
final String targetTable) {
634+
final String targetTable,
635+
final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
622636
readWriteLock.writeLock().lock();
623637
try {
624-
failIfMetadataLeaseFenced();
638+
failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
625639
boolean isUpdated = false;
626640
boolean targetTableIsStillDeleting = false;
627641

@@ -762,10 +776,13 @@ private String compareTable(final TsTable oldTable, final TsTable newTable) {
762776
return modified ? builder.toString() : DataNodeSchemaMessages.COMPARE_TABLE_NOT_MODIFIED;
763777
}
764778

765-
private TsTable getTableInCache(final String database, final String tableName) {
779+
private TsTable getTableInCache(
780+
final String database,
781+
final String tableName,
782+
final LeaseFencedRetryPolicy leaseFencedRetryPolicy) {
766783
readWriteLock.readLock().lock();
767784
try {
768-
failIfMetadataLeaseFenced();
785+
failIfMetadataLeaseFenced(leaseFencedRetryPolicy);
769786
final TsTable result =
770787
databaseTableMap.containsKey(database)
771788
? databaseTableMap.get(database).get(tableName)
@@ -779,7 +796,7 @@ private TsTable getTableInCache(final String database, final String tableName) {
779796
}
780797

781798
public boolean isDatabaseExist(final String database) {
782-
failIfMetadataLeaseFenced();
799+
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
783800
if (databaseTableMap.containsKey(database)) {
784801
return true;
785802
}
@@ -788,7 +805,7 @@ public boolean isDatabaseExist(final String database) {
788805
.containsKey(database)) {
789806
readWriteLock.readLock().lock();
790807
try {
791-
failIfMetadataLeaseFenced();
808+
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
792809
databaseTableMap.computeIfAbsent(database, k -> new ConcurrentHashMap<>());
793810
return true;
794811
} finally {

0 commit comments

Comments
 (0)