Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-14311. Multi-threading conflict at layoutVersion when loading block pool storage. Contributed by Yicong Cai.

This commit is contained in:
Wei-Chiu Chuang 2019-08-20 10:08:55 -07:00
parent aa6995fde2
commit 4cb22cd867
1 changed files with 14 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java
View File

@ -447,22 +447,23 @@ public class DataStorage extends Storage {
StartupOption startOpt, ExecutorService executor) throws IOException {
final String bpid = nsInfo.getBlockPoolID();
final BlockPoolSliceStorage bpStorage = getBlockPoolSliceStorage(nsInfo);
Map<StorageLocation, List<Callable<StorageDirectory>>> upgradeCallableMap =
new HashMap<>();
final List<StorageDirectory> success = Lists.newArrayList();
final List<UpgradeTask> tasks = Lists.newArrayList();
for (StorageLocation dataDir : dataDirs) {
dataDir.makeBlockPoolDir(bpid, null);
try {
final List<Callable<StorageDirectory>> callables = Lists.newArrayList();
final List<Callable<StorageDirectory>> sdCallables =
Lists.newArrayList();
final List<StorageDirectory> dirs = bpStorage.recoverTransitionRead(
nsInfo, dataDir, startOpt, callables, datanode.getConf());
if (callables.isEmpty()) {
nsInfo, dataDir, startOpt, sdCallables, datanode.getConf());
if (sdCallables.isEmpty()) {
for(StorageDirectory sd : dirs) {
success.add(sd);
}
} else {
for(Callable<StorageDirectory> c : callables) {
tasks.add(new UpgradeTask(dataDir, executor.submit(c)));
}
upgradeCallableMap.put(dataDir, sdCallables);
}
} catch (IOException e) {
LOG.warn("Failed to add storage directory {} for block pool {}",
@ -470,6 +471,13 @@ public class DataStorage extends Storage {
}
}
for (Map.Entry<StorageLocation, List<Callable<StorageDirectory>>> entry :
upgradeCallableMap.entrySet()) {
for(Callable<StorageDirectory> c : entry.getValue()) {
tasks.add(new UpgradeTask(entry.getKey(), executor.submit(c)));
}
}
if (!tasks.isEmpty()) {
LOG.info("loadBlockPoolSliceStorage: {} upgrade tasks", tasks.size());
for(UpgradeTask t : tasks) {