From 5ca66d531690719b0178008c68a872b19b633c7f Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:08:03 +0800 Subject: [PATCH 1/2] Fix active load reliability issues --- .../plan/node/load/LoadSingleTsFileNode.java | 28 ++-- .../scheduler/load/LoadTsFileScheduler.java | 15 +- .../load/active/ActiveLoadDirScanner.java | 11 +- .../load/active/ActiveLoadPendingQueue.java | 20 ++- .../load/active/ActiveLoadTsFileLoader.java | 40 ++++- .../db/storageengine/load/util/LoadUtil.java | 116 +++++++++---- .../planner/node/load/LoadTsFileNodeTest.java | 44 +++++ .../load/LoadTsFileSchedulerTest.java | 40 +++++ .../load/active/ActiveLoadDirScannerTest.java | 153 ++++++++++++++++++ .../active/ActiveLoadTsFileLoaderTest.java | 18 +++ .../storageengine/load/util/LoadUtilTest.java | 128 +++++++++++++++ 11 files changed, 563 insertions(+), 50 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java index dae0a63e97482..ecaf0aea6ec2f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java @@ -25,6 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.commons.utils.TimePartitionUtils; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; @@ -241,17 +242,24 @@ public String toString() { } public void clean() { + if (!deleteAfterLoad) { + return; + } + deleteFile(tsFile); + deleteFile(new File(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath()))); + deleteFile(ModificationFile.getExclusiveMods(tsFile)); + deleteFile(new File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath()))); + } + + private void deleteFile(final File file) { try { - if (deleteAfterLoad) { - Files.deleteIfExists(tsFile.toPath()); - Files.deleteIfExists( - new File(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath())).toPath()); - Files.deleteIfExists(ModificationFile.getExclusiveMods(tsFile).toPath()); - Files.deleteIfExists( - new File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath())).toPath()); - } - } catch (final IOException e) { - LOGGER.warn(DataNodeQueryMessages.DELETE_AFTER_LOADING_ERROR, tsFile, e); + RetryUtils.retryOnException( + () -> { + Files.deleteIfExists(file.toPath()); + return null; + }); + } catch (final Exception e) { + LOGGER.warn(DataNodeQueryMessages.DELETE_AFTER_LOADING_ERROR, file, e); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index d3beefc0802b6..ef2e948c0e6c2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -800,8 +800,7 @@ private boolean addOrSendChunkData(ChunkData chunkData) throws LoadFileException singleTsFileNode .getTsFileResource() .getTsFile()))); // can not just remove, because of deletion - dataSize -= pieceNode.getDataSize(); - block.reduceMemoryUsage(pieceNode.getDataSize()); + releaseMemoryUsage(pieceNode.getDataSize()); if (!isDispatchSuccess) { // Currently there is no retry, so return directly @@ -886,7 +885,7 @@ private boolean sendAllTsFileData() throws LoadFileException { boolean isAllSuccess = true; for (Map.Entry> entry : regionId2ReplicaSetAndNode.entrySet()) { - block.reduceMemoryUsage(entry.getValue().getRight().getDataSize()); + releaseMemoryUsage(entry.getValue().getRight().getDataSize()); if (isAllSuccess && !scheduler.dispatchOnePieceNode( entry.getValue().getRight(), entry.getValue().getLeft())) { @@ -900,7 +899,17 @@ private boolean sendAllTsFileData() throws LoadFileException { return isAllSuccess; } + private void releaseMemoryUsage(final long memorySize) { + dataSize -= memorySize; + block.reduceMemoryUsage(memorySize); + } + private void clear() { + if (dataSize > 0) { + block.reduceMemoryUsage(dataSize); + dataSize = 0; + } + nonDirectionalChunkData.clear(); regionId2ReplicaSetAndNode.clear(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java index cb3460d78e706..9d1ad235ecb1d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java @@ -108,21 +108,22 @@ private void scan() throws IOException { FileUtils.streamFiles(listeningDirFile, true, (String[]) null)) { try { fileStream + .map(file -> new File(LoadUtil.getTsFilePath(file.getAbsolutePath()))) + .distinct() .filter(file -> !activeLoadTsFileLoader.isFilePendingOrLoading(file)) .filter(File::exists) - .map(file -> LoadUtil.getTsFilePath(file.getAbsolutePath())) - .filter(this::isTsFileCompleted) + .filter(file -> isTsFileCompleted(file.getAbsolutePath())) .limit(currentAllowedPendingSize) .forEach( - filePath -> { - final File tsFile = new File(filePath); + tsFile -> { final Map attributes = ActiveLoadPathHelper.parseAttributes(tsFile, listeningDirFile); final File parentFile = tsFile.getParentFile(); final boolean isTableModel = ActiveLoadPathHelper.containsDatabaseName(attributes) - || (parentFile != null + || (attributes.isEmpty() + && parentFile != null && !Objects.equals( parentFile.getAbsoluteFile(), listeningDirFile.getAbsoluteFile())); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java index 88c18b19cb60b..060935f51d058 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPendingQueue.java @@ -61,9 +61,9 @@ public synchronized ActiveLoadEntry dequeueFromPending() { } public synchronized void removeFromLoading(final String file) { - loadingFileSet.remove(file); - - ActiveLoadingFilesNumberMetricsSet.getInstance().increaseLoadingFileCounter(-1); + if (loadingFileSet.remove(file)) { + ActiveLoadingFilesNumberMetricsSet.getInstance().increaseLoadingFileCounter(-1); + } } public synchronized boolean isFilePendingOrLoading(final String file) { @@ -78,6 +78,20 @@ public synchronized boolean isEmpty() { return pendingFileQueue.isEmpty() && loadingFileSet.isEmpty(); } + public synchronized void clear() { + final int loadingFileCount = loadingFileSet.size(); + clearPending(); + loadingFileSet.clear(); + ActiveLoadingFilesNumberMetricsSet.getInstance().increaseLoadingFileCounter(-loadingFileCount); + } + + public synchronized void clearPending() { + final int pendingFileCount = pendingFileSet.size(); + pendingFileSet.clear(); + pendingFileQueue.clear(); + ActiveLoadingFilesNumberMetricsSet.getInstance().increaseQueuingFileCounter(-pendingFileCount); + } + public static class ActiveLoadEntry { private final String file; private final String pendingDir; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java index ff5222e575440..1f0c178660889 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java @@ -160,12 +160,15 @@ private void adjustExecutorIfNecessary() { public void stop() { final WrappedThreadPoolExecutor executor = activeLoadExecutor.getAndSet(null); if (executor == null) { + pendingQueue.clearPending(); return; } - executor.shutdownNow(); + boolean isTerminated = false; try { - if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { + executor.shutdownNow(); + isTerminated = executor.awaitTermination(30, TimeUnit.SECONDS); + if (!isTerminated) { LOGGER.warn( StorageEngineMessages.STILL_NOT_EXIT_AFTER_30S, ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName()); @@ -175,6 +178,12 @@ public void stop() { StorageEngineMessages.STILL_NOT_EXIT_AFTER_30S, ThreadName.ACTIVE_LOAD_TSFILE_LOADER.getName()); Thread.currentThread().interrupt(); + } finally { + if (isTerminated) { + pendingQueue.clear(); + } else { + pendingQueue.clearPending(); + } } } @@ -213,6 +222,7 @@ private void tryLoadPendingTsFiles() { handleOtherException(loadEntry.get(), e); } finally { pendingQueue.removeFromLoading(loadEntry.get().getFile()); + cleanupEmptyDirectories(loadEntry.get()); } } } finally { @@ -364,6 +374,32 @@ private void removeToFailDir(final String filePath) { } } + private void cleanupEmptyDirectories(final ActiveLoadPendingQueue.ActiveLoadEntry entry) { + final File pendingDir = + entry.getPendingDir() == null + ? ActiveLoadPathHelper.findPendingDirectory(new File(entry.getFile())) + : new File(entry.getPendingDir()); + if (pendingDir == null) { + return; + } + + final Path pendingPath = pendingDir.toPath().toAbsolutePath().normalize(); + Path currentPath = new File(entry.getFile()).toPath().toAbsolutePath().normalize().getParent(); + while (currentPath != null + && currentPath.startsWith(pendingPath) + && !currentPath.equals(pendingPath)) { + try { + Files.delete(currentPath); + } catch (final IOException e) { + if (Files.exists(currentPath)) { + LOGGER.debug(StorageEngineMessages.FAILED_DELETE_FOLDER_CLEANING_UP, currentPath, e); + } + return; + } + currentPath = currentPath.getParent(); + } + } + public boolean isFilePendingOrLoading(final File file) { return pendingQueue.isFilePendingOrLoading(file.getAbsolutePath()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java index 88c71fb1cef28..590a8b0edbbca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java @@ -39,14 +39,16 @@ import java.io.File; import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Comparator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; - -import static org.apache.iotdb.commons.utils.FileUtils.copyFileWithMD5Check; -import static org.apache.iotdb.commons.utils.FileUtils.moveFileWithMD5Check; +import java.util.UUID; public class LoadUtil { @@ -130,13 +132,14 @@ private static boolean loadTsFilesToActiveDir( final Map attributes = appendCurrentUserIfAbsent(loadAttributes); final File targetDir = ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes); - loadTsFileAsyncToTargetDir( - targetDir, new File(getTsFileResourcePath(file.getAbsolutePath())), isDeleteAfterLoad); - loadTsFileAsyncToTargetDir( - targetDir, new File(getTsFileModsV1Path(file.getAbsolutePath())), isDeleteAfterLoad); - loadTsFileAsyncToTargetDir( - targetDir, new File(getTsFileModsV2Path(file.getAbsolutePath())), isDeleteAfterLoad); - loadTsFileAsyncToTargetDir(targetDir, file, isDeleteAfterLoad); + transferFilesToActiveDir( + targetDir, + Arrays.asList( + new File(getTsFileResourcePath(file.getAbsolutePath())), + new File(getTsFileModsV1Path(file.getAbsolutePath())), + new File(getTsFileModsV2Path(file.getAbsolutePath())), + file), + isDeleteAfterLoad); return true; } @@ -183,32 +186,91 @@ public static boolean loadFilesToActiveDir( final Map attributes = appendCurrentUserIfAbsent(loadAttributes); final File targetDir = ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes); + final List sourceFiles = new ArrayList<>(files.size()); for (final String file : files) { - loadTsFileAsyncToTargetDir(targetDir, new File(file), isDeleteAfterLoad); + sourceFiles.add(new File(file)); } + sourceFiles.sort(Comparator.comparing(LoadUtil::isTsFile)); + transferFilesToActiveDir(targetDir, sourceFiles, isDeleteAfterLoad); return true; } - private static void loadTsFileAsyncToTargetDir( - final File targetDir, final File file, final boolean isDeleteAfterLoad) throws IOException { - if (!file.exists()) { + static void transferFilesToActiveDir( + final File targetDir, final List sourceFiles, final boolean isDeleteAfterLoad) + throws IOException { + final List existingSourceFiles = new ArrayList<>(sourceFiles.size()); + for (final File sourceFile : sourceFiles) { + if (sourceFile.exists()) { + existingSourceFiles.add(sourceFile); + } + } + if (existingSourceFiles.isEmpty()) { return; } - if (!targetDir.exists() && !targetDir.mkdirs()) { - if (!targetDir.exists()) { - throw new IOException( - StorageEngineMessages.FAILED_TO_CREATE_TARGET_DIR + targetDir.getAbsolutePath()); + + final File transferDir = new File(targetDir, UUID.randomUUID().toString()); + try { + Files.createDirectories(transferDir.toPath()); + for (final File sourceFile : existingSourceFiles) { + final File targetFile = new File(transferDir, sourceFile.getName()); + RetryUtils.retryOnException( + () -> { + transferFile(sourceFile, targetFile, isDeleteAfterLoad); + return null; + }); + } + } catch (final IOException | RuntimeException e) { + org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectory(transferDir, true); + throw e; + } + + if (isDeleteAfterLoad) { + deleteSourceFiles(existingSourceFiles); + } + } + + private static void transferFile( + final File sourceFile, final File targetFile, final boolean useHardLink) throws IOException { + Exception linkException = null; + if (useHardLink) { + try { + Files.createLink(targetFile.toPath(), sourceFile.toPath()); + return; + } catch (final IOException | UnsupportedOperationException | SecurityException e) { + linkException = e; + } + } + + try { + Files.copy( + sourceFile.toPath(), + targetFile.toPath(), + StandardCopyOption.REPLACE_EXISTING, + StandardCopyOption.COPY_ATTRIBUTES); + } catch (final IOException e) { + if (linkException != null) { + e.addSuppressed(linkException); + } + throw e; + } + } + + private static void deleteSourceFiles(final List sourceFiles) { + for (final File sourceFile : sourceFiles) { + try { + RetryUtils.retryOnException( + () -> { + Files.deleteIfExists(sourceFile.toPath()); + return null; + }); + } catch (final Exception e) { + LOGGER.warn(StorageEngineMessages.FAILED_TO_DELETE_FILE_OR_DIR, sourceFile, e); } } - RetryUtils.retryOnException( - () -> { - if (isDeleteAfterLoad) { - moveFileWithMD5Check(file, targetDir); - } else { - copyFileWithMD5Check(file, targetDir); - } - return null; - }); + } + + private static boolean isTsFile(final File file) { + return file.getAbsolutePath().equals(getTsFilePath(file.getAbsolutePath())); } public static ILoadDiskSelector updateLoadDiskSelector() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java index e589810894610..c7fc55a85e230 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java @@ -23,7 +23,9 @@ import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.load.util.LoadUtil; import org.apache.tsfile.exception.NotImplementedException; import org.junit.Assert; @@ -31,6 +33,7 @@ import java.io.File; import java.nio.ByteBuffer; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Collections; @@ -89,4 +92,45 @@ public void testLoadTsFilePieceNode() { LoadTsFilePieceNode node1 = (LoadTsFilePieceNode) LoadTsFilePieceNode.deserialize(buffer); Assert.assertEquals(node.getTsFile(), node1.getTsFile()); } + + @Test + public void testCleanContinuesAfterOneFileCannotBeDeleted() throws Exception { + final File tempDir = Files.createTempDirectory("load-node-clean").toFile(); + try { + final File tsFile = new File(tempDir, "1-0-0-0.tsfile"); + Assert.assertTrue(tsFile.mkdirs()); + Assert.assertTrue(new File(tsFile, "non-empty").createNewFile()); + final File resourceFile = new File(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath())); + final File modsV2File = ModificationFile.getExclusiveMods(tsFile); + final File modsV1File = new File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath())); + Assert.assertTrue(resourceFile.createNewFile()); + Assert.assertTrue(modsV2File.createNewFile()); + Assert.assertTrue(modsV1File.createNewFile()); + + final LoadSingleTsFileNode node = + new LoadSingleTsFileNode( + new PlanNodeId(""), new TsFileResource(tsFile), false, null, true, 0L, false); + node.clean(); + + Assert.assertTrue(tsFile.exists()); + Assert.assertFalse(resourceFile.exists()); + Assert.assertFalse(modsV2File.exists()); + Assert.assertFalse(modsV1File.exists()); + } finally { + deleteRecursively(tempDir); + } + } + + private static void deleteRecursively(final File file) { + if (file == null || !file.exists()) { + return; + } + final File[] children = file.listFiles(); + if (children != null) { + for (final File child : children) { + deleteRecursively(child); + } + } + Assert.assertTrue(file.delete()); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java index 19d97490c8162..6a6d4868d04ca 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.SubPlan; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileDataCacheMemoryBlock; import org.junit.Assert; import org.junit.Before; @@ -37,6 +38,8 @@ import org.mockito.MockitoAnnotations; import java.io.File; +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; import java.lang.reflect.Method; import static org.mockito.Mockito.mock; @@ -118,4 +121,41 @@ public void testBuildRetryTreeLoadStatementUpdatesDatabaseLevel() throws Excepti Assert.assertEquals(2, statement.getDatabaseLevel()); Assert.assertTrue(statement.isGeneratedByPipe()); } + + @Test + public void testTsFileDataManagerClearReleasesCachedMemory() throws Exception { + final Constructor memoryBlockConstructor = + LoadTsFileDataCacheMemoryBlock.class.getDeclaredConstructor(long.class); + memoryBlockConstructor.setAccessible(true); + final LoadTsFileDataCacheMemoryBlock memoryBlock = + memoryBlockConstructor.newInstance(1024 * 1024L); + + final Class dataManagerClass = + Class.forName(LoadTsFileScheduler.class.getName() + "$TsFileDataManager"); + final Constructor dataManagerConstructor = + dataManagerClass.getDeclaredConstructor( + LoadTsFileScheduler.class, + LoadSingleTsFileNode.class, + LoadTsFileDataCacheMemoryBlock.class); + dataManagerConstructor.setAccessible(true); + final Object dataManager = + dataManagerConstructor.newInstance( + mock(LoadTsFileScheduler.class), mock(LoadSingleTsFileNode.class), memoryBlock); + + final long cachedMemorySize = 128L; + memoryBlock.addMemoryUsage(cachedMemorySize); + final Field dataSizeField = dataManagerClass.getDeclaredField("dataSize"); + dataSizeField.setAccessible(true); + dataSizeField.setLong(dataManager, cachedMemorySize); + + final Method clearMethod = dataManagerClass.getDeclaredMethod("clear"); + clearMethod.setAccessible(true); + clearMethod.invoke(dataManager); + + final Method getMemoryUsageMethod = + LoadTsFileDataCacheMemoryBlock.class.getDeclaredMethod("getMemoryUsageInBytes"); + getMemoryUsageMethod.setAccessible(true); + Assert.assertEquals(0L, getMemoryUsageMethod.invoke(memoryBlock)); + Assert.assertEquals(0L, dataSizeField.getLong(dataManager)); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java new file mode 100644 index 0000000000000..91785cf69350f --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.storageengine.load.active; + +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.load.util.LoadUtil; + +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.lang.reflect.Method; +import java.nio.file.Files; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class ActiveLoadDirScannerTest { + + private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + private String[] originalListeningDirs; + private String originalPipeDir; + private boolean originalListeningEnabled; + private File tempDir; + private File pendingDir; + private File pipeDir; + + @Before + public void setUp() throws Exception { + originalListeningDirs = config.getLoadActiveListeningDirs(); + originalPipeDir = config.getLoadActiveListeningPipeDir(); + originalListeningEnabled = config.getLoadActiveListeningEnable(); + + tempDir = Files.createTempDirectory("active-load-scanner").toFile(); + pendingDir = new File(tempDir, "pending"); + pipeDir = new File(tempDir, "pipe"); + Assert.assertTrue(pendingDir.mkdirs()); + Assert.assertTrue(pipeDir.mkdirs()); + config.setLoadActiveListeningDirs(new String[] {pendingDir.getAbsolutePath()}); + config.setLoadActiveListeningPipeDir(pipeDir.getAbsolutePath()); + config.setLoadActiveListeningEnable(true); + } + + @After + public void tearDown() { + config.setLoadActiveListeningDirs(originalListeningDirs); + config.setLoadActiveListeningPipeDir(originalPipeDir); + config.setLoadActiveListeningEnable(originalListeningEnabled); + LoadUtil.updateLoadDiskSelector(); + deleteRecursively(tempDir); + } + + @Test + public void testScanDeduplicatesTsFileAndCompanionFiles() throws Exception { + final File tsFile = createCompletedTsFile(pendingDir, "1-0-0-0.tsfile"); + Assert.assertTrue( + new File(tsFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).createNewFile()); + Assert.assertTrue( + new File(tsFile.getAbsolutePath() + ModificationFileV1.FILE_SUFFIX).createNewFile()); + Assert.assertTrue( + new File(tsFile.getAbsolutePath() + ModificationFile.FILE_SUFFIX).createNewFile()); + + final ActiveLoadTsFileLoader loader = mock(ActiveLoadTsFileLoader.class); + when(loader.getCurrentAllowedPendingSize()).thenReturn(10); + final ActiveLoadDirScanner scanner = new ActiveLoadDirScanner(loader); + try { + final Method scanMethod = ActiveLoadDirScanner.class.getDeclaredMethod("scan"); + scanMethod.setAccessible(true); + scanMethod.invoke(scanner); + } finally { + scanner.stop(); + } + + verify(loader, times(1)) + .tryTriggerTsFileLoad( + eq(tsFile.getAbsolutePath()), eq(pendingDir.getAbsolutePath()), eq(false), eq(false)); + } + + @Test + public void testAttributeAndTransferDirectoriesDoNotImplyTableModel() throws Exception { + final Map attributes = + ActiveLoadPathHelper.buildAttributes(null, 2, false, false, null, false, "test-user"); + final File attributeDir = ActiveLoadPathHelper.resolveTargetDir(pendingDir, attributes); + final File transferDir = new File(attributeDir, "transfer-id"); + Assert.assertTrue(transferDir.mkdirs()); + final File tsFile = createCompletedTsFile(transferDir, "2-0-0-0.tsfile"); + + final ActiveLoadTsFileLoader loader = mock(ActiveLoadTsFileLoader.class); + when(loader.getCurrentAllowedPendingSize()).thenReturn(10); + final ActiveLoadDirScanner scanner = new ActiveLoadDirScanner(loader); + try { + final Method scanMethod = ActiveLoadDirScanner.class.getDeclaredMethod("scan"); + scanMethod.setAccessible(true); + scanMethod.invoke(scanner); + } finally { + scanner.stop(); + } + + verify(loader, times(1)) + .tryTriggerTsFileLoad( + eq(tsFile.getAbsolutePath()), eq(pendingDir.getAbsolutePath()), eq(false), eq(false)); + } + + private static File createCompletedTsFile(final File dir, final String fileName) + throws Exception { + final File tsFile = new File(dir, fileName); + try (final TsFileIOWriter writer = new TsFileIOWriter(tsFile)) { + writer.endFile(); + } + return tsFile; + } + + private static void deleteRecursively(final File file) { + if (file == null || !file.exists()) { + return; + } + final File[] children = file.listFiles(); + if (children != null) { + for (final File child : children) { + deleteRecursively(child); + } + } + Assert.assertTrue(file.delete()); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java index 3208dc16ef552..9a7db28b3aebc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java @@ -35,6 +35,7 @@ import org.junit.Test; import java.io.File; +import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.file.Files; @@ -107,6 +108,23 @@ public void testPermanentFailureStatusMovesFileToFailDir() throws Exception { Assert.assertTrue(new File(failDir, tsFile.getName() + ModificationFile.FILE_SUFFIX).exists()); } + @Test + public void testStopClearsPendingFilesForRestart() throws Exception { + final ActiveLoadTsFileLoader loader = new ActiveLoadTsFileLoader(); + final Field pendingQueueField = ActiveLoadTsFileLoader.class.getDeclaredField("pendingQueue"); + pendingQueueField.setAccessible(true); + final ActiveLoadPendingQueue pendingQueue = + (ActiveLoadPendingQueue) pendingQueueField.get(loader); + final String tsFilePath = new File(tempDir, "pending.tsfile").getAbsolutePath(); + Assert.assertTrue(pendingQueue.enqueue(tsFilePath, tempDir.getAbsolutePath(), false, false)); + Assert.assertTrue(loader.isFilePendingOrLoading(new File(tsFilePath))); + + loader.stop(); + + Assert.assertFalse(loader.isFilePendingOrLoading(new File(tsFilePath))); + Assert.assertTrue(pendingQueue.isEmpty()); + } + private File createTsFileWithCompanionFiles(final String fileName) throws Exception { final File tsFile = new File(tempDir, fileName); Assert.assertTrue(tsFile.createNewFile()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java new file mode 100644 index 0000000000000..e9c4e06ee6699 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.storageengine.load.util; + +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.List; + +public class LoadUtilTest { + + private File tempDir; + private File sourceDir; + private File targetDir; + + @Before + public void setUp() throws Exception { + tempDir = Files.createTempDirectory("load-util").toFile(); + sourceDir = new File(tempDir, "source"); + targetDir = new File(tempDir, "target"); + Assert.assertTrue(sourceDir.mkdirs()); + Assert.assertTrue(targetDir.mkdirs()); + } + + @After + public void tearDown() { + deleteRecursively(tempDir); + } + + @Test + public void testTransferFilesKeepsCompanionsTogetherAndDeletesSourcesAfterHandoff() + throws Exception { + final List sourceFiles = createTsFileAndCompanions(); + for (final File sourceFile : sourceFiles) { + Files.write( + new File(targetDir, sourceFile.getName()).toPath(), + "existing".getBytes(StandardCharsets.UTF_8)); + } + + LoadUtil.transferFilesToActiveDir(targetDir, sourceFiles, true); + + final File[] transferDirs = targetDir.listFiles(File::isDirectory); + Assert.assertNotNull(transferDirs); + Assert.assertEquals(1, transferDirs.length); + for (final File sourceFile : sourceFiles) { + Assert.assertFalse(sourceFile.exists()); + final File transferredFile = new File(transferDirs[0], sourceFile.getName()); + Assert.assertTrue(transferredFile.exists()); + Assert.assertArrayEquals( + sourceFile.getName().getBytes(StandardCharsets.UTF_8), + Files.readAllBytes(transferredFile.toPath())); + Assert.assertArrayEquals( + "existing".getBytes(StandardCharsets.UTF_8), + Files.readAllBytes(new File(targetDir, sourceFile.getName()).toPath())); + } + } + + @Test + public void testTransferFailureDoesNotDeleteSources() throws Exception { + final List sourceFiles = createTsFileAndCompanions(); + final File invalidTargetDir = new File(tempDir, "target-file"); + Assert.assertTrue(invalidTargetDir.createNewFile()); + + try { + LoadUtil.transferFilesToActiveDir(invalidTargetDir, sourceFiles, true); + Assert.fail("Expected IOException"); + } catch (final IOException ignored) { + // expected + } + + for (final File sourceFile : sourceFiles) { + Assert.assertTrue(sourceFile.exists()); + } + } + + private List createTsFileAndCompanions() throws Exception { + final File tsFile = new File(sourceDir, "1-0-0-0.tsfile"); + final File resourceFile = new File(tsFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX); + final File modsV1File = new File(tsFile.getAbsolutePath() + ModificationFileV1.FILE_SUFFIX); + final File modsV2File = new File(tsFile.getAbsolutePath() + ModificationFile.FILE_SUFFIX); + final List sourceFiles = Arrays.asList(resourceFile, modsV1File, modsV2File, tsFile); + for (final File sourceFile : sourceFiles) { + Files.write(sourceFile.toPath(), sourceFile.getName().getBytes(StandardCharsets.UTF_8)); + } + return sourceFiles; + } + + private static void deleteRecursively(final File file) { + if (file == null || !file.exists()) { + return; + } + final File[] children = file.listFiles(); + if (children != null) { + for (final File child : children) { + deleteRecursively(child); + } + } + Assert.assertTrue(file.delete()); + } +} From d3caec7f29ae07e487db5984911b217c884b03bd Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:41:42 +0800 Subject: [PATCH 2/2] Test active load file group isolation --- .../apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 59 +++++++++++++++++++ .../planner/node/load/LoadTsFileNodeTest.java | 2 + .../load/LoadTsFileSchedulerTest.java | 2 + .../load/active/ActiveLoadDirScannerTest.java | 4 ++ .../active/ActiveLoadTsFileLoaderTest.java | 2 + .../storageengine/load/util/LoadUtilTest.java | 4 ++ 6 files changed, 73 insertions(+) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java index 1eda0f0622651..8a0800771cf9f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java @@ -20,8 +20,11 @@ package org.apache.iotdb.db.it; import org.apache.iotdb.commons.auth.entity.PrivilegeType; +import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; @@ -985,6 +988,62 @@ public void testLoadWithEmptyTsFile() throws Exception { } } + @Test + public void testAsyncLoadKeepsSameNamedTsFilesWithModsIsolated() throws Exception { + registerSchema(); + + long expectedPointCount = 0; + // Before each file group had its own transfer directory, these same-named TsFiles and mods + // were renamed independently in one shared directory and could be paired with the wrong file. + for (int i = 0; i < 2; i++) { + final File sourceDir = new File(tmpDir, "source-" + i); + Assert.assertTrue(sourceDir.mkdirs()); + try (final TsFileGenerator generator = + new TsFileGenerator(new File(sourceDir, "1-0-0-0.tsfile"))) { + generator.resetRandom(i); + generator.registerTimeseries( + SchemaConfig.DEVICE_0, Collections.singletonList(SchemaConfig.MEASUREMENT_00)); + generator.generateData(SchemaConfig.DEVICE_0, 20, 1, false, TimeUnit.SECONDS.toMillis(i)); + // Each group contributes 20 points and its own mods deletes exactly one. Losing either + // TsFile-to-mods pairing therefore leaves 39 points instead of the expected 38. + expectedPointCount += generator.getTotalNumber() - 1; + } + try (final ModificationFile modificationFile = + new ModificationFile( + new File(sourceDir, "1-0-0-0.tsfile" + ModificationFile.FILE_SUFFIX), false)) { + modificationFile.write( + new TreeDeletionEntry( + new MeasurementPath( + SchemaConfig.DEVICE_0, SchemaConfig.MEASUREMENT_00.getMeasurementName()), + TimeUnit.SECONDS.toMillis(i) + 1, + TimeUnit.SECONDS.toMillis(i) + 1)); + } + } + + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute( + String.format( + "load \"%s\" with ('async'='true','database-level'='2','on-success'='delete')", + tmpDir.getAbsolutePath())); + } + + TestUtils.assertDataEventuallyOnEnv( + EnvFactory.getEnv(), + "select count(" + + SchemaConfig.MEASUREMENT_00.getMeasurementName() + + ") from " + + SchemaConfig.DEVICE_0, + Collections.singletonMap( + "count(" + + SchemaConfig.DEVICE_0 + + "." + + SchemaConfig.MEASUREMENT_00.getMeasurementName() + + ")", + Long.toString(expectedPointCount)), + 30); + } + @Test public void testLoadTsFileWithWrongTimestampPrecision() throws Exception { try (final TsFileGenerator generator = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java index c7fc55a85e230..37cedc50b3430 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java @@ -97,6 +97,8 @@ public void testLoadTsFilePieceNode() { public void testCleanContinuesAfterOneFileCannotBeDeleted() throws Exception { final File tempDir = Files.createTempDirectory("load-node-clean").toFile(); try { + // A non-empty directory at the TsFile path makes that deletion fail deterministically. The + // companion cleanup must still continue instead of sharing the same try-catch block. final File tsFile = new File(tempDir, "1-0-0-0.tsfile"); Assert.assertTrue(tsFile.mkdirs()); Assert.assertTrue(new File(tsFile, "non-empty").createNewFile()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java index 6a6d4868d04ca..1b738f6496d49 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileSchedulerTest.java @@ -142,6 +142,8 @@ public void testTsFileDataManagerClearReleasesCachedMemory() throws Exception { dataManagerConstructor.newInstance( mock(LoadTsFileScheduler.class), mock(LoadSingleTsFileNode.class), memoryBlock); + // Simulate data buffered before split or routing aborts. clear() is the last chance to return + // this accounting to the shared LOAD memory block. final long cachedMemorySize = 128L; memoryBlock.addMemoryUsage(cachedMemorySize); final Field dataSizeField = dataManagerClass.getDeclaredField("dataSize"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java index 91785cf69350f..60532f749d8e0 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScannerTest.java @@ -80,6 +80,8 @@ public void tearDown() { @Test public void testScanDeduplicatesTsFileAndCompanionFiles() throws Exception { + // The recursive scanner sees all four paths, but every companion maps back to the same TsFile. + // Enqueuing each path separately used to consume queue capacity and schedule duplicate loads. final File tsFile = createCompletedTsFile(pendingDir, "1-0-0-0.tsfile"); Assert.assertTrue( new File(tsFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).createNewFile()); @@ -106,6 +108,8 @@ public void testScanDeduplicatesTsFileAndCompanionFiles() throws Exception { @Test public void testAttributeAndTransferDirectoriesDoNotImplyTableModel() throws Exception { + // Async tree loads add attribute and per-handoff transfer directories below pending. These are + // internal directories, not table database names inferred from a user-created subdirectory. final Map attributes = ActiveLoadPathHelper.buildAttributes(null, 2, false, false, null, false, "test-user"); final File attributeDir = ActiveLoadPathHelper.resolveTargetDir(pendingDir, attributes); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java index 9a7db28b3aebc..aa81de4c53b6d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoaderTest.java @@ -119,6 +119,8 @@ public void testStopClearsPendingFilesForRestart() throws Exception { Assert.assertTrue(pendingQueue.enqueue(tsFilePath, tempDir.getAbsolutePath(), false, false)); Assert.assertTrue(loader.isFilePendingOrLoading(new File(tsFilePath))); + // A loader restart reuses this queue. Stale pending membership otherwise makes the scanner + // believe the on-disk TsFile is already scheduled and it will never enqueue it again. loader.stop(); Assert.assertFalse(loader.isFilePendingOrLoading(new File(tsFilePath))); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java index e9c4e06ee6699..542887a0d5014 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/util/LoadUtilTest.java @@ -59,6 +59,8 @@ public void tearDown() { public void testTransferFilesKeepsCompanionsTogetherAndDeletesSourcesAfterHandoff() throws Exception { final List sourceFiles = createTsFileAndCompanions(); + // Existing same-named files represent a previous handoff. A new group must be isolated rather + // than independently renaming the TsFile and companions in this shared target directory. for (final File sourceFile : sourceFiles) { Files.write( new File(targetDir, sourceFile.getName()).toPath(), @@ -86,6 +88,8 @@ public void testTransferFilesKeepsCompanionsTogetherAndDeletesSourcesAfterHandof @Test public void testTransferFailureDoesNotDeleteSources() throws Exception { final List sourceFiles = createTsFileAndCompanions(); + // A regular file cannot contain the temporary transfer directory, forcing handoff to fail + // before ownership of any source file can be released. final File invalidTargetDir = new File(tempDir, "target-file"); Assert.assertTrue(invalidTargetDir.createNewFile());