This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch cp15048 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3a24f70dc82bf77d4bf8478f1c5078d8a17bdfa4 Author: Steve Yurong Su <[email protected]> AuthorDate: Mon Mar 10 19:34:36 2025 +0800 Pipe: Optimize realtime performace when pipe starts after long time stop with heavy data backlog (#15048) --- .../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 54 +++++++++++++++++++++- .../PipeRealtimePriorityBlockingQueue.java | 49 ++++++++++++++------ .../async/IoTDBDataRegionAsyncConnector.java | 20 +++++++- .../PipeRealtimeDataRegionHybridExtractor.java | 7 +++ .../dataregion/wal/utils/WALInsertNodeCache.java | 7 ++- .../apache/iotdb/commons/conf/CommonConfig.java | 44 +++++++++++++++--- .../iotdb/commons/conf/CommonDescriptor.java | 28 +++++++++-- .../commons/pipe/agent/task/PipeTaskAgent.java | 2 +- .../iotdb/commons/pipe/config/PipeConfig.java | 27 +++++++++-- 9 files changed, 205 insertions(+), 33 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java index 610424a9531..139c5cef37a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java @@ -120,6 +120,54 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent { return new PipeDataNodeBuilder(pipeMetaFromConfigNode).build(); } + ////////////////////////// Manage by Pipe Name ////////////////////////// + + @Override + protected void startPipe(final String pipeName, final long creationTime) { + final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); + final PipeStatus status = existedPipeMeta.getRuntimeMeta().getStatus().get(); + if (PipeStatus.STOPPED.equals(status) || status == null) { + restartPipeToReloadResourceIfNeeded(existedPipeMeta); + } + + super.startPipe(pipeName, creationTime); + } + + private void restartPipeToReloadResourceIfNeeded(final PipeMeta pipeMeta) { + if (System.currentTimeMillis() - pipeMeta.getStaticMeta().getCreationTime() + < PipeConfig.getInstance().getPipeStuckRestartMinIntervalMs()) { + return; + } + + final AtomicLong lastRestartTime = + PIPE_NAME_TO_LAST_RESTART_TIME_MAP.get(pipeMeta.getStaticMeta().getPipeName()); + if (lastRestartTime != null + && System.currentTimeMillis() - lastRestartTime.get() + < PipeConfig.getInstance().getPipeStuckRestartMinIntervalMs()) { + LOGGER.info( + "Skipping reload resource for stopped pipe {} before starting it because reloading resource is too frequent.", + pipeMeta.getStaticMeta().getPipeName()); + return; + } + + if (PIPE_NAME_TO_LAST_RESTART_TIME_MAP.isEmpty()) { + LOGGER.info( + "Flushing storage engine before restarting pipe {}.", + pipeMeta.getStaticMeta().getPipeName()); + final long currentTime = System.currentTimeMillis(); + StorageEngine.getInstance().syncCloseAllProcessor(); + WALManager.getInstance().syncDeleteOutdatedFilesInWALNodes(); + LOGGER.info( + "Finished flushing storage engine, time cost: {} ms.", + System.currentTimeMillis() - currentTime); + } + + restartStuckPipe(pipeMeta); + LOGGER.info( + "Reloaded resource for stopped pipe {} before starting it.", + pipeMeta.getStaticMeta().getPipeName()); + } + ///////////////////////// Manage by regionGroupId ///////////////////////// @Override @@ -672,7 +720,9 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent { } private void restartStuckPipe(final PipeMeta pipeMeta) { - LOGGER.warn("Pipe {} will be restarted because of stuck.", pipeMeta.getStaticMeta()); + LOGGER.warn( + "Pipe {} will be restarted because it is stuck or has encountered issues such as data backlog or being stopped for too long.", + pipeMeta.getStaticMeta()); acquireWriteLock(); try { final long startTime = System.currentTimeMillis(); @@ -686,7 +736,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent { handleSinglePipeMetaChanges(originalPipeMeta); LOGGER.warn( - "Pipe {} was restarted because of stuck, time cost: {} ms.", + "Pipe {} was restarted because of stuck or data backlog, time cost: {} ms.", originalPipeMeta.getStaticMeta(), System.currentTimeMillis() - startTime); } catch (final Exception e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeRealtimePriorityBlockingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeRealtimePriorityBlockingQueue.java index a170b19bd42..d9eac5f5625 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeRealtimePriorityBlockingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeRealtimePriorityBlockingQueue.java @@ -33,6 +33,7 @@ import java.util.Objects; import java.util.concurrent.BlockingDeque; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQueue<Event> { @@ -40,10 +41,13 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ private final BlockingDeque<TsFileInsertionEvent> tsfileInsertEventDeque = new LinkedBlockingDeque<>(); - private final AtomicInteger eventCount = new AtomicInteger(0); + private static final int POLL_TSFILE_THRESHOLD = + PipeConfig.getInstance().getPipeRealTimeQueuePollTsFileThreshold(); + private final AtomicInteger pollTsFileCounter = new AtomicInteger(0); - private static final int pollHistoryThreshold = - PipeConfig.getInstance().getPipeRealTimeQueuePollHistoryThreshold(); + private static final int POLL_HISTORICAL_TSFILE_THRESHOLD = + Math.max(PipeConfig.getInstance().getPipeRealTimeQueuePollHistoricalTsFileThreshold(), 1); + private final AtomicLong pollHistoricalTsFileCounter = new AtomicLong(0); public PipeRealtimePriorityBlockingQueue() { super(new PipeDataRegionEventCounter()); @@ -81,18 +85,24 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ @Override public Event directPoll() { Event event = null; - if (eventCount.get() >= pollHistoryThreshold) { - event = tsfileInsertEventDeque.pollFirst(); - eventCount.set(0); + if (pollTsFileCounter.get() >= POLL_TSFILE_THRESHOLD) { + event = + pollHistoricalTsFileCounter.incrementAndGet() % POLL_HISTORICAL_TSFILE_THRESHOLD == 0 + ? tsfileInsertEventDeque.pollFirst() + : tsfileInsertEventDeque.pollLast(); + pollTsFileCounter.set(0); } if (Objects.isNull(event)) { // Sequentially poll the first offered non-TsFileInsertionEvent event = super.directPoll(); if (Objects.isNull(event)) { - event = tsfileInsertEventDeque.pollFirst(); + event = + pollHistoricalTsFileCounter.incrementAndGet() % POLL_HISTORICAL_TSFILE_THRESHOLD == 0 + ? tsfileInsertEventDeque.pollFirst() + : tsfileInsertEventDeque.pollLast(); } if (event != null) { - eventCount.incrementAndGet(); + pollTsFileCounter.incrementAndGet(); } } @@ -113,18 +123,24 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ @Override public Event waitedPoll() { Event event = null; - if (eventCount.get() >= pollHistoryThreshold) { - event = tsfileInsertEventDeque.pollFirst(); - eventCount.set(0); + if (pollTsFileCounter.get() >= POLL_TSFILE_THRESHOLD) { + event = + pollHistoricalTsFileCounter.incrementAndGet() % POLL_HISTORICAL_TSFILE_THRESHOLD == 0 + ? tsfileInsertEventDeque.pollFirst() + : tsfileInsertEventDeque.pollLast(); + pollTsFileCounter.set(0); } if (event == null) { // Sequentially poll the first offered non-TsFileInsertionEvent event = super.directPoll(); if (event == null && !tsfileInsertEventDeque.isEmpty()) { - event = tsfileInsertEventDeque.pollFirst(); + event = + pollHistoricalTsFileCounter.incrementAndGet() % POLL_HISTORICAL_TSFILE_THRESHOLD == 0 + ? tsfileInsertEventDeque.pollFirst() + : tsfileInsertEventDeque.pollLast(); } if (event != null) { - eventCount.incrementAndGet(); + pollTsFileCounter.incrementAndGet(); } } @@ -132,10 +148,13 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ if (Objects.isNull(event)) { event = super.waitedPoll(); if (Objects.isNull(event)) { - event = tsfileInsertEventDeque.pollFirst(); + event = + pollHistoricalTsFileCounter.incrementAndGet() % POLL_HISTORICAL_TSFILE_THRESHOLD == 0 + ? tsfileInsertEventDeque.pollFirst() + : tsfileInsertEventDeque.pollLast(); } if (event != null) { - eventCount.incrementAndGet(); + pollTsFileCounter.incrementAndGet(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java index 8ef6d3d9d6e..d3c3c73b728 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/thrift/async/IoTDBDataRegionAsyncConnector.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.pipe.connector.protocol.thrift.async; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; +import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.connector.protocol.IoTDBConnector; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.pipe.agent.task.subtask.connector.PipeConnectorSubtask; @@ -96,6 +97,8 @@ public class IoTDBDataRegionAsyncConnector extends IoTDBConnector { private final IoTDBDataRegionSyncConnector retryConnector = new IoTDBDataRegionSyncConnector(); private final BlockingQueue<Event> retryEventQueue = new LinkedBlockingQueue<>(); + private final long maxRetryExecutionTimeMsPerCall = + PipeConfig.getInstance().getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall(); private IoTDBDataNodeAsyncClientManager clientManager; @@ -431,11 +434,21 @@ public class IoTDBDataRegionAsyncConnector extends IoTDBConnector { * @see PipeConnector#transfer(TsFileInsertionEvent) for more details. */ private void transferQueuedEventsIfNecessary() throws Exception { + if (retryEventQueue.isEmpty()) { + // Trigger cron heartbeat event in retry connector to send batch in time + retryConnector.transfer(PipeConnectorSubtask.CRON_HEARTBEAT_EVENT); + return; + } + + final long retryStartTime = System.currentTimeMillis(); while (!retryEventQueue.isEmpty()) { synchronized (this) { - if (isClosed.get() || retryEventQueue.isEmpty()) { + if (isClosed.get()) { return; } + if (retryEventQueue.isEmpty()) { + break; + } final Event peekedEvent = retryEventQueue.peek(); @@ -468,6 +481,11 @@ public class IoTDBDataRegionAsyncConnector extends IoTDBConnector { LOGGER.debug("Polled event {} from retry queue.", polledEvent); } } + + // Stop retrying if the execution time exceeds the threshold for better realtime performance + if (System.currentTimeMillis() - retryStartTime > maxRetryExecutionTimeMsPerCall) { + break; + } } // Trigger cron heartbeat event in retry connector to send batch in time diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java index 74360f244fc..cf5e88a72a6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/PipeRealtimeDataRegionHybridExtractor.java @@ -48,6 +48,9 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio private static final Logger LOGGER = LoggerFactory.getLogger(PipeRealtimeDataRegionHybridExtractor.class); + private final boolean isPipeEpochKeepTsFileAfterStuckRestartEnabled = + PipeConfig.getInstance().isPipeEpochKeepTsFileAfterStuckRestartEnabled(); + @Override protected void doExtract(final PipeRealtimeEvent event) { final Event eventToExtract = event.getEvent(); @@ -223,6 +226,10 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio } private boolean isPipeTaskCurrentlyRestarted(final PipeRealtimeEvent event) { + if (!isPipeEpochKeepTsFileAfterStuckRestartEnabled) { + return false; + } + final boolean isPipeTaskCurrentlyRestarted = PipeDataNodeAgent.task().isPipeTaskCurrentlyRestarted(pipeName); if (isPipeTaskCurrentlyRestarted && event.mayExtractorUseTablets(this)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java index 2c03a605caf..2e1c6616b33 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCache.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.utils; +import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.conf.DataNodeMemoryConfig; import org.apache.iotdb.db.conf.IoTDBConfig; @@ -61,6 +62,7 @@ public class WALInsertNodeCache { private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); private static final DataNodeMemoryConfig MEMORY_CONFIG = IoTDBDescriptor.getInstance().getMemoryConfig(); + private static final PipeConfig PIPE_CONFIG = PipeConfig.getInstance(); private final PipeMemoryBlock allocatedMemoryBlock; // Used to adjust the memory usage of the cache @@ -78,8 +80,9 @@ public class WALInsertNodeCache { final long requestedAllocateSize = (long) Math.min( - (double) 2 * CONFIG.getWalFileSizeThresholdInByte(), - MEMORY_CONFIG.getPipeMemoryManager().getTotalMemorySizeInBytes() * 0.8 / 5); + (double) PIPE_CONFIG.getPipeMaxAllowedPinnedMemTableCount() + * CONFIG.getWalFileSizeThresholdInByte(), + MEMORY_CONFIG.getPipeMemoryManager().getTotalMemorySizeInBytes() * 0.45); allocatedMemoryBlock = PipeDataNodeResourceManager.memory() .tryAllocate(requestedAllocateSize) diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index cb69d5db391..d37e1c1eda9 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -204,7 +204,8 @@ public class CommonConfig { private boolean pipeFileReceiverFsyncEnabled = true; - private int pipeRealTimeQueuePollHistoryThreshold = 1; + private int pipeRealTimeQueuePollTsFileThreshold = 10; + private int pipeRealTimeQueuePollHistoricalTsFileThreshold = 3; /** The maximum number of threads that can be used to execute subtasks in PipeSubtaskExecutor. */ private int pipeSubtaskExecutorMaxThreadNum = @@ -234,6 +235,7 @@ public class CommonConfig { private long pipeConnectorRetryIntervalMs = 1000L; private boolean pipeConnectorRPCThriftCompressionEnabled = false; + private long pipeAsyncConnectorMaxRetryExecutionTimeMsPerCall = 500; private int pipeAsyncConnectorSelectorNumber = Math.max(4, Runtime.getRuntime().availableProcessors() / 2); private int pipeAsyncConnectorMaxClientNumber = @@ -258,12 +260,13 @@ public class CommonConfig { private long pipeReceiverLoginPeriodicVerificationIntervalMs = 300000; private int pipeMaxAllowedHistoricalTsFilePerDataRegion = 100; - private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = 10; + private int pipeMaxAllowedPendingTsFileEpochPerDataRegion = 5; private int pipeMaxAllowedPinnedMemTableCount = 10; // per data region private long pipeMaxAllowedLinkedTsFileCount = 300; private float pipeMaxAllowedLinkedDeletedTsFileDiskUsagePercentage = 0.1F; private long pipeStuckRestartIntervalSeconds = 120; private long pipeStuckRestartMinIntervalMs = 5 * 60 * 1000L; // 5 minutes + private boolean pipeEpochKeepTsFileAfterStuckRestartEnabled = false; private long pipeStorageEngineFlushTimeIntervalMs = Long.MAX_VALUE; private int pipeMetaReportMaxLogNumPerRound = 10; @@ -857,6 +860,16 @@ public class CommonConfig { return pipeConnectorRPCThriftCompressionEnabled; } + public void setPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall( + long pipeAsyncConnectorMaxRetryExecutionTimeMsPerCall) { + this.pipeAsyncConnectorMaxRetryExecutionTimeMsPerCall = + pipeAsyncConnectorMaxRetryExecutionTimeMsPerCall; + } + + public long getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall() { + return pipeAsyncConnectorMaxRetryExecutionTimeMsPerCall; + } + public int getPipeAsyncConnectorSelectorNumber() { return pipeAsyncConnectorSelectorNumber; } @@ -990,12 +1003,22 @@ public class CommonConfig { this.pipeSubtaskExecutorForcedRestartIntervalMs = pipeSubtaskExecutorForcedRestartIntervalMs; } - public int getPipeRealTimeQueuePollHistoryThreshold() { - return pipeRealTimeQueuePollHistoryThreshold; + public int getPipeRealTimeQueuePollTsFileThreshold() { + return pipeRealTimeQueuePollTsFileThreshold; + } + + public void setPipeRealTimeQueuePollTsFileThreshold(int pipeRealTimeQueuePollTsFileThreshold) { + this.pipeRealTimeQueuePollTsFileThreshold = pipeRealTimeQueuePollTsFileThreshold; } - public void setPipeRealTimeQueuePollHistoryThreshold(int pipeRealTimeQueuePollHistoryThreshold) { - this.pipeRealTimeQueuePollHistoryThreshold = pipeRealTimeQueuePollHistoryThreshold; + public int getPipeRealTimeQueuePollHistoricalTsFileThreshold() { + return pipeRealTimeQueuePollHistoricalTsFileThreshold; + } + + public void setPipeRealTimeQueuePollHistoricalTsFileThreshold( + int pipeRealTimeQueuePollHistoricalTsFileThreshold) { + this.pipeRealTimeQueuePollHistoricalTsFileThreshold = + pipeRealTimeQueuePollHistoricalTsFileThreshold; } public void setPipeAirGapReceiverEnabled(boolean pipeAirGapReceiverEnabled) { @@ -1077,6 +1100,10 @@ public class CommonConfig { return pipeStuckRestartMinIntervalMs; } + public boolean isPipeEpochKeepTsFileAfterStuckRestartEnabled() { + return pipeEpochKeepTsFileAfterStuckRestartEnabled; + } + public long getPipeStorageEngineFlushTimeIntervalMs() { return pipeStorageEngineFlushTimeIntervalMs; } @@ -1089,6 +1116,11 @@ public class CommonConfig { this.pipeStuckRestartMinIntervalMs = pipeStuckRestartMinIntervalMs; } + public void setPipeEpochKeepTsFileAfterStuckRestartEnabled( + boolean pipeEpochKeepTsFileAfterStuckRestartEnabled) { + this.pipeEpochKeepTsFileAfterStuckRestartEnabled = pipeEpochKeepTsFileAfterStuckRestartEnabled; + } + public void setPipeStorageEngineFlushTimeIntervalMs(long pipeStorageEngineFlushTimeIntervalMs) { this.pipeStorageEngineFlushTimeIntervalMs = pipeStorageEngineFlushTimeIntervalMs; } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java index 5eee1b68ef7..dff80612307 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java @@ -304,11 +304,19 @@ public class CommonDescriptor { String.valueOf( config.getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold())))); - config.setPipeRealTimeQueuePollHistoryThreshold( + config.setPipeRealTimeQueuePollTsFileThreshold( + Integer.parseInt( + Optional.ofNullable( + properties.getProperty("pipe_realtime_queue_poll_history_threshold")) + .orElse( + properties.getProperty( + "pipe_realtime_queue_poll_tsfile_threshold", + String.valueOf(config.getPipeRealTimeQueuePollTsFileThreshold()))))); + config.setPipeRealTimeQueuePollHistoricalTsFileThreshold( Integer.parseInt( properties.getProperty( - "pipe_realtime_queue_poll_history_threshold", - Integer.toString(config.getPipeRealTimeQueuePollHistoryThreshold())))); + "pipe_realtime_queue_poll_historical_tsfile_threshold", + String.valueOf(config.getPipeRealTimeQueuePollHistoricalTsFileThreshold())))); int pipeSubtaskExecutorMaxThreadNum = Integer.parseInt( @@ -417,6 +425,15 @@ public class CommonDescriptor { properties.getProperty( "pipe_connector_rpc_thrift_compression_enabled", String.valueOf(config.isPipeConnectorRPCThriftCompressionEnabled()))))); + config.setPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall( + Long.parseLong( + Optional.ofNullable( + properties.getProperty("pipe_async_sink_max_retry_execution_time_ms_per_call")) + .orElse( + properties.getProperty( + "pipe_async_connector_max_retry_execution_time_ms_per_call", + String.valueOf( + config.getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall()))))); int pipeAsyncConnectorSelectorNumber = Integer.parseInt( Optional.ofNullable(properties.getProperty("pipe_sink_selector_number")) @@ -536,6 +553,11 @@ public class CommonDescriptor { properties.getProperty( "pipe_stuck_restart_min_interval_ms", String.valueOf(config.getPipeStuckRestartMinIntervalMs())))); + config.setPipeEpochKeepTsFileAfterStuckRestartEnabled( + Boolean.parseBoolean( + properties.getProperty( + "pipe_epoch_keep_tsfile_after_stuck_restart_enabled", + String.valueOf(config.isPipeEpochKeepTsFileAfterStuckRestartEnabled())))); config.setPipeStorageEngineFlushTimeIntervalMs( Long.parseLong( properties.getProperty( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java index bd1ec3e2592..9b099cd4cdc 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java @@ -578,7 +578,7 @@ public abstract class PipeTaskAgent { return true; } - private void startPipe(final String pipeName, final long creationTime) { + protected void startPipe(final String pipeName, final long creationTime) { final PipeMeta existedPipeMeta = pipeMetaKeeper.getPipeMeta(pipeName); if (!checkBeforeStartPipe(existedPipeMeta, pipeName, creationTime)) { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java index 4492df7a5db..0f558782f00 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java @@ -78,8 +78,12 @@ public class PipeConfig { /////////////////////////////// Subtask Connector /////////////////////////////// - public int getPipeRealTimeQueuePollHistoryThreshold() { - return COMMON_CONFIG.getPipeRealTimeQueuePollHistoryThreshold(); + public int getPipeRealTimeQueuePollTsFileThreshold() { + return COMMON_CONFIG.getPipeRealTimeQueuePollTsFileThreshold(); + } + + public int getPipeRealTimeQueuePollHistoricalTsFileThreshold() { + return COMMON_CONFIG.getPipeRealTimeQueuePollHistoricalTsFileThreshold(); } /////////////////////////////// Subtask Executor /////////////////////////////// @@ -148,6 +152,10 @@ public class PipeConfig { return COMMON_CONFIG.isPipeConnectorRPCThriftCompressionEnabled(); } + public long getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall() { + return COMMON_CONFIG.getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall(); + } + public int getPipeAsyncConnectorSelectorNumber() { return COMMON_CONFIG.getPipeAsyncConnectorSelectorNumber(); } @@ -264,6 +272,10 @@ public class PipeConfig { return COMMON_CONFIG.getPipeStuckRestartMinIntervalMs(); } + public boolean isPipeEpochKeepTsFileAfterStuckRestartEnabled() { + return COMMON_CONFIG.isPipeEpochKeepTsFileAfterStuckRestartEnabled(); + } + public long getPipeStorageEngineFlushTimeIntervalMs() { return COMMON_CONFIG.getPipeStorageEngineFlushTimeIntervalMs(); } @@ -373,7 +385,10 @@ public class PipeConfig { getPipeDataStructureTsFileMemoryBlockAllocationRejectThreshold()); LOGGER.info( - "PipeRealTimeQueuePollHistoryThreshold: {}", getPipeRealTimeQueuePollHistoryThreshold()); + "PipeRealTimeQueuePollTsFileThreshold: {}", getPipeRealTimeQueuePollTsFileThreshold()); + LOGGER.info( + "PipeRealTimeQueuePollHistoricalTsFileThreshold: {}", + getPipeRealTimeQueuePollHistoricalTsFileThreshold()); LOGGER.info("PipeSubtaskExecutorMaxThreadNum: {}", getPipeSubtaskExecutorMaxThreadNum()); LOGGER.info( @@ -423,6 +438,9 @@ public class PipeConfig { "PipeRemainingTimeCommitRateAverageTime: {}", getPipeRemainingTimeCommitRateAverageTime()); LOGGER.info("PipeTsFileScanParsingThreshold(): {}", getPipeTsFileScanParsingThreshold()); + LOGGER.info( + "PipeAsyncConnectorMaxRetryExecutionTimeMsPerCall: {}", + getPipeAsyncConnectorMaxRetryExecutionTimeMsPerCall()); LOGGER.info("PipeAsyncConnectorSelectorNumber: {}", getPipeAsyncConnectorSelectorNumber()); LOGGER.info("PipeAsyncConnectorMaxClientNumber: {}", getPipeAsyncConnectorMaxClientNumber()); @@ -468,6 +486,9 @@ public class PipeConfig { getPipeMaxAllowedLinkedDeletedTsFileDiskUsagePercentage()); LOGGER.info("PipeStuckRestartIntervalSeconds: {}", getPipeStuckRestartIntervalSeconds()); LOGGER.info("PipeStuckRestartMinIntervalMs: {}", getPipeStuckRestartMinIntervalMs()); + LOGGER.info( + "PipeEpochKeepTsFileAfterStuckRestartEnabled: {}", + isPipeEpochKeepTsFileAfterStuckRestartEnabled()); LOGGER.info( "PipeStorageEngineFlushTimeIntervalMs: {}", getPipeStorageEngineFlushTimeIntervalMs());
