Skip to content

Commit 14dc895

Browse files
CRZbulabulaCRZbulabula
committed
Fix schema region visibility race and remove datanode message
1 parent faba002 commit 14dc895

7 files changed

Lines changed: 165 additions & 18 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,38 @@ public void tearDown() throws InterruptedException {
135135
// ConsensusFactory.IOT_CONSENSUS_V2);
136136
// }
137137

138+
@Test
139+
public void failWhenDataReplicationFactorIsOneUseSQL() throws Exception {
140+
EnvFactory.getEnv()
141+
.getConfig()
142+
.getCommonConfig()
143+
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
144+
.setSchemaReplicationFactor(3)
145+
.setDataReplicationFactor(1)
146+
.setDefaultDataRegionGroupNumPerDatabase(1);
147+
EnvFactory.getEnv().initClusterEnvironment(1, 3);
148+
149+
try (final Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection());
150+
final Statement statement = makeItCloseQuietly(connection.createStatement());
151+
final ResultSet resultSet = statement.executeQuery(SHOW_DATANODES)) {
152+
final Set<Integer> allDataNodeId = new HashSet<>();
153+
while (resultSet.next()) {
154+
allDataNodeId.add(resultSet.getInt(ColumnHeaderConstant.NODE_ID));
155+
}
156+
157+
final String removeDataNodeSQL =
158+
generateRemoveString(selectRemoveDataNodes(allDataNodeId, 1));
159+
try {
160+
statement.execute(removeDataNodeSQL);
161+
Assert.fail("Remove DataNode should fail when data_replication_factor is 1");
162+
} catch (final IoTDBSQLException e) {
163+
Assert.assertTrue(e.getMessage(), e.getMessage().contains("data_replication_factor is 1"));
164+
Assert.assertFalse(
165+
e.getMessage(), e.getMessage().contains("Failed to remove all requested data nodes"));
166+
}
167+
}
168+
}
169+
138170
@Test
139171
public void fail1C3DTestIoTUseSQL() throws Exception {
140172
// Setup 1C3D with schema replication factor = 3, and remove 1D, this test should fail due to

‎iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -445,6 +445,10 @@ public final class ProcedureMessages {
445445
"Failed to push topic meta to dataNodes, details: %s";
446446
public static final String FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN =
447447
"Failed to remove data node {} because it is not in running and the configuration of cluster is one replication";
448+
449+
public static final String
450+
FAILED_TO_REMOVE_DATA_NODE_BECAUSE_DATA_REPLICATION_FACTOR_IS_ONE =
451+
"Cannot remove DataNode because data_replication_factor is 1 or at least one DataRegion has only one replica. Removing a DataNode may cause data loss. Increase data_replication_factor and ensure each DataRegion has more than one replica before removing DataNodes.";
448452
public static final String FAILED_TO_ROLLBACK_ALTER_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED =
449453
"Failed to rollback alter pipe {}, details: {}, metadata will be synchronized later.";
450454
public static final String FAILED_TO_ROLLBACK_COMMIT_SET_TEMPLATE_ON_PATH_DUE_TO =

‎iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -445,6 +445,10 @@ public final class ProcedureMessages {
445445
"Failed to push topic meta to dataNodes, details: %s";
446446
public static final String FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN =
447447
"Failed to remove data node {} because it is not in running and the configuration of cluster is one replication";
448+
449+
public static final String
450+
FAILED_TO_REMOVE_DATA_NODE_BECAUSE_DATA_REPLICATION_FACTOR_IS_ONE =
451+
"不能移除 DataNode，因为 data_replication_factor 为 1，或至少存在一个 DataRegion 只有一个副本。移除 DataNode 可能造成数据丢失。请先提高 data_replication_factor，并确保每个 DataRegion 都有多个副本，再移除 DataNode。";
448452
public static final String FAILED_TO_ROLLBACK_ALTER_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED =
449453
"Failed to rollback alter pipe {}, details: {}, metadata will be synchronized later.";
450454
public static final String FAILED_TO_ROLLBACK_COMMIT_SET_TEMPLATE_ON_PATH_DUE_TO =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java‎

Lines changed: 67 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,9 @@ public class PartitionManager {
147147
public static final String CONSENSUS_WRITE_ERROR =
148148
"Failed in the write API executing the consensus layer due to: ";
149149

150+
private static final long REGION_GROUP_VISIBILITY_TIMEOUT_MS = 10_000L;
151+
private static final long REGION_GROUP_VISIBILITY_CHECK_INTERVAL_MS = 20L;
152+
150153
// Monitor for leadership change
151154
private final Object scheduleMonitor = new Object();
152155

@@ -715,12 +718,75 @@ private TSStatus generateAndAllocateRegionGroups(
715718
getLoadManager().allocateRegionGroups(allotmentMap, consensusGroupType);
716719
LOGGER.info(ManagerMessages.CREATEREGIONGROUPS_STARTING_TO_CREATE_THE_FOLLOWING_REGIONGROUPS);
717720
createRegionGroupsPlan.planLog(LOGGER);
718-
return getProcedureManager().createRegionGroups(consensusGroupType, createRegionGroupsPlan);
721+
final TSStatus createStatus =
722+
getProcedureManager().createRegionGroups(consensusGroupType, createRegionGroupsPlan);
723+
if (createStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
724+
return createStatus;
725+
}
726+
return waitForRegionGroupsVisible(createRegionGroupsPlan, consensusGroupType);
719727
} else {
720728
return RpcUtils.SUCCESS_STATUS;
721729
}
722730
}
723731

732+
private TSStatus waitForRegionGroupsVisible(
733+
final CreateRegionGroupsPlan createRegionGroupsPlan,
734+
final TConsensusGroupType consensusGroupType) {
735+
final Map<String, Set<TConsensusGroupId>> expectedRegionGroups = new HashMap<>();
736+
createRegionGroupsPlan
737+
.getRegionGroupMap()
738+
.forEach(
739+
(database, regionReplicaSets) -> {
740+
final Set<TConsensusGroupId> regionGroupIds =
741+
regionReplicaSets.stream()
742+
.map(TRegionReplicaSet::getRegionId)
743+
.filter(regionGroupId -> consensusGroupType.equals(regionGroupId.getType()))
744+
.collect(Collectors.toSet());
745+
if (!regionGroupIds.isEmpty()) {
746+
expectedRegionGroups.put(database, regionGroupIds);
747+
}
748+
});
749+
750+
final long startTime = System.currentTimeMillis();
751+
while (System.currentTimeMillis() - startTime <= REGION_GROUP_VISIBILITY_TIMEOUT_MS) {
752+
if (areRegionGroupsVisible(expectedRegionGroups, consensusGroupType)) {
753+
return RpcUtils.SUCCESS_STATUS;
754+
}
755+
try {
756+
TimeUnit.MILLISECONDS.sleep(REGION_GROUP_VISIBILITY_CHECK_INTERVAL_MS);
757+
} catch (final InterruptedException e) {
758+
Thread.currentThread().interrupt();
759+
return new TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode())
760+
.setMessage(
761+
String.format(
762+
"Interrupted while waiting for created %s RegionGroups %s to become visible in PartitionInfo.",
763+
consensusGroupType, expectedRegionGroups));
764+
}
765+
}
766+
767+
final String message =
768+
String.format(
769+
"Created %s RegionGroups %s are not visible in PartitionInfo within %d ms.",
770+
consensusGroupType, expectedRegionGroups, REGION_GROUP_VISIBILITY_TIMEOUT_MS);
771+
LOGGER.warn(message);
772+
return new TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode()).setMessage(message);
773+
}
774+
775+
private boolean areRegionGroupsVisible(
776+
final Map<String, Set<TConsensusGroupId>> expectedRegionGroups,
777+
final TConsensusGroupType consensusGroupType) {
778+
for (final Map.Entry<String, Set<TConsensusGroupId>> entry : expectedRegionGroups.entrySet()) {
779+
final Set<TConsensusGroupId> visibleRegionGroups =
780+
partitionInfo.getRegionGroupSlotsCounter(entry.getKey(), consensusGroupType).stream()
781+
.map(Pair::getRight)
782+
.collect(Collectors.toSet());
783+
if (!visibleRegionGroups.containsAll(entry.getValue())) {
784+
return false;
785+
}
786+
}
787+
return true;
788+
}
789+
724790
/**
725791
* Only leader use this interface. Checks whether the specified DataPartition has a successor and
726792
* returns if it does.

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -504,12 +504,15 @@ public List<TSStatus> notifyRegionMigrationToAllDataNodes(
504504
return clientHandler.getResponseList();
505505
}
506506

507-
public void persistRegionGroup(CreateRegionGroupsPlan createRegionGroupsPlan) {
507+
public TSStatus persistRegionGroup(CreateRegionGroupsPlan createRegionGroupsPlan) {
508508
// Persist the allocation result
509509
try {
510-
getConsensusManager().write(createRegionGroupsPlan);
510+
return getConsensusManager().write(createRegionGroupsPlan);
511511
} catch (ConsensusException e) {
512512
LOG.warn("Failed in the write API executing the consensus layer due to: ", e);
513+
return new TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode())
514+
.setMessage(
515+
"Failed to persist RegionGroup allocation in the consensus layer: " + e.getMessage());
513516
}
514517
}
515518

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RemoveDataNodeHandler.java‎

Lines changed: 44 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -559,27 +559,41 @@ public TSStatus checkRegionReplication(RemoveDataNodePlan removeDataNodePlan) {
559559
TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
560560
List<TDataNodeLocation> removedDataNodes = removeDataNodePlan.getDataNodeLocations();
561561

562+
if (hasSingleDataRegionReplica()) {
563+
status.setCode(TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode());
564+
status.setMessage(
565+
ProcedureMessages.FAILED_TO_REMOVE_DATA_NODE_BECAUSE_DATA_REPLICATION_FACTOR_IS_ONE);
566+
LOGGER.error(status.getMessage());
567+
return status;
568+
}
569+
562570
int availableDatanodeSize =
563571
configManager
564572
.getNodeManager()
565573
.filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.ReadOnly)
566574
.size();
567575
// when the configuration is one replication, it will be failed if the data node is not in
568576
// running state.
569-
if (CONF.getSchemaReplicationFactor() == 1 || CONF.getDataReplicationFactor() == 1) {
570-
for (TDataNodeLocation dataNodeLocation : removedDataNodes) {
571-
// check whether removed data node is in running state
572-
if (!NodeStatus.Running.equals(
573-
configManager.getLoadManager().getNodeStatus(dataNodeLocation.getDataNodeId()))) {
574-
removedDataNodes.remove(dataNodeLocation);
575-
LOGGER.error(
576-
ProcedureMessages.FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN, dataNodeLocation);
577-
}
578-
if (removedDataNodes.isEmpty()) {
579-
status.setCode(TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode());
580-
status.setMessage(ProcedureMessages.FAILED_TO_REMOVE_ALL_REQUESTED_DATA_NODES);
581-
return status;
582-
}
577+
if (CONF.getSchemaReplicationFactor() == 1) {
578+
final List<TDataNodeLocation> notRunningDataNodes =
579+
removedDataNodes.stream()
580+
.filter(
581+
dataNodeLocation ->
582+
!NodeStatus.Running.equals(
583+
configManager
584+
.getLoadManager()
585+
.getNodeStatus(dataNodeLocation.getDataNodeId())))
586+
.collect(Collectors.toList());
587+
notRunningDataNodes.forEach(
588+
dataNodeLocation ->
589+
LOGGER.error(
590+
ProcedureMessages.FAILED_TO_REMOVE_DATA_NODE_BECAUSE_IT_IS_NOT_IN,
591+
dataNodeLocation));
592+
removedDataNodes.removeAll(notRunningDataNodes);
593+
if (removedDataNodes.isEmpty()) {
594+
status.setCode(TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode());
595+
status.setMessage(ProcedureMessages.FAILED_TO_REMOVE_ALL_REQUESTED_DATA_NODES);
596+
return status;
583597
}
584598
}
585599

@@ -604,6 +618,22 @@ public TSStatus checkRegionReplication(RemoveDataNodePlan removeDataNodePlan) {
604618
return status;
605619
}
606620

621+
private boolean hasSingleDataRegionReplica() {
622+
return CONF.getDataReplicationFactor() == 1
623+
|| configManager
624+
.getClusterSchemaManager()
625+
.getMatchedDatabaseSchemasByName(
626+
configManager.getClusterSchemaManager().getDatabaseNames(null), null)
627+
.values()
628+
.stream()
629+
.anyMatch(databaseSchema -> databaseSchema.getDataReplicationFactor() == 1)
630+
|| configManager
631+
.getPartitionManager()
632+
.getAllReplicaSets(TConsensusGroupType.DataRegion)
633+
.stream()
634+
.anyMatch(replicaSet -> replicaSet.getDataNodeLocationsSize() <= 1);
635+
}
636+
607637
/**
608638
* Checks whether all DataNodes specified for deletion exist in the cluster.
609639
*

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/CreateRegionGroupsProcedure.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,9 @@
2323
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
2424
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
2525
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
26+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2627
import org.apache.iotdb.commons.cluster.RegionStatus;
28+
import org.apache.iotdb.commons.exception.IoTDBException;
2729
import org.apache.iotdb.commons.utils.TestOnly;
2830
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
2931
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
@@ -36,10 +38,12 @@
3638
import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionCreateTask;
3739
import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionDeleteTask;
3840
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
41+
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
3942
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
4043
import org.apache.iotdb.confignode.procedure.state.CreateRegionGroupsState;
4144
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
4245
import org.apache.iotdb.consensus.exception.ConsensusException;
46+
import org.apache.iotdb.rpc.TSStatusCode;
4347

4448
import org.slf4j.Logger;
4549
import org.slf4j.LoggerFactory;
@@ -173,7 +177,11 @@ protected Flow executeFromState(
173177
}
174178
}));
175179

176-
env.persistRegionGroup(persistPlan);
180+
final TSStatus persistStatus = env.persistRegionGroup(persistPlan);
181+
if (persistStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
182+
setFailure(new ProcedureException(new IoTDBException(persistStatus)));
183+
return Flow.NO_MORE_STATE;
184+
}
177185
try {
178186
env.getConfigManager().getConsensusManager().write(offerPlan);
179187
} catch (final ConsensusException e) {

0 commit comments

Comments
 (0)