This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 4092a780097 Fix pipe TsFile transfer when mod file is missing (#18756)
4092a780097 is described below
commit 4092a78009765b2b2a9b92c663f140f7f05ec799
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 29 15:55:20 2026 +0800
Fix pipe TsFile transfer when mod file is missing (#18756)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 ++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 ++
.../thrift/async/IoTDBDataRegionAsyncSink.java | 106 ++++++++++++++++++++-
.../async/handler/PipeTransferTsFileHandler.java | 100 ++++++++++++++++---
.../thrift/async/IoTDBDataRegionAsyncSinkTest.java | 49 ++++++++++
.../PipeTransferTsFileHandlerCleanupTest.java | 56 +++++++++++
6 files changed, 307 insertions(+), 16 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 8fd63338892..1c0362b627b 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -820,10 +820,16 @@ public final class DataNodePipeMessages {
"Failed to transfer tsfile batch ({}).";
public static final String FAILED_TO_TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY =
"Failed to transfer tsfile event {} asynchronously.";
+ public static final String
+
EXCEPTION_FAILED_TO_TRANSFER_TSFILE_ARG_BECAUSE_FILE_ARG_IS_MISSING_AFTER_ARG_RETRIES_B656E84E
=
+ "Failed to transfer TsFile {} because file {} is missing after {}
retries.";
public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID =
"Failed to transfer TsFileInsertionEvent {} (committer key {}, commit id
{}).";
public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_BATCHED_TABLE_EVENTS =
"Failed to transfer TsFileInsertionEvent {} (batched
TableInsertionEvents).";
+ public static final String
+
LOG_TSFILE_ARG_MODIFICATION_FILE_ARG_IS_MISSING_TRANSFER_THE_TSFILE_WITHOUT_MODIFICATIONS_8588D1E9
=
+ "TsFile {}: modification file {} is missing, transfer the TsFile
without modifications.";
public static final String FAILED_TO_UPDATE_LEADER_CACHE_FOR_DEVICE =
"Failed to update leader cache for device {} with endpoint {}:{}.";
public static final String FAILED_TO_WRITE = "Failed to write ";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 62918788834..49096659ade 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -774,10 +774,16 @@ public final class DataNodePipeMessages {
public static final String FAILED_TO_TRANSFER_TSFILE_BATCH = "传输 tsfile
batch ({}) 失败。";
public static final String FAILED_TO_TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY =
"传输 tsfile event {} asynchronously 失败。";
+ public static final String
+
EXCEPTION_FAILED_TO_TRANSFER_TSFILE_ARG_BECAUSE_FILE_ARG_IS_MISSING_AFTER_ARG_RETRIES_B656E84E
=
+ "TsFile {} 传输失败，文件 {} 重试 {} 次后仍不存在。";
public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID =
"传输 TsFileInsertionEvent {}（committer key {}，commit id {}）失败。";
public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_BATCHED_TABLE_EVENTS =
"传输 TsFileInsertionEvent {}（批量 TableInsertionEvent）失败。";
+ public static final String
+
LOG_TSFILE_ARG_MODIFICATION_FILE_ARG_IS_MISSING_TRANSFER_THE_TSFILE_WITHOUT_MODIFICATIONS_8588D1E9
=
+ "TsFile {}：修改文件 {} 不存在，将不带修改记录传输 TsFile。";
public static final String FAILED_TO_UPDATE_LEADER_CACHE_FOR_DEVICE =
"更新 leader cache for device {} with endpoint {}:{} 失败。";
public static final String FAILED_TO_WRITE = "写入失败 ";
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index 4b7f332dbe3..64f25919750 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.client.ThriftClient;
import
org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
@@ -82,6 +83,7 @@ import
org.apache.tsfile.exception.write.WriteProcessException;
import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.slf4j.helpers.MessageFormatter;
import java.io.File;
import java.io.FileNotFoundException;
@@ -132,6 +134,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
"Exception occurred while sending to receiver %s:%s.";
private static final boolean isSplitTSFileBatchModeEnabled = true;
+ private static final int MAX_FILE_NOT_FOUND_RETRY_TIMES = 5;
private final IoTDBDataRegionSyncSink syncSink = new
IoTDBDataRegionSyncSink();
@@ -143,6 +146,10 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
// semantics because the same payload may compare equal.
private final Map<Event, PipeResourceFailureType>
retryEvent2ResourceFailureType =
new IdentityHashMap<>();
+ // File disappearance is permanent for a given TsFile event. Keep its retry
count separate from
+ // transient receiver failures so one missing file cannot retry forever in
the async queue.
+ private final Map<Event, Integer> missingFileRetryTimes = new
IdentityHashMap<>();
+ private volatile PipeRuntimeSinkCriticalException
missingFileRetryLimitException;
// Keep only the latest text to avoid retaining the complete exception chain
for every event.
private volatile String lastRetryFailureMessage;
// Events removed from the retry queue remain in this map while their next
transfer is being
@@ -576,6 +583,22 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
throw new
FileNotFoundException(pipeTsFileInsertionEvent.getTsFile().getAbsolutePath());
}
+ final boolean transferMod =
+ pipeTsFileInsertionEvent.isWithMod()
+ && clientManager.supportModsIfIsDataNodeReceiver()
+ && pipeTsFileInsertionEvent.getModFile() != null
+ && pipeTsFileInsertionEvent.getModFile().exists();
+ if (pipeTsFileInsertionEvent.isWithMod()
+ && clientManager.supportModsIfIsDataNodeReceiver()
+ && pipeTsFileInsertionEvent.getModFile() != null
+ && !pipeTsFileInsertionEvent.getModFile().exists()) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_TSFILE_ARG_MODIFICATION_FILE_ARG_IS_MISSING_TRANSFER_THE_TSFILE_WITHOUT_MODIFICATIONS_8588D1E9,
+ pipeTsFileInsertionEvent.getTsFile(),
+ pipeTsFileInsertionEvent.getModFile());
+ }
+
pipeTransferTsFileHandler =
new PipeTransferTsFileHandler(
this,
@@ -589,8 +612,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
new AtomicBoolean(false),
pipeTsFileInsertionEvent.getTsFile(),
pipeTsFileInsertionEvent.getModFile(),
- pipeTsFileInsertionEvent.isWithMod()
- && clientManager.supportModsIfIsDataNodeReceiver(),
+ transferMod,
pipeTsFileInsertionEvent.isTableModelEvent()
? pipeTsFileInsertionEvent.getTableModelDatabaseName()
: pipeTsFileInsertionEvent.getTreeModelDatabaseName());
@@ -775,6 +797,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
* @see PipeConnector#transfer(TsFileInsertionEvent) for more details.
*/
private void transferQueuedEventsIfNecessary(final boolean forced) {
+ throwIfMissingFileRetryLimitExceeded();
throwIfReceiverProbeIsDelayed();
if ((retryEventQueue.isEmpty() && retryTsFileQueue.isEmpty())
@@ -838,6 +861,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
}
}
+ throwIfMissingFileRetryLimitExceeded();
throwIfReceiverProbeIsDelayed();
// Stop retrying if the execution time exceeds the threshold for better
realtime performance
@@ -985,10 +1009,12 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
if (enrichedEvent.isReleased()) {
retryingEvent2Handlers.remove(event);
+ missingFileRetryTimes.remove(event);
return;
}
if (isDroppedPipe(enrichedEvent)) {
retryingEvent2Handlers.remove(event);
+ missingFileRetryTimes.remove(event);
enrichedEvent.clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
return;
}
@@ -996,6 +1022,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
if (isClosed.get()) {
retryingEvent2Handlers.remove(event);
+ missingFileRetryTimes.remove(event);
if (event instanceof EnrichedEvent) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
}
@@ -1030,6 +1057,14 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
final PipeResourceFailureType previousResourceFailureType =
retryEvent2ResourceFailureType.get(event);
+ if (!alreadyInRetryQueue && isMissingFileFailure(event, e)) {
+ final int failureCount = missingFileRetryTimes.merge(event, 1,
Integer::sum);
+ if (failureCount > MAX_FILE_NOT_FOUND_RETRY_TIMES) {
+ reportMissingFileEventAfterRetryLimit(event, e, failureCount - 1);
+ return;
+ }
+ }
+
if (resourceFailureType != null
&& event instanceof EnrichedEvent
&& (!alreadyInRetryQueue || previousResourceFailureType !=
resourceFailureType)) {
@@ -1070,6 +1105,61 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
}
}
+ private static boolean isMissingFileFailure(final Event event, final
Exception exception) {
+ if (!(event instanceof PipeTsFileInsertionEvent) || exception == null) {
+ return false;
+ }
+ return ErrorHandlingCommonUtils.getRootCause(exception) instanceof
FileNotFoundException;
+ }
+
+ private void reportMissingFileEventAfterRetryLimit(
+ final Event event, final Exception cause, final int retryTimes) {
+ final String missingFile = getMissingFile(cause);
+ final File tsFile = ((PipeTsFileInsertionEvent) event).getTsFile();
+ final String message =
+ MessageFormatter.arrayFormat(
+ DataNodePipeMessages
+
.EXCEPTION_FAILED_TO_TRANSFER_TSFILE_ARG_BECAUSE_FILE_ARG_IS_MISSING_AFTER_ARG_RETRIES_B656E84E,
+ new Object[] {tsFile, missingFile, retryTimes})
+ .getMessage();
+ final PipeRuntimeSinkCriticalException criticalException =
+ new PipeRuntimeSinkCriticalException(message, cause);
+
+ retryingEvent2Handlers.remove(event);
+ retryEvent2ResourceFailureType.put(event, null);
+ missingFileRetryTimes.remove(event);
+ lastRetryFailureMessage = message;
+ missingFileRetryLimitException = criticalException;
+ retryTsFileQueue.offer((PipeTsFileInsertionEvent) event);
+ retryEventQueueEventCounter.increaseEventCount(event);
+ if (event instanceof EnrichedEvent) {
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ PipeLogger.log(
+ LOGGER::error,
+ criticalException,
+ DataNodePipeMessages
+
.EXCEPTION_FAILED_TO_TRANSFER_TSFILE_ARG_BECAUSE_FILE_ARG_IS_MISSING_AFTER_ARG_RETRIES_B656E84E,
+ tsFile,
+ missingFile,
+ retryTimes);
+ PipeDataNodeAgent.runtime().report(enrichedEvent, criticalException);
+ } else {
+ LOGGER.error(message, cause);
+ }
+ }
+
+ private void throwIfMissingFileRetryLimitExceeded() {
+ final PipeRuntimeSinkCriticalException exception =
missingFileRetryLimitException;
+ if (exception != null) {
+ throw exception;
+ }
+ }
+
+ private static String getMissingFile(final Exception exception) {
+ final Throwable rootCause =
ErrorHandlingCommonUtils.getRootCause(exception);
+ return hasText(rootCause.getMessage()) ? rootCause.getMessage() :
rootCause.toString();
+ }
+
/**
* Add failure {@link EnrichedEvent}s to retry queue.
*
@@ -1121,6 +1211,12 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
return lastRetryFailureMessage;
}
+ public synchronized void clearFileNotFoundRetryTimes(final Iterable<?
extends Event> events) {
+ for (final Event event : events) {
+ missingFileRetryTimes.remove(event);
+ }
+ }
+
public void addFailureEventsToRetryQueue(
final Iterable<EnrichedEvent> events,
final Exception e,
@@ -1289,6 +1385,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
retryEvent2ResourceFailureType.remove(event);
+ missingFileRetryTimes.remove(event);
retryingEvent2Handlers.remove(event);
return true;
}
@@ -1302,6 +1399,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
retryEvent2ResourceFailureType.remove(event);
+ missingFileRetryTimes.remove(event);
retryingEvent2Handlers.remove(event);
return true;
}
@@ -1315,6 +1413,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
&& isDroppedPipe((EnrichedEvent) event, committerKey));
if (retryEventQueue.isEmpty() && retryTsFileQueue.isEmpty()) {
lastRetryFailureMessage = null;
+ missingFileRetryLimitException = null;
}
}
@@ -1397,12 +1496,15 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
retryTsFileQueue.isEmpty() ? retryEventQueue.poll() :
retryTsFileQueue.poll();
retryEventQueueEventCounter.decreaseEventCount(event);
retryEvent2ResourceFailureType.remove(event);
+ missingFileRetryTimes.remove(event);
retryingEvent2Handlers.remove(event);
if (event instanceof EnrichedEvent) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
}
}
retryEvent2ResourceFailureType.clear();
+ missingFileRetryTimes.clear();
+ missingFileRetryLimitException = null;
lastRetryFailureMessage = null;
retryingEvent2Handlers.clear();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
index 8bf1a1cede2..327cc570b00 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
@@ -51,6 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.Arrays;
@@ -79,14 +80,20 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
private File currentFile;
private final boolean transferMod;
+ // A mod file can be cleaned up after the event is created. Keep the
requested mode immutable,
+ // but downgrade the requests sent by this handler when the sidecar is no
longer available.
+ private boolean effectiveTransferMod;
private final String dataBaseName;
- private final String conversionTaskId;
+ private String conversionTaskId;
+ private final int outputIndex;
private final int readFileBufferSize;
private PipeTsFileMemoryBlock memoryBlock;
private byte[] readBuffer;
private long position;
+ private boolean transferringModFile;
+ private boolean transferringTsFile;
private RandomAccessFile reader;
@@ -141,13 +148,17 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
this.tsFile = tsFile;
this.modFile = modFile;
this.transferMod = transferMod;
+ this.effectiveTransferMod = transferMod && modFile != null &&
modFile.exists();
this.dataBaseName = dataBaseName;
+ this.outputIndex = outputIndex;
+ this.transferringModFile = false;
+ this.transferringTsFile = false;
conversionTaskId =
connector.shouldAsyncLoadTsFileOnTypeMismatch()
? PipeTransferTsFileSealWithModReq.generateConversionTaskId(
- connector.getSinkTaskId(), events, dataBaseName, outputIndex,
transferMod)
+ connector.getSinkTaskId(), events, dataBaseName, outputIndex,
effectiveTransferMod)
: null;
- currentFile = transferMod ? modFile : tsFile;
+ currentFile = effectiveTransferMod ? modFile : tsFile;
// NOTE: Waiting for resource enough for slicing here may cause deadlock!
// TsFile events are producing and consuming at the same time, and the
memory of a TsFile
@@ -157,7 +168,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
// will cause a deadlock.
waitForResourceEnough4Slicing((long) ((1 + Math.random()) * 20 * 1000));
// 20 - 40 seconds
final long maxFileLength =
- transferMod && Objects.nonNull(modFile)
+ effectiveTransferMod && Objects.nonNull(modFile)
? Math.max(tsFile.length(), modFile.length())
: tsFile.length();
readFileBufferSize =
@@ -204,7 +215,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
}
if (reader == null) {
- reader = transferMod ? new RandomAccessFile(modFile, "r") : new
RandomAccessFile(tsFile, "r");
+ openNextFile();
}
client.setShouldReturnSelf(false);
@@ -213,18 +224,15 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
final int readLength = readNextFilePiece(reader, readBuffer);
if (readLength == -1) {
- if (currentFile == modFile) {
- currentFile = tsFile;
- position = 0;
- reader.close();
- reader = null;
- reader = new RandomAccessFile(tsFile, "r");
+ reader.close();
+ reader = null;
+ if (openNextFile()) {
transfer(clientManager, client);
- } else if (currentFile == tsFile) {
+ } else {
isSealSignalSent.set(true);
final TPipeTransferReq uncompressedReq =
- transferMod
+ effectiveTransferMod
? PipeTransferTsFileSealWithModReq.toTPipeTransferReq(
modFile.getName(),
modFile.length(),
@@ -263,7 +271,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
? readBuffer
: Arrays.copyOfRange(readBuffer, 0, readLength);
final TPipeTransferReq uncompressedReq =
- transferMod
+ effectiveTransferMod
? PipeTransferTsFilePieceWithModReq.toTPipeTransferReq(
currentFile.getName(), position, payload)
: PipeTransferTsFilePieceReq.toTPipeTransferReq(
@@ -338,6 +346,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
try {
final int referenceCount = eventsReferenceCount.decrementAndGet();
if (referenceCount <= 0) {
+ sink.clearFileNotFoundRetryTimes(events);
events.forEach(
event ->
event.decreaseReferenceCount(PipeTransferTsFileHandler.class.getName(), true));
@@ -497,6 +506,69 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
events.forEach(event ->
event.clearReferenceCount(PipeTransferTsFileHandler.class.getName()));
}
+ private boolean advanceToNextFile() {
+ if (transferMod && !transferringModFile) {
+ transferringModFile = true;
+ if (modFile != null && modFile.exists()) {
+ effectiveTransferMod = true;
+ conversionTaskId = generateConversionTaskId();
+ currentFile = modFile;
+ position = 0;
+ return true;
+ }
+
+ // The mod file is an optional sidecar. It may have been cleaned up
after the event was
+ // created, so transfer the TsFile without modifications instead of
creating a WithMod
+ // request that the receiver can never finish.
+ effectiveTransferMod = false;
+ conversionTaskId = generateConversionTaskId();
+ if (modFile != null) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_TSFILE_ARG_MODIFICATION_FILE_ARG_IS_MISSING_TRANSFER_THE_TSFILE_WITHOUT_MODIFICATIONS_8588D1E9,
+ tsFile,
+ modFile);
+ }
+ }
+ if (!transferringTsFile) {
+ transferringTsFile = true;
+ currentFile = tsFile;
+ position = 0;
+ return true;
+ }
+ return false;
+ }
+
+ private boolean openNextFile() throws IOException {
+ while (advanceToNextFile()) {
+ try {
+ reader = new RandomAccessFile(currentFile, "r");
+ return true;
+ } catch (final FileNotFoundException e) {
+ if (currentFile == modFile && effectiveTransferMod) {
+ // Re-check after advanceToNextFile(). The sidecar can disappear in
this small window.
+ effectiveTransferMod = false;
+ conversionTaskId = generateConversionTaskId();
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_TSFILE_ARG_MODIFICATION_FILE_ARG_IS_MISSING_TRANSFER_THE_TSFILE_WITHOUT_MODIFICATIONS_8588D1E9,
+ tsFile,
+ modFile);
+ continue;
+ }
+ throw e;
+ }
+ }
+ return false;
+ }
+
+ private String generateConversionTaskId() {
+ return sink.shouldAsyncLoadTsFileOnTypeMismatch()
+ ? PipeTransferTsFileSealWithModReq.generateConversionTaskId(
+ sink.getSinkTaskId(), events, dataBaseName, outputIndex,
effectiveTransferMod)
+ : null;
+ }
+
@Override
public synchronized void close() {
try {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSinkTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSinkTest.java
index cd0213bc735..fb924b31e30 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSinkTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSinkTest.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.pipe.sink.protocol.thrift.async;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
+import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -26,6 +28,11 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.lang.reflect.Field;
+import java.util.Map;
+
public class IoTDBDataRegionAsyncSinkTest {
@Test
@@ -63,4 +70,46 @@ public class IoTDBDataRegionAsyncSinkTest {
Assert.assertEquals("java.lang.NullPointerException",
sink.getLastRetryFailureMessage());
}
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testMissingTsFileRetryLimitKeepsEventQueuedAndStopsSink() throws
Exception {
+ final IoTDBDataRegionAsyncSink sink = new IoTDBDataRegionAsyncSink();
+ final PipeTsFileInsertionEvent event =
Mockito.mock(PipeTsFileInsertionEvent.class);
+ final File missingTsFile = new
File("missing-retry-limit.tsfile").getAbsoluteFile();
+ Mockito.when(event.getTsFile()).thenReturn(missingTsFile);
+
+ final Field retryLimitField =
+
IoTDBDataRegionAsyncSink.class.getDeclaredField("MAX_FILE_NOT_FOUND_RETRY_TIMES");
+ retryLimitField.setAccessible(true);
+ final int retryLimit = retryLimitField.getInt(null);
+
+ final Field retryTimesField =
+
IoTDBDataRegionAsyncSink.class.getDeclaredField("missingFileRetryTimes");
+ retryTimesField.setAccessible(true);
+ ((Map<Event, Integer>) retryTimesField.get(sink)).put(event, retryLimit);
+
+ try {
+ sink.addFailureEventToRetryQueue(
+ event, new FileNotFoundException(missingTsFile.getAbsolutePath()));
+
+ Assert.assertEquals(1, sink.getRetryEventQueueSize());
+ Assert.assertTrue(
+
sink.getLastRetryFailureMessage().contains(missingTsFile.getAbsolutePath()));
+
Assert.assertTrue(sink.getLastRetryFailureMessage().contains(String.valueOf(retryLimit)));
+ Mockito.verify(event, Mockito.never())
+ .clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
+
+ try {
+ sink.transfer(Mockito.mock(Event.class));
+ Assert.fail("The sink should stop transferring after the missing-file
retry limit.");
+ } catch (final PipeRuntimeSinkCriticalException e) {
+
Assert.assertTrue(e.getMessage().contains(missingTsFile.getAbsolutePath()));
+ }
+ } finally {
+ sink.clearRetryEventsReferenceCount();
+ }
+
+
Mockito.verify(event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java
index 3cca3285cca..563961d983d 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandlerCleanupTest.java
@@ -36,6 +36,7 @@ import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.nio.file.Files;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -95,6 +96,47 @@ public class PipeTransferTsFileHandlerCleanupTest {
}
}
+ @Test
+ public void testModFileDisappearingBeforeOpenDowngradesToTsFileOnly() throws
Exception {
+ final File tsFile = Files.createTempFile("pipe-transfer-source",
".tsfile").toFile();
+ final File modFile =
+ new DisappearingFile(
+ Files.createTempDirectory("pipe-transfer-missing-mod")
+ .resolve("source.tsfile.mods2")
+ .toString());
+ final PipeTsFileInsertionEvent event =
Mockito.mock(PipeTsFileInsertionEvent.class);
+ final PipeTransferTsFileHandler handler =
+ new PipeTransferTsFileHandler(
+ Mockito.mock(IoTDBDataRegionAsyncSink.class),
+ Collections.emptyMap(),
+ Collections.singletonList(event),
+ new AtomicInteger(1),
+ new AtomicBoolean(false),
+ tsFile,
+ modFile,
+ true,
+ null);
+
+ try {
+ final Method openNextFile =
PipeTransferTsFileHandler.class.getDeclaredMethod("openNextFile");
+ openNextFile.setAccessible(true);
+ Assert.assertTrue((Boolean) openNextFile.invoke(handler));
+
+ final Field currentFile =
PipeTransferTsFileHandler.class.getDeclaredField("currentFile");
+ currentFile.setAccessible(true);
+ Assert.assertSame(tsFile, currentFile.get(handler));
+
+ final Field effectiveTransferMod =
+
PipeTransferTsFileHandler.class.getDeclaredField("effectiveTransferMod");
+ effectiveTransferMod.setAccessible(true);
+ Assert.assertFalse(effectiveTransferMod.getBoolean(handler));
+ } finally {
+ handler.close();
+ Assert.assertTrue(tsFile.delete());
+ Assert.assertTrue(modFile.getParentFile().delete());
+ }
+ }
+
@Test
public void testSealFailurePassesNestedReceiverMessageToRetryQueue() throws
Exception {
final File file = Files.createTempFile("pipe-transfer-seal-failure",
".tsfile").toFile();
@@ -160,4 +202,18 @@ public class PipeTransferTsFileHandlerCleanupTest {
false,
null);
}
+
+ private static final class DisappearingFile extends File {
+
+ private final AtomicInteger existenceCheckCount = new AtomicInteger();
+
+ private DisappearingFile(final String pathname) {
+ super(pathname);
+ }
+
+ @Override
+ public boolean exists() {
+ return existenceCheckCount.incrementAndGet() <= 2;
+ }
+ }
}
