This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/fix-history-pipe-flush-race-pr in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 159e3dd203724db8e7e81826aec54d18165882a4 Author: Caideyipi <[email protected]> AuthorDate: Mon Nov 2 18:52:21 2026 +0800 Fix history-only pipe replication and TsFile extraction races --- .../iotdb/consensus/i18n/IoTConsensusMessages.java | 2 + .../iotdb/consensus/i18n/IoTConsensusMessages.java | 3 + .../consensus/iot/IoTConsensusServerImpl.java | 41 +++- .../consensus/iot/IoTConsensusServerImplTest.java | 137 +++++++++++++ .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../common/tsfile/PipeTsFileInsertionEvent.java | 1 + .../source/dataregion/IoTDBDataRegionSource.java | 61 ++++++ ...istoricalDataRegionTsFileAndDeletionSource.java | 23 ++- .../pipe/event/PipeTsFileInsertionEventTest.java | 32 +++ .../db/pipe/source/IoTDBDataRegionSourceTest.java | 190 ++++++++++++++++++ ...ricalDataRegionTsFileAndDeletionSourceTest.java | 222 +++++++++++++++++++++ .../pipe/config/constant/PipeSourceConstant.java | 6 + 13 files changed, 708 insertions(+), 14 deletions(-) diff --git a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java index d07373015a8..1ec6587107e 100644 --- a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java +++ b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java @@ -154,6 +154,8 @@ public final class IoTConsensusMessages { "error when waiting %s to complete SyncLog. %s"; public static final String THREAD_INTERRUPTED_WAITING_SYNC_LOG = "thread interrupted when waiting %s to complete SyncLog. %s"; + public static final String EXCEPTION_TIMED_OUT_AFTER_ARG_MS_WAITING_FOR_ARG_TO_REPLICATE_THROUGH_INDEX_ARG_CURRENT_SYNC_INDEX_ARG_11DCFE73 = + "Timed out after %s ms waiting for %s to replicate through index %s (current sync index: %s)."; // ===================== Index controller ===================== diff --git a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java index 0d908aa8613..4da1a3cab16 100644 --- a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java +++ b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java @@ -154,6 +154,9 @@ public final class IoTConsensusMessages { public static final String THREAD_INTERRUPTED_WAITING_SYNC_LOG = "等待 %s 完成 SyncLog 时线程被中断。%s"; + public static final String EXCEPTION_TIMED_OUT_AFTER_ARG_MS_WAITING_FOR_ARG_TO_REPLICATE_THROUGH_INDEX_ARG_CURRENT_SYNC_INDEX_ARG_11DCFE73 = + "等待 %s ms 后，%s 仍未复制到 index %s（当前同步 index：%s）。"; + // ===================== Index 控制器 ===================== public static final String UPDATE_INDEX = diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index df7126cde4f..b29e75ebd35 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -866,27 +866,58 @@ public class IoTConsensusServerImpl { public void waitTargetPeerUntilSyncLogCompleted(Peer targetPeer) throws ConsensusGroupModifyPeerException { - long checkIntervalInMs = 10_000L; + waitTargetPeerUntilSyncLogCompleted(targetPeer, Long.MAX_VALUE, 10_000L, false); + } + + /** Wait for writes accepted before the first check, without chasing subsequent writes. */ + public void waitTargetPeerUntilSyncLogCompleted(Peer targetPeer, long timeoutMs) + throws ConsensusGroupModifyPeerException { + waitTargetPeerUntilSyncLogCompleted(targetPeer, timeoutMs, 1000L, true); + } + + private void waitTargetPeerUntilSyncLogCompleted( + Peer targetPeer, long timeoutMs, long checkIntervalInMs, boolean useFixedBoundary) + throws ConsensusGroupModifyPeerException { + final long startNanos = System.nanoTime(); + final long timeoutNanos = TimeUnit.MILLISECONDS.toNanos(timeoutMs); try (SyncIoTConsensusServiceClient client = syncClientManager.borrowClient(targetPeer.getEndpoint())) { + long targetSearchIndex = -1; while (true) { TWaitSyncLogCompleteRes res = client.waitSyncLogComplete( new TWaitSyncLogCompleteReq(targetPeer.getGroupId().convertToTConsensusGroupId())); - if (res.complete) { + // Capture a finite replication boundary even if the writer continues accepting data. + if (targetSearchIndex < 0) { + targetSearchIndex = res.searchIndex; + } + final long currentTargetIndex = useFixedBoundary ? targetSearchIndex : res.searchIndex; + if (useFixedBoundary ? res.safeIndex >= targetSearchIndex : res.complete) { logger.info( IoTConsensusMessages.WAIT_SYNC_LOG_COMPLETED, targetPeer, - res.searchIndex, + currentTargetIndex, res.safeIndex); return; } logger.info( IoTConsensusMessages.WAIT_SYNC_LOG_IN_PROGRESS, targetPeer, - res.searchIndex, + currentTargetIndex, res.safeIndex); - Thread.sleep(checkIntervalInMs); + final long remainingNanos = timeoutNanos - (System.nanoTime() - startNanos); + if (remainingNanos <= 0) { + throw new ConsensusGroupModifyPeerException( + String.format( + IoTConsensusMessages + .EXCEPTION_TIMED_OUT_AFTER_ARG_MS_WAITING_FOR_ARG_TO_REPLICATE_THROUGH_INDEX_ARG_CURRENT_SYNC_INDEX_ARG_11DCFE73, + timeoutMs, + targetPeer, + targetSearchIndex, + res.safeIndex)); + } + TimeUnit.NANOSECONDS.sleep( + Math.min(TimeUnit.MILLISECONDS.toNanos(checkIntervalInMs), remainingNanos)); } } catch (ClientManagerException | TException e) { throw new ConsensusGroupModifyPeerException( diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java index 291a8d99d1b..7d23b17d888 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java @@ -22,6 +22,7 @@ package org.apache.iotdb.consensus.iot; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.audit.UserDataTransferAuditHandler; +import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; @@ -31,7 +32,11 @@ import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsens import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.config.IoTConsensusConfig; import org.apache.iotdb.consensus.config.UserDataTransferAuditClassifier; +import org.apache.iotdb.consensus.exception.ConsensusGroupModifyPeerException; import org.apache.iotdb.consensus.i18n.ConsensusMessages; +import org.apache.iotdb.consensus.iot.client.SyncIoTConsensusServiceClient; +import org.apache.iotdb.consensus.iot.thrift.TWaitSyncLogCompleteReq; +import org.apache.iotdb.consensus.iot.thrift.TWaitSyncLogCompleteRes; import org.apache.iotdb.consensus.iot.util.TestEntry; import org.apache.iotdb.consensus.iot.util.TestStateMachine; import org.apache.iotdb.rpc.TSStatusCode; @@ -39,6 +44,7 @@ import org.apache.iotdb.rpc.TSStatusCode; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; import java.lang.reflect.Field; import java.util.Collections; @@ -57,6 +63,7 @@ import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class IoTConsensusServerImplTest { @@ -67,6 +74,136 @@ public class IoTConsensusServerImplTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test(timeout = 5000) + @SuppressWarnings("unchecked") + public void testReplicationWaitUsesFixedBoundaryWhileWriterContinues() throws Exception { + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final Peer thisNode = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667)); + final Peer remotePeer = new Peer(new DataRegionId(1), 1, new TEndPoint("127.0.0.1", 6668)); + final IClientManager<TEndPoint, SyncIoTConsensusServiceClient> manager = + Mockito.mock(IClientManager.class); + final SyncIoTConsensusServiceClient client = Mockito.mock(SyncIoTConsensusServiceClient.class); + Mockito.when(manager.borrowClient(remotePeer.getEndpoint())).thenReturn(client); + Mockito.when(client.waitSyncLogComplete(Mockito.any(TWaitSyncLogCompleteReq.class))) + .thenReturn(new TWaitSyncLogCompleteRes(false, 10, 0)) + // The snapshot is replicated although newer writes have not caught up yet. + .thenReturn(new TWaitSyncLogCompleteRes(false, 11, 10)) + .thenThrow(new AssertionError("replication boundary moved with the writer")); + try { + final IoTConsensusServerImpl server = + new IoTConsensusServerImpl( + temporaryFolder.getRoot().getAbsolutePath(), + null, + DirectoryStrategyType.SEQUENCE_STRATEGY, + thisNode, + Collections.singletonList(thisNode), + new TestStateMachine(), + backgroundTaskService, + null, + manager, + IoTConsensusConfig.newBuilder().build()); + server.waitTargetPeerUntilSyncLogCompleted(remotePeer, 3000); + Mockito.verify(client, Mockito.times(2)) + .waitSyncLogComplete(Mockito.any(TWaitSyncLogCompleteReq.class)); + Mockito.verify(client).close(); + } finally { + backgroundTaskService.shutdownNow(); + } + } + + @Test(timeout = 5000) + public void testReplicationWaitTimesOutAndReleasesClient() throws Exception { + assertReplicationWaitFailure(false); + } + + @Test(timeout = 30000) + @SuppressWarnings("unchecked") + public void testPeerRemovalWaitStillRequiresReplicationComplete() throws Exception { + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final Peer thisNode = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667)); + final Peer remotePeer = new Peer(new DataRegionId(1), 1, new TEndPoint("127.0.0.1", 6668)); + final IClientManager<TEndPoint, SyncIoTConsensusServiceClient> manager = + Mockito.mock(IClientManager.class); + final SyncIoTConsensusServiceClient client = Mockito.mock(SyncIoTConsensusServiceClient.class); + Mockito.when(manager.borrowClient(remotePeer.getEndpoint())).thenReturn(client); + Mockito.when(client.waitSyncLogComplete(Mockito.any(TWaitSyncLogCompleteReq.class))) + .thenReturn(new TWaitSyncLogCompleteRes(false, 10, 0)) + .thenReturn(new TWaitSyncLogCompleteRes(false, 11, 10)) + .thenReturn(new TWaitSyncLogCompleteRes(true, 11, 11)); + try { + final IoTConsensusServerImpl server = + new IoTConsensusServerImpl( + temporaryFolder.getRoot().getAbsolutePath(), + null, + DirectoryStrategyType.SEQUENCE_STRATEGY, + thisNode, + Collections.singletonList(thisNode), + new TestStateMachine(), + backgroundTaskService, + null, + manager, + IoTConsensusConfig.newBuilder().build()); + server.waitTargetPeerUntilSyncLogCompleted(remotePeer); + Mockito.verify(client, Mockito.times(3)) + .waitSyncLogComplete(Mockito.any(TWaitSyncLogCompleteReq.class)); + Mockito.verify(client).close(); + } finally { + backgroundTaskService.shutdownNow(); + } + } + + @Test(timeout = 5000) + public void testReplicationWaitPreservesInterruptAndReleasesClient() throws Exception { + assertReplicationWaitFailure(true); + } + + @SuppressWarnings("unchecked") + private void assertReplicationWaitFailure(final boolean interrupt) throws Exception { + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final Peer thisNode = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667)); + final Peer remotePeer = new Peer(new DataRegionId(1), 1, new TEndPoint("127.0.0.1", 6668)); + final IClientManager<TEndPoint, SyncIoTConsensusServiceClient> manager = + Mockito.mock(IClientManager.class); + final SyncIoTConsensusServiceClient client = Mockito.mock(SyncIoTConsensusServiceClient.class); + Mockito.when(manager.borrowClient(remotePeer.getEndpoint())).thenReturn(client); + Mockito.when(client.waitSyncLogComplete(Mockito.any(TWaitSyncLogCompleteReq.class))) + .thenReturn(new TWaitSyncLogCompleteRes(false, 10, 0)); + try { + final IoTConsensusServerImpl server = + new IoTConsensusServerImpl( + temporaryFolder.getRoot().getAbsolutePath(), + null, + DirectoryStrategyType.SEQUENCE_STRATEGY, + thisNode, + Collections.singletonList(thisNode), + new TestStateMachine(), + backgroundTaskService, + null, + manager, + IoTConsensusConfig.newBuilder().build()); + if (interrupt) { + Thread.currentThread().interrupt(); + } + final ConsensusGroupModifyPeerException exception = + assertThrows( + ConsensusGroupModifyPeerException.class, + () -> server.waitTargetPeerUntilSyncLogCompleted(remotePeer, interrupt ? 3000 : 0)); + assertEquals(interrupt, Thread.currentThread().isInterrupted()); + if (interrupt) { + assertTrue(exception.getCause() instanceof InterruptedException); + } else { + assertTrue(exception.getMessage().contains(remotePeer.toString())); + } + Mockito.verify(client).close(); + } finally { + Thread.interrupted(); + backgroundTaskService.shutdownNow(); + } + } + @Test public void testSyncLogBecomesReadOnlyWhileWaiting() throws Exception { final ScheduledExecutorService backgroundTaskService = 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 f56589e52c0..078333ad7d2 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 @@ -593,6 +593,8 @@ public final class DataNodePipeMessages { public static final String FAILED_TO_LOAD_SNAPSHOT_FROM_BYTEBUFFER = "Failed to load snapshot from byteBuffer {}."; public static final String FAILED_TO_START_SOURCES = "failed to start sources."; + public static final String EXCEPTION_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_MUST_BE_GREATER_THAN_0_4A4E80CB = + "history.wait-for-replication.timeout-ms must be greater than 0."; public static final String HEARTBEAT_EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE = "Heartbeat Event {} can not be supplied because the reference count can not be increased"; public static final String EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST = 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 1b8b488f30f..a7e4e69d95a 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 @@ -564,6 +564,8 @@ public final class DataNodePipeMessages { public static final String FAILED_TO_LOAD_SNAPSHOT_FROM_BYTEBUFFER = "加载 snapshot from byteBuffer {} 失败。"; public static final String FAILED_TO_START_SOURCES = "启动 sources 失败。"; + public static final String EXCEPTION_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_MUST_BE_GREATER_THAN_0_4A4E80CB = + "history.wait-for-replication.timeout-ms 必须大于 0。"; public static final String HEARTBEAT_EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE = "Heartbeat Event {} 无法被提供，因为其引用计数无法增加"; public static final String EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 5d06cf577b4..92674ee8afe 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -967,6 +967,7 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent public boolean shouldParseTime() { if (!isTimeParsed && Objects.nonNull(resource) + && (!isGeneratedByHistoricalExtractor || resource.isClosed()) && startTime <= resource.getFileStartTime() && resource.getFileEndTime() <= endTime) { isTimeParsed = true; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java index 9940127488a..02b37fca080 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java @@ -27,6 +27,12 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePatternOpera import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; import org.apache.iotdb.commons.pipe.source.IoTDBSource; import org.apache.iotdb.commons.queryengine.common.SqlDialect; +import org.apache.iotdb.consensus.IConsensus; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException; +import org.apache.iotdb.consensus.iot.IoTConsensus; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics; @@ -63,6 +69,7 @@ import javax.annotation.Nonnull; import java.time.ZoneId; import java.util.Arrays; import java.util.Objects; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -75,6 +82,8 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.E import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_DEFAULT_VALUE; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STREAMING_DEFAULT_VALUE; @@ -111,6 +120,7 @@ import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.S import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_LOOSE_RANGE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_TSFILE_ORDER_BY_QUERY_PRIORITY_KEY; +import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_SNAPSHOT_KEY; import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_STREAMING_KEY; @@ -142,6 +152,9 @@ public class IoTDBDataRegionSource extends IoTDBSource { private boolean hasNoExtractionNeed = true; private boolean shouldExtractDeletion = false; + private boolean shouldWaitForHistoricalReplication = false; + private long historicalReplicationTimeoutMs = + EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_DEFAULT_VALUE; private final AtomicBoolean completionSourceStarted = new AtomicBoolean(false); @Override @@ -253,6 +266,29 @@ public class IoTDBDataRegionSource extends IoTDBSource { constructHistoricalSource(); constructRealtimeSource(validator.getParameters()); + shouldWaitForHistoricalReplication = + realtimeSource instanceof PipeRealtimeDataRegionHeartbeatSource + && validator + .getParameters() + .getBooleanOrDefault( + Arrays.asList(EXTRACTOR_HISTORY_ENABLE_KEY, SOURCE_HISTORY_ENABLE_KEY), + EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE); + if (shouldWaitForHistoricalReplication) { + historicalReplicationTimeoutMs = + validator + .getParameters() + .getLongOrDefault( + Arrays.asList( + EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY, + SOURCE_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY), + EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_DEFAULT_VALUE); + validator.validate( + timeout -> (long) timeout > 0, + DataNodePipeMessages + .EXCEPTION_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_MUST_BE_GREATER_THAN_0_4A4E80CB, + historicalReplicationTimeoutMs); + } + historicalSource.validate(validator); realtimeSource.validate(validator); } @@ -531,6 +567,31 @@ public class IoTDBDataRegionSource extends IoTDBSource { return; } + // A history-only source cannot recover writes that reach this replica after its scan. + // Wait outside the DataRegion write lock so consensus replication can apply those writes. + if (shouldWaitForHistoricalReplication) { + final IConsensus consensus = DataRegionConsensusImpl.getInstance(); + if (consensus instanceof IoTConsensus) { + final DataRegionId dataRegionId = new DataRegionId(regionId); + final IoTConsensusServerImpl impl = ((IoTConsensus) consensus).getImpl(dataRegionId); + if (impl == null) { + throw new ConsensusGroupNotExistException(dataRegionId); + } + final long replicationStartNanos = System.nanoTime(); + for (final Peer peer : impl.getConfiguration()) { + if (!peer.equals(impl.getThisNode())) { + impl.waitTargetPeerUntilSyncLogCompleted( + peer, + Math.max( + 0, + historicalReplicationTimeoutMs + - TimeUnit.NANOSECONDS.toMillis( + System.nanoTime() - replicationStartNanos))); + } + } + } + } + final long startTime = System.currentTimeMillis(); LOGGER.info( DataNodePipeMessages.PIPE_STARTING_HISTORICAL_SOURCE_AND_REALTIME_SOURCE, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java index 80730d8f156..fed95151942 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java @@ -68,6 +68,7 @@ import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; import org.apache.tsfile.file.metadata.IDeviceID; @@ -868,8 +869,8 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource return false; } - // Some resource may not be closed due to the control of PIPE_MIN_FLUSH_INTERVAL_IN_MS. We - // simply ignore them. + // Closing files may still have incomplete progress and time metadata. Select them + // conservatively; the event waits for the file to close before parsing or transfer. if (!resource.isClosed() && Optional.ofNullable(resource.getProcessor()) .map(TsFileProcessor::alreadyMarkedClosing) @@ -1057,7 +1058,10 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource } private boolean isTsFileResourceCoveredByTimeRange(final TsFileResource resource) { - return historicalDataExtractionStartTime <= resource.getFileStartTime() + // In particular, a sequence file's end time is not final until it is closed. Do not disable + // row-level filtering based on the bounds observed while an asynchronous close is in flight. + return resource.isClosed() + && historicalDataExtractionStartTime <= resource.getFileStartTime() && historicalDataExtractionEndTime >= resource.getFileEndTime(); } @@ -1318,16 +1322,17 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource event.increaseReferenceCount( PipeHistoricalDataRegionTsFileAndDeletionSource.class.getName()); if (!isReferenceCountIncreased) { - LOGGER.warn( - DataNodePipeMessages.PIPE_FAILED_TO_INCREASE_REFERENCE_COUNT_FOR_1, - pipeName, - dataRegionId, - event); + // Keep the queue entry, pin, and replicate index until an event owns the resource. A + // failed reference acquisition must not advance historical progress past this file. + throw new PipeException( + String.format( + DataNodePipeMessages.INCREASE_REFERENCE_COUNT_ERROR_HOLDER_FMT, + event.coreReportMessage())); } filteredTsFileResources2TableNames.remove(resource); shouldUnpinResource = true; shouldClearReplicateIndex = true; - return isReferenceCountIncreased ? event : null; + return event; } finally { if (shouldClearReplicateIndex) { clearReplicateIndexForResource(resource); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java index e905e201ec2..b1a88bf7506 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java @@ -78,6 +78,38 @@ import static org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern.bu import static org.apache.iotdb.db.auth.AuthorityChecker.NO_PERMISSION_PROMOTION; public class PipeTsFileInsertionEventTest { + @Test + public void testRealtimeTsFileTimeParsingRemainsUnchanged() { + final TsFileResource resource = new TsFileResource(new File("1-1-0-0.tsfile")); + final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"t1", "d1"}); + resource.updateStartTime(device, 0); + resource.updateEndTime(device, Long.MIN_VALUE); + resource.setStatusForTest(TsFileResourceStatus.UNCLOSED); + final PipeTsFileInsertionEvent event = + new PipeTsFileInsertionEvent( + true, + "db1", + resource, + null, + false, + false, + false, + Collections.singleton("t1"), + "realtime_time_filter", + 1L, + null, + null, + new TablePattern(true, null, null), + null, + null, + null, + true, + -10, + 10); + Assert.assertFalse(event.shouldParseTime()); + event.close(); + } + @Test public void testHistoricalModsSurviveOriginalFileRemoval() throws Exception { assertHistoricalModsAfterOriginalFileRemoval(true, true); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java index f660423b0b0..b90dc4abfe4 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/IoTDBDataRegionSourceTest.java @@ -19,18 +19,208 @@ package org.apache.iotdb.db.pipe.source; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; +import org.apache.iotdb.commons.pipe.source.IoTDBSource; +import org.apache.iotdb.consensus.IConsensus; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.exception.ConsensusGroupModifyPeerException; +import org.apache.iotdb.consensus.iot.IoTConsensus; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource; +import org.apache.iotdb.db.pipe.source.dataregion.historical.PipeHistoricalDataRegionSource; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionHeartbeatSource; +import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource; +import org.apache.iotdb.db.storageengine.StorageEngine; +import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException; import org.junit.Assert; import org.junit.Test; +import org.mockito.InOrder; +import org.mockito.Mockito; +import java.lang.reflect.Field; +import java.util.Arrays; import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; public class IoTDBDataRegionSourceTest { + + @Test + public void testHistoryOnlyWaitsForReplicationBeforeLockingAndScanning() throws Exception { + assertReplicationBeforeHistoricalStart(false, false); + } + + @Test + public void testHistoryOnlyCanRetryAfterReplicationBarrierFailure() throws Exception { + assertReplicationBeforeHistoricalStart(false, true); + } + + @Test + public void testRealtimeSourceDoesNotWaitForReplicationBarrier() throws Exception { + for (final boolean historyEnabled : new boolean[] {false, true}) { + for (final String mode : new String[] {"hybrid", "log", "file", "forced-log"}) { + final Map<String, String> attributes = new HashMap<>(); + attributes.put( + PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY, String.valueOf(historyEnabled)); + attributes.put(PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY, "true"); + attributes.put(PipeSourceConstant.SOURCE_REALTIME_MODE_KEY, mode); + // This option belongs to history-only sources and must not affect realtime sources. + attributes.put( + PipeSourceConstant.SOURCE_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY, "unused"); + assertReplicationBeforeHistoricalStart(attributes, false, 30_000); + } + } + } + + @Test + public void testSnapshotSourceWaitsForReplicationBarrier() throws Exception { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(PipeSourceConstant.SOURCE_MODE_KEY, "snapshot"); + assertReplicationBeforeHistoricalStart(attributes, false, 30_000); + } + + @Test + public void testHistoryOnlyReplicationTimeoutIsConfigurable() throws Exception { + for (final String key : + new String[] { + PipeSourceConstant.SOURCE_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY, + PipeSourceConstant.EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY, + "history.wait-for-replication.timeout-ms" + }) { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY, "false"); + attributes.put(key, "90000"); + assertReplicationBeforeHistoricalStart(attributes, false, 90_000); + } + } + + @Test + public void testHistoryOnlyRejectsNonPositiveReplicationTimeout() throws Exception { + for (final String value : new String[] {"0", "-1"}) { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY, "false"); + attributes.put(PipeSourceConstant.SOURCE_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY, value); + try (final IoTDBDataRegionSource source = new IoTDBDataRegionSource()) { + Assert.assertThrows( + PipeParameterNotValidException.class, + () -> source.validate(new PipeParameterValidator(new PipeParameters(attributes)))); + } + } + } + + private void assertReplicationBeforeHistoricalStart( + final boolean realtimeEnabled, final boolean failFirst) throws Exception { + final Map<String, String> attributes = new HashMap<>(); + attributes.put(PipeSourceConstant.SOURCE_REALTIME_ENABLE_KEY, String.valueOf(realtimeEnabled)); + assertReplicationBeforeHistoricalStart(attributes, failFirst, 30_000); + } + + @SuppressWarnings("unchecked") + private void assertReplicationBeforeHistoricalStart( + final Map<String, String> attributes, final boolean failFirst, final long expectedTimeoutMs) + throws Exception { + final IoTDBDataRegionSource source = new IoTDBDataRegionSource(); + source.validate(new PipeParameterValidator(new PipeParameters(attributes))); + final Field realtimeSourceField = + IoTDBDataRegionSource.class.getDeclaredField("realtimeSource"); + realtimeSourceField.setAccessible(true); + final PipeRealtimeDataRegionSource actualRealtimeSource = + (PipeRealtimeDataRegionSource) realtimeSourceField.get(source); + final boolean realtimeEnabled = + !(actualRealtimeSource instanceof PipeRealtimeDataRegionHeartbeatSource); + final IConsensus originalConsensus = DataRegionConsensusImpl.getInstance(); + final IoTConsensus consensus = Mockito.mock(IoTConsensus.class); + final IoTConsensusServerImpl impl = Mockito.mock(IoTConsensusServerImpl.class); + final DataRegionId regionId = new DataRegionId(2147483644); + final Peer localPeer = new Peer(regionId, 1, new TEndPoint("127.0.0.1", 6667)); + final Peer remotePeer = new Peer(regionId, 2, new TEndPoint("127.0.0.1", 6668)); + Mockito.when(consensus.getImpl(regionId)).thenReturn(impl); + Mockito.when(impl.getConfiguration()).thenReturn(Arrays.asList(localPeer, remotePeer)); + Mockito.when(impl.getThisNode()).thenReturn(localPeer); + final DataRegion region = Mockito.mock(DataRegion.class); + final PipeHistoricalDataRegionSource historical = + Mockito.mock(PipeHistoricalDataRegionSource.class); + final PipeRealtimeDataRegionSource realtime = Mockito.mock(actualRealtimeSource.getClass()); + setField(IoTDBDataRegionSource.class, source, "historicalSource", historical); + setField(IoTDBDataRegionSource.class, source, "realtimeSource", realtime); + setField(IoTDBSource.class, source, "regionId", regionId.getId()); + setField(IoTDBSource.class, source, "pipeName", "historical_replication_barrier"); + final AtomicBoolean replicated = new AtomicBoolean(); + final AtomicInteger attempts = new AtomicInteger(); + Mockito.doAnswer( + invocation -> { + Mockito.verify(region, Mockito.never()).writeLock(Mockito.anyString()); + final long timeoutMs = invocation.getArgument(1); + Assert.assertTrue( + timeoutMs >= expectedTimeoutMs / 2 && timeoutMs <= expectedTimeoutMs); + if (failFirst && attempts.getAndIncrement() == 0) { + throw new ConsensusGroupModifyPeerException("mock replication failure"); + } + replicated.set(true); + return null; + }) + .when(impl) + .waitTargetPeerUntilSyncLogCompleted(Mockito.eq(remotePeer), Mockito.anyLong()); + Mockito.doAnswer( + invocation -> { + Assert.assertTrue(realtimeEnabled || replicated.get()); + return null; + }) + .when(historical) + .start(); + + final Field regionsField = StorageEngine.class.getDeclaredField("dataRegionMap"); + regionsField.setAccessible(true); + final Map<DataRegionId, DataRegion> regions = + (Map<DataRegionId, DataRegion>) regionsField.get(StorageEngine.getInstance()); + Assert.assertFalse(regions.containsKey(regionId)); + regions.put(regionId, region); + DataRegionConsensusImpl.setInstance(consensus); + try { + if (failFirst) { + Assert.assertThrows(ConsensusGroupModifyPeerException.class, source::start); + Mockito.verify(historical, Mockito.never()).start(); + Mockito.verify(region, Mockito.never()).writeLock(Mockito.anyString()); + } + source.start(); + final InOrder order = Mockito.inOrder(impl, region, realtime, historical); + if (realtimeEnabled) { + Mockito.verify(consensus, Mockito.never()).getImpl(Mockito.any(DataRegionId.class)); + Mockito.verify(impl, Mockito.never()) + .waitTargetPeerUntilSyncLogCompleted(Mockito.any(Peer.class), Mockito.anyLong()); + } else { + order + .verify(impl, Mockito.times(failFirst ? 2 : 1)) + .waitTargetPeerUntilSyncLogCompleted(Mockito.eq(remotePeer), Mockito.anyLong()); + Mockito.verify(impl, Mockito.never()) + .waitTargetPeerUntilSyncLogCompleted(Mockito.eq(localPeer), Mockito.anyLong()); + } + order.verify(region).writeLock(Mockito.anyString()); + order.verify(realtime).start(); + order.verify(historical).start(); + order.verify(region).writeUnlock(); + } finally { + regions.remove(regionId); + DataRegionConsensusImpl.setInstance(originalConsensus); + } + } + + private static void setField( + final Class<?> declaringClass, final Object object, final String name, final Object value) + throws ReflectiveOperationException { + final Field field = declaringClass.getDeclaredField(name); + field.setAccessible(true); + field.set(object, value); + } + @Test public void testIoTDBDataRegionExtractor() { try (final IoTDBDataRegionSource extractor = new IoTDBDataRegionSource()) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java index 991dab795f9..2ba204d4def 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java @@ -32,17 +32,24 @@ import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration; import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment; import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector; import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; import org.apache.iotdb.pipe.api.event.Event; +import org.apache.iotdb.pipe.api.exception.PipeException; +import org.apache.tsfile.file.metadata.IDeviceID; import org.junit.Assert; import org.junit.Test; @@ -54,14 +61,203 @@ import java.nio.file.Files; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; public class PipeHistoricalDataRegionTsFileAndDeletionSourceTest { + @Test + public void testClosingTsFileKeepsTimeFiltering() throws Exception { + // A closing sequence file can temporarily look wholly inside the extraction window even + // though the eventual file also contains rows outside it. + assertSuppliedTsFileTimeFiltering(false, 0, Long.MIN_VALUE, false, true); + } + + @Test + public void testClosedTsFileTimeFiltering() throws Exception { + assertSuppliedTsFileTimeFiltering(true, -10, 10, false, false); + assertSuppliedTsFileTimeFiltering(true, -20, 20, false, true); + } + + @Test + public void testClosingTsFileHonorsLooseTimeRange() throws Exception { + assertSuppliedTsFileTimeFiltering(false, 0, Long.MIN_VALUE, true, false); + } + + @Test + public void testHistoricalTsFileReferenceFailureRetriesWithoutAdvancingProgress() + throws Exception { + assertHistoricalTsFileReferenceFailure(true); + } + + @Test + public void testHistoricalTsFileReferenceFailureReleasesPinOnClose() throws Exception { + assertHistoricalTsFileReferenceFailure(false); + } + + @SuppressWarnings("unchecked") + private void assertHistoricalTsFileReferenceFailure(final boolean retry) throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalReferenceFailure").toFile(); + final String pipeName = "historical_reference_retry"; + final String resourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, 1L); + final PipeTsFileResourceManager manager = PipeDataNodeResourceManager.tsfile(); + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + final AtomicBoolean failPathResolution = new AtomicBoolean(); + PipeTsFileInsertionEvent event = null; + try { + final File partitionDir = new File(tempDir, "sequence/db/2147483645/0"); + Assert.assertTrue(partitionDir.mkdirs()); + final TsFileResource resource = createTsFileResource(partitionDir, "1-1-0-0.tsfile"); + resource.setFile(new CanonicalPathFailingFile(resource.getTsFilePath(), failPathResolution)); + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + resource.updateProgressIndex(new SimpleProgressIndex(0, 1)); + + setPrivateField(source, "pipeName", pipeName); + setPrivateField(source, "pipeNameWithCreationTime", resourcePipeName); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 2147483645); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "shouldExtractInsertion", true); + setPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority", true); + setPrivateField(source, "isModelDetected", true); + setPrivateField(source, "isTableModel", true); + setPrivateField(source, "tablePattern", new TablePattern(true, null, null)); + setPrivateField(source, "isDbNameCoveredByPattern", true); + final Map<TsFileResource, Set<String>> selectedResources = + (Map<TsFileResource, Set<String>>) + getPrivateField(source, "filteredTsFileResources2TableNames"); + selectedResources.put(resource, Collections.emptySet()); + final Set<PersistentResource> progressReportResources = + (Set<PersistentResource>) getPrivateField(source, "historicalProgressReportResources"); + progressReportResources.add(resource); + setPrivateField( + source, + "pendingQueue", + new ArrayDeque<PersistentResource>(Collections.singletonList(resource))); + + manager.pinTsFileResource(resource, false, resourcePipeName); + final File pinnedFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + resource.getTsFile(), resourcePipeName); + failPathResolution.set(true); + Assert.assertThrows(PipeException.class, source::supply); + Assert.assertEquals(1, source.getPendingQueueSize()); + Assert.assertTrue(selectedResources.containsKey(resource)); + Assert.assertTrue(progressReportResources.contains(resource)); + Assert.assertNull(getPrivateField(source, "pendingHistoricalProgressIndexToReport")); + Assert.assertEquals(1, manager.getFileReferenceCount(pinnedFile, resourcePipeName)); + + failPathResolution.set(false); + if (retry) { + event = (PipeTsFileInsertionEvent) source.supply(); + Assert.assertNotNull(event); + final Field resourceField = PipeTsFileInsertionEvent.class.getDeclaredField("resource"); + resourceField.setAccessible(true); + Assert.assertSame(resource, resourceField.get(event)); + Assert.assertEquals(0, source.getPendingQueueSize()); + Assert.assertFalse(selectedResources.containsKey(resource)); + // The event now owns the only remaining reference; the source has released its pin. + Assert.assertEquals(1, manager.getFileReferenceCount(pinnedFile, resourcePipeName)); + final Event progressEvent = source.supply(); + Assert.assertTrue(progressEvent instanceof ProgressReportEvent); + Assert.assertEquals( + new TimePartitionProgressIndex( + resource.getTimePartition(), resource.getMaxProgressIndex()), + ((ProgressReportEvent) progressEvent).getProgressIndex()); + } else { + source.close(); + Assert.assertEquals(0, manager.getFileReferenceCount(pinnedFile, resourcePipeName)); + } + } finally { + failPathResolution.set(false); + if (event != null) { + event.clearReferenceCount(getClass().getName()); + } + source.close(); + manager.cleanPipeTsFileDir(resourcePipeName); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + + @SuppressWarnings("unchecked") + private void assertSuppliedTsFileTimeFiltering( + final boolean closed, + final long fileStartTime, + final long fileEndTime, + final boolean looseTimeRange, + final boolean shouldParseTime) + throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalClosingTimeFilter").toFile(); + final String pipeName = "closing_time_filter"; + final String resourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, 1L); + final PipeTsFileResourceManager manager = PipeDataNodeResourceManager.tsfile(); + final PipeHistoricalDataRegionTsFileAndDeletionSource source = + new PipeHistoricalDataRegionTsFileAndDeletionSource(); + TsFileResource resource = null; + PipeTsFileInsertionEvent event = null; + boolean pinned = false; + try { + final File partitionDir = new File(tempDir, "sequence/db/2147483646/0"); + Assert.assertTrue(partitionDir.mkdirs()); + resource = createTsFileResource(partitionDir, "1-1-0-0.tsfile"); + final IDeviceID device = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1", "device1"}); + resource.updateStartTime(device, fileStartTime); + resource.updateEndTime(device, fileEndTime); + resource.setStatusForTest( + closed ? TsFileResourceStatus.NORMAL : TsFileResourceStatus.UNCLOSED); + manager.pinTsFileResource(resource, false, resourcePipeName); + pinned = true; + + setPrivateField(source, "pipeName", pipeName); + setPrivateField(source, "pipeNameWithCreationTime", resourcePipeName); + setPrivateField(source, "creationTime", 1L); + setPrivateField(source, "dataRegionId", 2147483646); + setPrivateField(source, "isModelDetected", true); + setPrivateField(source, "isTableModel", true); + setPrivateField(source, "tablePattern", new TablePattern(true, null, null)); + setPrivateField(source, "isDbNameCoveredByPattern", true); + setPrivateField(source, "historicalDataExtractionStartTime", -10L); + setPrivateField(source, "historicalDataExtractionEndTime", 10L); + setPrivateField(source, "sloppyTimeRange", looseTimeRange); + ((Map<TsFileResource, Set<String>>) + getPrivateField(source, "filteredTsFileResources2TableNames")) + .put(resource, Collections.emptySet()); + + event = (PipeTsFileInsertionEvent) source.supplyTsFileEvent(resource); + pinned = false; + Assert.assertNotNull(event); + // The collector must not send this entire file without filtering after its close completes. + Assert.assertEquals(shouldParseTime, event.shouldParseTime()); + Assert.assertEquals(!shouldParseTime, PipeEventCollector.canSkipParsing4TsFileEvent(event)); + if (!closed && !looseTimeRange) { + resource.updateStartTime(device, -20); + resource.updateEndTime(device, 20); + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + Assert.assertTrue(event.shouldParseTime()); + Assert.assertFalse(PipeEventCollector.canSkipParsing4TsFileEvent(event)); + } + } finally { + if (event != null) { + event.clearReferenceCount(getClass().getName()); + } + if (pinned) { + manager.unpinTsFileResource(resource, resourcePipeName, false); + } + source.close(); + manager.cleanPipeTsFileDir(resourcePipeName); + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testDeletionOnlyCustomizeInitializesSourceContext() throws Exception { final PipeHistoricalDataRegionTsFileAndDeletionSource source = @@ -865,6 +1061,32 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSourceTest { return field.get(source); } + private static class CanonicalPathFailingFile extends File { + + private final AtomicBoolean failPathResolution; + + private CanonicalPathFailingFile(final String path, final AtomicBoolean failPathResolution) { + super(path); + this.failPathResolution = failPathResolution; + } + + @Override + public File getParentFile() { + final File parent = super.getParentFile(); + return parent == null + ? null + : new CanonicalPathFailingFile(parent.getPath(), failPathResolution); + } + + @Override + public String getCanonicalPath() throws IOException { + if (failPathResolution.get()) { + throw new IOException("mock path resolution failure"); + } + return super.getCanonicalPath(); + } + } + private static class TestablePipeHistoricalDataRegionTsFileAndDeletionSource extends PipeHistoricalDataRegionTsFileAndDeletionSource { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java index 70d452ae36a..cf83f3e1c32 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java @@ -98,6 +98,12 @@ public class PipeSourceConstant { public static final String EXTRACTOR_HISTORY_ENABLE_KEY = "extractor.history.enable"; public static final String SOURCE_HISTORY_ENABLE_KEY = "source.history.enable"; public static final boolean EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE = true; + public static final String EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY = + "extractor.history.wait-for-replication.timeout-ms"; + public static final String SOURCE_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_KEY = + "source.history.wait-for-replication.timeout-ms"; + public static final long EXTRACTOR_HISTORY_WAIT_FOR_REPLICATION_TIMEOUT_MS_DEFAULT_VALUE = + 30_000L; public static final String EXTRACTOR_HISTORY_START_TIME_KEY = "extractor.history.start-time"; public static final String SOURCE_HISTORY_START_TIME_KEY = "source.history.start-time"; public static final String EXTRACTOR_HISTORY_END_TIME_KEY = "extractor.history.end-time";
