This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch fix-can-not-delete-at-once in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 453a2ab2ee75e0d6158886d6098836d952fb8bd2 Author: Steve Yurong Su <[email protected]> AuthorDate: Tue Jul 22 16:14:57 2025 +0800 Pipe: Handle unsuccessful file deletions and hardlink creation Adds tracking and periodic retry for files that fail to delete in PipeTsFileResourceManager. Enhances FileUtils to handle FileAlreadyExistsException during hardlink creation by checking file content and retrying if necessary. These changes improve robustness in file management and cleanup. --- .../pipe/resource/tsfile/PipeTsFileResource.java | 9 +++ .../resource/tsfile/PipeTsFileResourceManager.java | 69 ++++++++++++++++++++++ .../org/apache/iotdb/commons/utils/FileUtils.java | 29 ++++++++- 3 files changed, 106 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java index 8b37f877094..3db0617a8ed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResource.java @@ -24,6 +24,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.nio.file.Files; +import java.util.Optional; import java.util.concurrent.atomic.AtomicInteger; public class PipeTsFileResource implements AutoCloseable { @@ -36,6 +37,8 @@ public class PipeTsFileResource implements AutoCloseable { private final AtomicInteger referenceCount; + private File unsuccessfulDeletedPath = null; + public PipeTsFileResource(final File hardlinkOrCopiedFile) { this.hardlinkOrCopiedFile = hardlinkOrCopiedFile; @@ -57,6 +60,10 @@ public class PipeTsFileResource implements AutoCloseable { return fileSize; } + public Optional<File> getUnsuccessfulDeletedPath() { + return Optional.ofNullable(unsuccessfulDeletedPath); + } + ///////////////////// Reference Count ///////////////////// public int getReferenceCount() { @@ -82,6 +89,7 @@ public class PipeTsFileResource implements AutoCloseable { @Override public synchronized void close() { boolean successful = false; + try { successful = Files.deleteIfExists(hardlinkOrCopiedFile.toPath()); } catch (final Exception e) { @@ -90,6 +98,7 @@ public class PipeTsFileResource implements AutoCloseable { hardlinkOrCopiedFile, e.getMessage(), e); + unsuccessfulDeletedPath = hardlinkOrCopiedFile; } if (successful) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index d1d1286991b..77d38f03d50 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.tsfile.enums.TSDataType; @@ -35,15 +36,22 @@ import javax.annotation.Nullable; import java.io.File; import java.io.IOException; +import java.nio.file.Files; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.TimeUnit; public class PipeTsFileResourceManager { private static final Logger LOGGER = LoggerFactory.getLogger(PipeTsFileResourceManager.class); + private static final Set<File> UNSUCCESSFULLY_DELETED = new CopyOnWriteArraySet<>(); + // This is used to hold the assigner pinned tsFiles. // Also, it is used to provide metadata cache of the tsFile, and is shared by all the pipe's // tsFiles. @@ -55,6 +63,66 @@ public class PipeTsFileResourceManager { hardlinkOrCopiedFileToPipeTsFileResourceMap = new ConcurrentHashMap<>(); private final PipeTsFileResourceSegmentLock segmentLock = new PipeTsFileResourceSegmentLock(); + public PipeTsFileResourceManager() { + PipeDataNodeAgent.runtime() + .registerPeriodicalJob( + "PipeTsFileResourceManager#tryDeleteAgainForUnsuccessfulUnlinked()", + this::tryDeleteAgainForUnsuccessfulUnlinked, + // 30 mins in seconds + 30 * 60); + } + + private void tryDeleteAgainForUnsuccessfulUnlinked() { + if (UNSUCCESSFULLY_DELETED.isEmpty()) { + return; + } + + LOGGER.info( + "Trying to delete {} unsuccessfully unlinked files {} again", + UNSUCCESSFULLY_DELETED.size(), + UNSUCCESSFULLY_DELETED); + + try { + deleteAgainForUnsuccessfulUnlinked(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn("failed to try lock when deleting unsuccessfully unlinked files", e); + } catch (final Exception e) { + LOGGER.warn("failed to try lock when deleting unsuccessfully unlinked files", e); + } + } + + private void deleteAgainForUnsuccessfulUnlinked() throws InterruptedException { + final Iterator<File> iterator = UNSUCCESSFULLY_DELETED.iterator(); + final long timeout = + PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds() >> 1; + + while (iterator.hasNext()) { + final File file = iterator.next(); + if (!segmentLock.tryLock(file, timeout, TimeUnit.SECONDS)) { + LOGGER.warn( + "failed to try lock when deleting unsuccessfully unlinked file {} because of timeout ({}s)", + file, + timeout); + continue; + } + + try { + if (Files.deleteIfExists(file.toPath())) { + LOGGER.info("Successfully deleted unsuccessfully unlinked file: {}", file); + } else { + LOGGER.info( + "Unsuccessfully unlinked file {} does not exist, no need to delete it again.", file); + } + iterator.remove(); + } catch (final Exception e) { + LOGGER.warn("failed to delete unsuccessfully unlinked file {}", file, e); + } finally { + segmentLock.unlock(file); + } + } + } + public File increaseFileReference( final File file, final boolean isTsFile, final @Nullable String pipeName) throws IOException { return increaseFileReference(file, isTsFile, pipeName, null); @@ -220,6 +288,7 @@ public class PipeTsFileResourceManager { final PipeTsFileResource resource = getResourceMap(pipeName).get(filePath); if (resource != null && resource.decreaseReferenceCount()) { getResourceMap(pipeName).remove(filePath); + resource.getUnsuccessfulDeletedPath().ifPresent(UNSUCCESSFULLY_DELETED::add); } // Decrease the assigner's file to clear hard-link and memory cache // Note that it does not exist for historical files diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java index ea023b1487e..df24fd58d0c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java @@ -35,6 +35,7 @@ import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; import java.nio.file.DirectoryNotEmptyException; +import java.nio.file.FileAlreadyExistsException; import java.nio.file.FileSystems; import java.nio.file.Files; import java.nio.file.NoSuchFileException; @@ -347,7 +348,33 @@ public class FileUtils { final Path sourcePath = FileSystems.getDefault().getPath(sourceFile.getAbsolutePath()); final Path linkPath = FileSystems.getDefault().getPath(hardlink.getAbsolutePath()); - Files.createLink(linkPath, sourcePath); + try { + Files.createLink(linkPath, sourcePath); + } catch (final FileAlreadyExistsException fileAlreadyExistsException) { + if (haveSameMD5(sourceFile, hardlink)) { + LOGGER.warn( + "Hardlink {} already exists, will not create it again. Source file: {}", + hardlink.getAbsolutePath(), + sourceFile.getAbsolutePath()); + } else { + LOGGER.warn( + "Hardlink {} already exists but does not match source file {}, will try create it again.", + hardlink.getAbsolutePath(), + sourceFile.getAbsolutePath()); + deleteFileIfExist(hardlink); + try { + Files.createLink(linkPath, sourcePath); + } catch (final Exception e) { + LOGGER.error( + "Failed to create hardlink {} for file {}: {}", + hardlink.getAbsolutePath(), + sourceFile.getAbsolutePath(), + e.getMessage(), + e); + throw e; + } + } + } return hardlink; }
