jt2594838 commented on code in PR #18421:
URL: https://github.com/apache/iotdb/pull/18421#discussion_r3734968385
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java:
##########
@@ -186,53 +189,160 @@ public static boolean loadFilesToActiveDir(
return false;
}
final Map<String, String> attributes =
appendCurrentUserIfAbsent(loadAttributes);
- final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
+ final File targetDir =
+ ActiveLoadPathHelper.resolvePipeTransferTargetDir(targetFilePath,
attributes);
final List<File> sourceFiles = new ArrayList<>(files.size());
for (final String file : files) {
sourceFiles.add(new File(file));
}
sourceFiles.sort(Comparator.comparing(LoadUtil::isTsFile));
- transferFilesToActiveDir(targetDir, sourceFiles, isDeleteAfterLoad);
+ transferFilesToActiveDir(
+ targetDir,
+ sourceFiles,
+ isDeleteAfterLoad,
+ attributes.get(ActiveLoadPathHelper.PIPE_CONVERSION_TASK_ID_KEY));
return true;
}
static void transferFilesToActiveDir(
final File targetDir, final List<File> sourceFiles, final boolean
isDeleteAfterLoad)
throws IOException {
+ transferFilesToActiveDir(targetDir, sourceFiles, isDeleteAfterLoad, null);
+ }
+
+ static void transferFilesToActiveDir(
+ final File targetDir,
+ final List<File> sourceFiles,
+ final boolean isDeleteAfterLoad,
+ final String deterministicDirectoryName)
+ throws IOException {
+ final File transferDir =
+ new File(
+ targetDir,
+ deterministicDirectoryName == null
+ ? UUID.randomUUID().toString()
+ : ActiveLoadPathHelper.formatPipeTaskTransferDirectoryName(
+ deterministicDirectoryName));
final List<File> existingSourceFiles = new ArrayList<>(sourceFiles.size());
for (final File sourceFile : sourceFiles) {
if (sourceFile.exists()) {
existingSourceFiles.add(sourceFile);
}
}
+
+ if (deterministicDirectoryName != null && transferDir.exists()) {
+ if (!isExistingTaskComplete(transferDir, sourceFiles)) {
+ throw new
IOException(StorageEngineMessages.FAIL_TO_LOAD_TSFILE_TO_ACTIVE_DIR);
+ }
+ if (isDeleteAfterLoad) {
+ deleteSourceFiles(existingSourceFiles);
+ }
+ return;
+ }
if (existingSourceFiles.isEmpty()) {
+ if (deterministicDirectoryName != null) {
+ // A retry is successful only when either the source or the published
deterministic target
+ // proves that the handoff completed. Reporting success for two
missing paths would make
+ // the receiver claim ownership of a task whose TsFile was lost.
+ throw new
IOException(StorageEngineMessages.FAIL_TO_LOAD_TSFILE_TO_ACTIVE_DIR);
+ }
return;
}
- final File transferDir = new File(targetDir, UUID.randomUUID().toString());
+ final File stagingDir =
+ new File(
+ targetDir,
+
ActiveLoadPathHelper.formatTransferStagingDirectoryName(UUID.randomUUID().toString()));
try {
- Files.createDirectories(transferDir.toPath());
+ Files.createDirectories(stagingDir.toPath());
for (final File sourceFile : existingSourceFiles) {
- final File targetFile = new File(transferDir, sourceFile.getName());
+ final File targetFile = new File(stagingDir, sourceFile.getName());
RetryUtils.retryOnException(
() -> {
transferFile(sourceFile, targetFile, isDeleteAfterLoad);
return null;
});
}
+ try {
+ publishTransferDirectory(stagingDir, transferDir);
+ } catch (final IOException e) {
+ // Another retry may have published the same deterministic task
between the existence
+ // check above and this rename. Reuse that complete handoff instead of
overwriting it.
+ if (deterministicDirectoryName == null
+ || !isExistingTaskComplete(transferDir, sourceFiles)) {
+ throw e;
+ }
+ }
} catch (final IOException | RuntimeException e) {
- if (transferDir.exists()) {
- FileUtils.deleteFileOrDirectoryWithRetry(transferDir);
+ if (stagingDir.exists()) {
+ FileUtils.deleteFileOrDirectoryWithRetry(stagingDir);
}
throw e;
}
+ if (stagingDir.exists()) {
+ FileUtils.deleteFileOrDirectoryWithRetry(stagingDir);
+ }
if (isDeleteAfterLoad) {
deleteSourceFiles(existingSourceFiles);
}
}
+ private static boolean isExistingTaskComplete(
+ final File transferDir, final List<File> sourceFiles) {
+ if (!transferDir.isDirectory()) {
+ return false;
+ }
+
+ final File[] targetFiles = transferDir.listFiles(File::isFile);
+ if (targetFiles == null || targetFiles.length == 0) {
+ return false;
+ }
Review Comment:
Why does a non-existent or empty dir indicate a task is not complete?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
