This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new fd0f76bbcb2 [To dev/1.3] Fix Load task cleanup and input handling
(#18936)
fd0f76bbcb2 is described below
commit fd0f76bbcb22766c7d7797fc968c26f51cb3ead5
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 20 18:39:27 2026 +0800
[To dev/1.3] Fix Load task cleanup and input handling (#18936)
Port the improvements from PR #18772 to the legacy Load implementation.
Adapt the active loader tests and cover rollback while nested Load work remains
active.
---
.../iotdb/db/storageengine/StorageEngine.java | 43 +++---
.../db/storageengine/load/LoadTsFileManager.java | 126 +++++++++++++-----
.../load/LoadTsFilePieceNodeAssembler.java | 2 +-
.../load/active/ActiveLoadDirScanner.java | 10 ++
.../storageengine/load/LoadTsFileManagerTest.java | 145 +++++++++++++++++++++
.../load/LoadTsFilePieceNodeAssemblerTest.java | 12 ++
.../load/active/ActiveLoadDirScannerTest.java | 30 +++++
7 files changed, 312 insertions(+), 56 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index becd7404a56..b3179e8556e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -999,26 +999,31 @@ public class StorageEngine implements IService {
final int sliceIndex,
final int sliceCount,
final int originBodySize) {
- final LoadTsFilePieceNodeAssembler.Result result =
- loadTsFileManager.appendPieceNodeSlice(
- dataRegionId, uuid, body, sliceIndex, sliceCount, originBodySize);
- if (!result.isValid()) {
- return RpcUtils.getStatus(
- TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR,
result.getErrorMessage());
- }
- if (!result.isComplete()) {
- return RpcUtils.SUCCESS_STATUS;
- }
+ return loadTsFileManager.executeLoadTask(
+ uuid,
+ () -> {
+ final LoadTsFilePieceNodeAssembler.Result result =
+ loadTsFileManager.appendPieceNodeSlice(
+ dataRegionId, uuid, body, sliceIndex, sliceCount,
originBodySize);
+ if (!result.isValid()) {
+ return RpcUtils.getStatus(
+ TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR,
result.getErrorMessage());
+ }
+ if (!result.isComplete()) {
+ return RpcUtils.SUCCESS_STATUS;
+ }
- try {
- final Object planNode = PlanNodeType.deserialize(result.getBody());
- if (!(planNode instanceof LoadTsFilePieceNode)) {
- return new
TSStatus(TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR.getStatusCode());
- }
- return writeLoadTsFileNode(dataRegionId, (LoadTsFilePieceNode) planNode,
uuid);
- } catch (final Exception e) {
- return new
TSStatus(TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR.getStatusCode());
- }
+ try {
+ final Object planNode = PlanNodeType.deserialize(result.getBody());
+ if (!(planNode instanceof LoadTsFilePieceNode)) {
+ return new
TSStatus(TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR.getStatusCode());
+ }
+ return writeLoadTsFileNode(dataRegionId, (LoadTsFilePieceNode)
planNode, uuid);
+ } catch (final Exception e) {
+ return new
TSStatus(TSStatusCode.DESERIALIZE_PIECE_OF_TSFILE_ERROR.getStatusCode());
+ }
+ },
+ new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()));
}
public TSStatus executeLoadCommand(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
index abbce7022a3..b99808f2696 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java
@@ -82,12 +82,12 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -163,6 +163,57 @@ public class LoadTsFileManager {
}
}
+ private CleanupTask beginLoadTask(final String uuid) {
+ synchronized (uuid2CleanupTask) {
+ createCleanupTaskIfAbsent(uuid);
+ final CleanupTask cleanupTask = uuid2CleanupTask.get(uuid);
+ if (cleanupTask.isCleanupRequested || cleanupTask.isCanceled) {
+ return null;
+ }
+
+ cleanupTask.runningTaskCount++;
+ cleanupTask.isLoadTaskRunning = true;
+ rescheduleCleanupTask(cleanupTask);
+ return cleanupTask;
+ }
+ }
+
+ private void finishLoadTask(final CleanupTask cleanupTask) {
+ finishLoadTask(cleanupTask, false);
+ }
+
+ private void finishLoadTask(final CleanupTask cleanupTask, final boolean
cleanupAfter) {
+ synchronized (uuid2CleanupTask) {
+ if (uuid2CleanupTask.get(cleanupTask.uuid) != cleanupTask) {
+ return;
+ }
+
+ cleanupTask.isCleanupRequested |= cleanupAfter;
+ cleanupTask.runningTaskCount--;
+ cleanupTask.isLoadTaskRunning = cleanupTask.runningTaskCount > 0;
+
+ if (!cleanupTask.isLoadTaskRunning && cleanupTask.isCleanupRequested) {
+ clean(cleanupTask.uuid);
+ } else {
+ rescheduleCleanupTask(cleanupTask);
+ }
+ }
+ }
+
+ public <T> T executeLoadTask(
+ final String uuid, final Supplier<T> operation, final T rejectedResult) {
+ final CleanupTask cleanupTask = beginLoadTask(uuid);
+ if (cleanupTask == null) {
+ return rejectedResult;
+ }
+
+ try {
+ return operation.get();
+ } finally {
+ finishLoadTask(cleanupTask);
+ }
+ }
+
private void rescheduleCleanupTask(final CleanupTask cleanupTask) {
synchronized (uuid2CleanupTask) {
if (uuid2CleanupTask.get(cleanupTask.uuid) != cleanupTask) {
@@ -252,10 +303,10 @@ public class LoadTsFileManager {
public void writeToDataRegion(DataRegion dataRegion, LoadTsFilePieceNode
pieceNode, String uuid)
throws IOException, PageException {
- createCleanupTaskIfAbsent(uuid);
-
- final Optional<CleanupTask> cleanupTask =
Optional.ofNullable(uuid2CleanupTask.get(uuid));
- cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);
+ final CleanupTask cleanupTask = beginLoadTask(uuid);
+ if (cleanupTask == null) {
+ throw new
IOException(String.format(MESSAGE_WRITER_MANAGER_HAS_BEEN_CLOSED, uuid));
+ }
try {
final AtomicReference<Exception> exception = new AtomicReference<>();
final TsFileWriterManager writerManager =
@@ -287,7 +338,7 @@ public class LoadTsFileManager {
}
}
} finally {
- cleanupTask.ifPresent(CleanupTask::markLoadTaskNotRunning);
+ finishLoadTask(cleanupTask);
}
}
@@ -298,10 +349,11 @@ public class LoadTsFileManager {
final int sliceIndex,
final int sliceCount,
final int originBodySize) {
- createCleanupTaskIfAbsent(uuid);
-
- final Optional<CleanupTask> cleanupTask =
Optional.ofNullable(uuid2CleanupTask.get(uuid));
- cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);
+ final CleanupTask cleanupTask = beginLoadTask(uuid);
+ if (cleanupTask == null) {
+ return LoadTsFilePieceNodeAssembler.Result.invalid(
+ String.format(MESSAGE_WRITER_MANAGER_HAS_BEEN_CLOSED, uuid));
+ }
try {
final Map<DataRegionId, LoadTsFilePieceNodeAssembler> regionId2Assembler
=
uuid2PieceNodeAssembler.computeIfAbsent(uuid, key -> new
ConcurrentHashMap<>());
@@ -330,7 +382,7 @@ public class LoadTsFileManager {
return result;
}
} finally {
- cleanupTask.ifPresent(CleanupTask::markLoadTaskNotRunning);
+ finishLoadTask(cleanupTask);
}
}
@@ -373,32 +425,42 @@ public class LoadTsFileManager {
boolean isGeneratedByPipe,
Map<TTimePartitionSlot, ProgressIndex> timePartitionProgressIndexMap)
throws IOException, LoadFileException {
- if (!uuid2WriterManager.containsKey(uuid) ||
uuid2PieceNodeAssembler.containsKey(uuid)) {
+ final CleanupTask cleanupTask = beginLoadTask(uuid);
+ if (cleanupTask == null) {
return false;
}
- createCleanupTaskIfAbsent(uuid);
-
- final Optional<CleanupTask> cleanupTask =
Optional.ofNullable(uuid2CleanupTask.get(uuid));
- cleanupTask.ifPresent(CleanupTask::markLoadTaskRunning);
+ boolean loaded = false;
try {
+ if (!uuid2WriterManager.containsKey(uuid) ||
uuid2PieceNodeAssembler.containsKey(uuid)) {
+ return false;
+ }
uuid2WriterManager.get(uuid).loadAll(isGeneratedByPipe,
timePartitionProgressIndexMap);
+ loaded = true;
+ return true;
} finally {
- cleanupTask.ifPresent(CleanupTask::markLoadTaskNotRunning);
+ finishLoadTask(cleanupTask, loaded);
}
-
- clean(uuid);
- return true;
}
public boolean deleteAll(String uuid) {
- if (!uuid2WriterManager.containsKey(uuid)
- && !uuid2PieceNodeAssembler.containsKey(uuid)
- && !uuid2CleanupTask.containsKey(uuid)) {
- return false;
+ synchronized (uuid2CleanupTask) {
+ if (!uuid2WriterManager.containsKey(uuid)
+ && !uuid2PieceNodeAssembler.containsKey(uuid)
+ && !uuid2CleanupTask.containsKey(uuid)) {
+ return false;
+ }
+
+ final CleanupTask cleanupTask = uuid2CleanupTask.get(uuid);
+ if (cleanupTask != null && cleanupTask.isLoadTaskRunning) {
+ cleanupTask.isCleanupRequested = true;
+ rescheduleCleanupTask(cleanupTask);
+ return true;
+ }
+
+ clean(uuid);
+ return true;
}
- clean(uuid);
- return true;
}
private void clean(String uuid) {
@@ -824,6 +886,8 @@ public class LoadTsFileManager {
private long scheduledTime;
private volatile boolean isLoadTaskRunning = false;
+ private int runningTaskCount;
+ private boolean isCleanupRequested;
private volatile boolean isCanceled = false;
private CleanupTask(String uuid, long delayInMs) {
@@ -832,16 +896,6 @@ public class LoadTsFileManager {
resetScheduledTime();
}
- public void markLoadTaskRunning() {
- isLoadTaskRunning = true;
- rescheduleCleanupTask(this);
- }
-
- public void markLoadTaskNotRunning() {
- isLoadTaskRunning = false;
- rescheduleCleanupTask(this);
- }
-
public void resetScheduledTime() {
scheduledTime = System.currentTimeMillis() + delayInMs;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssembler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssembler.java
index ccf9d77ac8f..b35ecafd20f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssembler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssembler.java
@@ -34,7 +34,7 @@ public final class LoadTsFilePieceNodeAssembler {
LoadTsFilePieceNodeAssembler(final int sliceCount, final int originBodySize)
{
this.sliceCount = sliceCount;
this.originBodySize = originBodySize;
- this.assembledBody = new PublicBAOS(Math.max(0, originBodySize));
+ this.assembledBody = new PublicBAOS();
}
synchronized Result append(
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 c07c0d8e61a..68735c9e386 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,6 +108,7 @@ public class ActiveLoadDirScanner extends
ActiveLoadScheduledExecutorService {
FileUtils.streamFiles(listeningDirFile, true, (String[]) null)) {
try {
fileStream
+ .filter(file -> !isUnderFailDir(file))
.filter(file ->
!ActiveLoadPathHelper.isTransferStagingFile(file, listeningDirFile))
.map(file -> new File(getTsFilePath(file.getAbsolutePath())))
.distinct()
@@ -142,6 +143,15 @@ public class ActiveLoadDirScanner extends
ActiveLoadScheduledExecutorService {
}
}
+ private boolean isUnderFailDir(final File file) {
+ final String failDir = IOTDB_CONFIG.getLoadActiveListeningFailDir();
+ if (failDir == null || failDir.isEmpty()) {
+ return false;
+ }
+ final Path failDirPath = new
File(failDir).toPath().toAbsolutePath().normalize();
+ return file.toPath().toAbsolutePath().normalize().startsWith(failDirPath);
+ }
+
private boolean checkPermission(final String listeningDir) {
try {
final Path listeningDirPath = new File(listeningDir).toPath();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
new file mode 100644
index 00000000000..9a2a9e6ee9c
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManagerTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+public class LoadTsFileManagerTest {
+
+ private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private String[][] originalTierDataDirs;
+ private File tempDir;
+ private LoadTsFileManager loadTsFileManager;
+
+ @Before
+ public void setUp() throws Exception {
+ originalTierDataDirs = copyDirs(config.getTierDataDirs());
+ tempDir = Files.createTempDirectory("load-tsfile-manager").toFile();
+ config.setTierDataDirs(new String[][] {{tempDir.getAbsolutePath()}});
+ loadTsFileManager = new LoadTsFileManager();
+ }
+
+ @After
+ public void tearDown() {
+ if (loadTsFileManager != null) {
+ loadTsFileManager.stop();
+ }
+ config.setTierDataDirs(originalTierDataDirs);
+ deleteRecursively(tempDir);
+ }
+
+ @Test
+ public void testRollbackDefersCleanupUntilActiveLoadTaskFinishes() throws
Exception {
+ final String uuid = "rollback-test";
+ final CountDownLatch taskStarted = new CountDownLatch(1);
+ final CountDownLatch finishTask = new CountDownLatch(1);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ final Future<String> activeTask =
+ executor.submit(
+ () ->
+ loadTsFileManager.executeLoadTask(
+ uuid,
+ () -> {
+ taskStarted.countDown();
+ try {
+ if (!finishTask.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting to
finish the load task");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ return "finished";
+ },
+ "rejected"));
+
+ Assert.assertTrue(taskStarted.await(10, TimeUnit.SECONDS));
+ Assert.assertTrue(loadTsFileManager.deleteAll(uuid));
+ Assert.assertEquals(
+ "rejected", loadTsFileManager.executeLoadTask(uuid, () ->
"unexpected", "rejected"));
+
+ finishTask.countDown();
+ Assert.assertEquals("finished", activeTask.get(10, TimeUnit.SECONDS));
+ Assert.assertFalse(loadTsFileManager.deleteAll(uuid));
+ } finally {
+ finishTask.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testRollbackWaitsForOuterLoadTaskAfterNestedTaskFinishes() {
+ final String uuid = "nested-rollback-test";
+ Assert.assertEquals(
+ "finished",
+ loadTsFileManager.executeLoadTask(
+ uuid,
+ () -> {
+ Assert.assertEquals(
+ "nested",
+ loadTsFileManager.executeLoadTask(
+ uuid,
+ () -> {
+ Assert.assertTrue(loadTsFileManager.deleteAll(uuid));
+ return "nested";
+ },
+ "rejected"));
+ Assert.assertEquals(
+ "rejected",
+ loadTsFileManager.executeLoadTask(uuid, () -> "unexpected",
"rejected"));
+ return "finished";
+ },
+ "rejected"));
+ Assert.assertFalse(loadTsFileManager.deleteAll(uuid));
+ }
+
+ private static String[][] copyDirs(final String[][] dirs) {
+ return Arrays.stream(dirs).map(String[]::clone).toArray(String[][]::new);
+ }
+
+ 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/LoadTsFilePieceNodeAssemblerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssemblerTest.java
index 79c572a14ed..74ca74cdb11 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssemblerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePieceNodeAssemblerTest.java
@@ -133,6 +133,18 @@ public class LoadTsFilePieceNodeAssemblerTest {
}
}
+ @Test
+ public void testLargeDeclaredBodySizeDoesNotPreallocateTheWholeBody() {
+ final LoadTsFilePieceNodeAssembler assembler =
+ new LoadTsFilePieceNodeAssembler(2, Integer.MAX_VALUE);
+
+ final LoadTsFilePieceNodeAssembler.Result result =
+ assembler.append(ByteBuffer.wrap(new byte[] {1}), 0, 2,
Integer.MAX_VALUE);
+
+ Assert.assertTrue(result.isValid());
+ Assert.assertFalse(result.isComplete());
+ }
+
@Test
public void testRejectBodySizeViolationsWithReason() {
final LoadTsFilePieceNodeAssembler.Result overflow =
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 5703ee0ec55..1eb0594eb33 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
@@ -45,6 +45,7 @@ public class ActiveLoadDirScannerTest {
private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private String[] originalListeningDirs;
private String originalPipeDir;
+ private String originalFailDir;
private boolean originalListeningEnabled;
private File tempDir;
private File pendingDir;
@@ -54,6 +55,7 @@ public class ActiveLoadDirScannerTest {
public void setUp() throws Exception {
originalListeningDirs = config.getLoadActiveListeningDirs();
originalPipeDir = config.getLoadActiveListeningPipeDir();
+ originalFailDir = config.getLoadActiveListeningFailDir();
originalListeningEnabled = config.getLoadActiveListeningEnable();
tempDir = Files.createTempDirectory("active-load-scanner").toFile();
@@ -70,6 +72,7 @@ public class ActiveLoadDirScannerTest {
public void tearDown() {
config.setLoadActiveListeningDirs(originalListeningDirs);
config.setLoadActiveListeningPipeDir(originalPipeDir);
+ config.setLoadActiveListeningFailDir(originalFailDir);
config.setLoadActiveListeningEnable(originalListeningEnabled);
ActiveLoadUtil.updateLoadDiskSelector();
deleteRecursively(tempDir);
@@ -121,6 +124,33 @@ public class ActiveLoadDirScannerTest {
eq(tsFile.getAbsolutePath()), eq(pendingDir.getAbsolutePath()),
eq(false));
}
+ @Test
+ public void testScanIgnoresFailDirectoryNestedUnderListeningDirectory()
throws Exception {
+ final File failDir = new File(pendingDir, "failed");
+ Assert.assertTrue(failDir.mkdirs());
+ config.setLoadActiveListeningFailDir(failDir.getAbsolutePath());
+ final File failedTsFile = createCompletedTsFile(failDir, "failed.tsfile");
+ final File pendingTsFile = createCompletedTsFile(pendingDir,
"pending.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(pendingTsFile.getAbsolutePath()),
eq(pendingDir.getAbsolutePath()), eq(false));
+ verify(loader, times(0))
+ .tryTriggerTsFileLoad(
+ eq(failedTsFile.getAbsolutePath()),
eq(pendingDir.getAbsolutePath()), eq(false));
+ }
+
private static File createCompletedTsFile(final File dir, final String
fileName)
throws Exception {
final File tsFile = new File(dir, fileName);
