You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2025/02/19 02:01:15 UTC
(iotdb) branch master updated: Pipe: Reduce degrade log from PipeRealtimeDataRegionHybridExtractor (#14888)
This is an automated email from the ASF dual-hosted git repository.
rong 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 f8082c651aa Pipe: Reduce degrade log from PipeRealtimeDataRegionHybridExtractor (#14888)
f8082c651aa is described below
commit f8082c651aa7cf98b054f245fbb60a237207ec43
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Wed Feb 19 10:01:08 2025 +0800
Pipe: Reduce degrade log from PipeRealtimeDataRegionHybridExtractor (#14888)
---
.../db/pipe/event/realtime/PipeRealtimeEvent.java | 6 +++
.../PipeRealtimeDataRegionHybridExtractor.java | 48 +++++++++++-----------
2 files changed, 30 insertions(+), 24 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java
index bf0abd9c9fa..94a6ff9191c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEvent.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.db.pipe.event.common.PipeInsertionEvent;
+import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.PipeRealtimeDataRegionExtractor;
import org.apache.iotdb.db.pipe.extractor.dataregion.realtime.epoch.TsFileEpoch;
import org.apache.tsfile.file.metadata.IDeviceID;
@@ -102,6 +103,11 @@ public class PipeRealtimeEvent extends EnrichedEvent {
device2Measurements = null;
}
+ public boolean mayExtractorUseTablets(final PipeRealtimeDataRegionExtractor extractor) {
+ final TsFileEpoch.State state = tsFileEpoch.getState(extractor);
+ return state.equals(TsFileEpoch.State.EMPTY) || state.equals(TsFileEpoch.State.USING_TABLET);
+ }
+
public void markAsTableModelEvent() {
if (event instanceof PipeInsertionEvent) {
((PipeInsertionEvent) event).markAsTableModelEvent();
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 4df39838cbe..1a8138672b9 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
@@ -78,7 +78,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
}
private void extractTabletInsertion(final PipeRealtimeEvent event) {
- if (canNotUseTabletAnyMore()) {
+ if (canNotUseTabletAnyMore(event)) {
event
.getTsFileEpoch()
.migrateState(
@@ -201,7 +201,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
}
}
- private boolean canNotUseTabletAnyMore() {
+ private boolean canNotUseTabletAnyMore(final PipeRealtimeEvent event) {
// In the following 7 cases, we should not extract any more tablet events. all the data
// represented by the tablet events should be carried by the following tsfile event:
// 0. If the pipe task is currently restarted.
@@ -212,19 +212,19 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
// 4. The number of realtime tsfile events to transfer has exceeded the limit.
// 5. The number of linked tsfiles has reached the dangerous threshold.
// 6. The shallow memory usage of the insert node has reached the dangerous threshold.
- return isPipeTaskCurrentlyRestarted()
- || mayWalSizeReachThrottleThreshold()
- || mayMemTablePinnedCountReachDangerousThreshold()
- || isHistoricalTsFileEventCountExceededLimit()
- || isRealtimeTsFileEventCountExceededLimit()
- || mayTsFileLinkedCountReachDangerousThreshold()
- || mayInsertNodeMemoryReachDangerousThreshold();
+ return isPipeTaskCurrentlyRestarted(event)
+ || mayWalSizeReachThrottleThreshold(event)
+ || mayMemTablePinnedCountReachDangerousThreshold(event)
+ || isHistoricalTsFileEventCountExceededLimit(event)
+ || isRealtimeTsFileEventCountExceededLimit(event)
+ || mayTsFileLinkedCountReachDangerousThreshold(event)
+ || mayInsertNodeMemoryReachDangerousThreshold(event);
}
- private boolean isPipeTaskCurrentlyRestarted() {
+ private boolean isPipeTaskCurrentlyRestarted(final PipeRealtimeEvent event) {
final boolean isPipeTaskCurrentlyRestarted =
PipeDataNodeAgent.task().isPipeTaskCurrentlyRestarted(pipeName);
- if (isPipeTaskCurrentlyRestarted) {
+ if (isPipeTaskCurrentlyRestarted && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore1: Pipe task is currently restarted",
pipeName,
@@ -233,11 +233,11 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return isPipeTaskCurrentlyRestarted;
}
- private boolean mayWalSizeReachThrottleThreshold() {
+ private boolean mayWalSizeReachThrottleThreshold(final PipeRealtimeEvent event) {
final boolean mayWalSizeReachThrottleThreshold =
3 * WALManager.getInstance().getTotalDiskUsage()
> IoTDBDescriptor.getInstance().getConfig().getThrottleThreshold();
- if (mayWalSizeReachThrottleThreshold) {
+ if (mayWalSizeReachThrottleThreshold && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore2: Wal size {} has reached throttle threshold {}",
pipeName,
@@ -248,11 +248,11 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return mayWalSizeReachThrottleThreshold;
}
- private boolean mayMemTablePinnedCountReachDangerousThreshold() {
+ private boolean mayMemTablePinnedCountReachDangerousThreshold(final PipeRealtimeEvent event) {
final boolean mayMemTablePinnedCountReachDangerousThreshold =
PipeDataNodeResourceManager.wal().getPinnedWalCount()
>= PipeConfig.getInstance().getPipeMaxAllowedPinnedMemTableCount();
- if (mayMemTablePinnedCountReachDangerousThreshold) {
+ if (mayMemTablePinnedCountReachDangerousThreshold && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore3: The number of pinned memtables {} has reached the dangerous threshold {}",
pipeName,
@@ -263,14 +263,14 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return mayMemTablePinnedCountReachDangerousThreshold;
}
- private boolean isHistoricalTsFileEventCountExceededLimit() {
+ private boolean isHistoricalTsFileEventCountExceededLimit(final PipeRealtimeEvent event) {
final IoTDBDataRegionExtractor extractor =
PipeDataRegionExtractorMetrics.getInstance().getExtractorMap().get(getTaskID());
final boolean isHistoricalTsFileEventCountExceededLimit =
Objects.nonNull(extractor)
&& extractor.getHistoricalTsFileInsertionEventCount()
>= PipeConfig.getInstance().getPipeMaxAllowedHistoricalTsFilePerDataRegion();
- if (isHistoricalTsFileEventCountExceededLimit) {
+ if (isHistoricalTsFileEventCountExceededLimit && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore4: The number of historical tsFile events {} has exceeded the limit {}",
pipeName,
@@ -281,11 +281,11 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return isHistoricalTsFileEventCountExceededLimit;
}
- private boolean isRealtimeTsFileEventCountExceededLimit() {
+ private boolean isRealtimeTsFileEventCountExceededLimit(final PipeRealtimeEvent event) {
final boolean isRealtimeTsFileEventCountExceededLimit =
pendingQueue.getTsFileInsertionEventCount()
>= PipeConfig.getInstance().getPipeMaxAllowedPendingTsFileEpochPerDataRegion();
- if (isRealtimeTsFileEventCountExceededLimit) {
+ if (isRealtimeTsFileEventCountExceededLimit && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore5: The number of realtime tsFile events {} has exceeded the limit {}",
pipeName,
@@ -296,11 +296,11 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return isRealtimeTsFileEventCountExceededLimit;
}
- private boolean mayTsFileLinkedCountReachDangerousThreshold() {
+ private boolean mayTsFileLinkedCountReachDangerousThreshold(final PipeRealtimeEvent event) {
final boolean mayTsFileLinkedCountReachDangerousThreshold =
PipeDataNodeResourceManager.tsfile().getLinkedTsfileCount()
>= PipeConfig.getInstance().getPipeMaxAllowedLinkedTsFileCount();
- if (mayTsFileLinkedCountReachDangerousThreshold) {
+ if (mayTsFileLinkedCountReachDangerousThreshold && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore6: The number of linked tsfiles {} has reached the dangerous threshold {}",
pipeName,
@@ -311,7 +311,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return mayTsFileLinkedCountReachDangerousThreshold;
}
- private boolean mayInsertNodeMemoryReachDangerousThreshold() {
+ private boolean mayInsertNodeMemoryReachDangerousThreshold(final PipeRealtimeEvent event) {
final long floatingMemoryUsageInByte =
PipeDataNodeAgent.task().getFloatingMemoryUsageInByte(pipeName);
final long pipeCount = PipeDataNodeAgent.task().getPipeCount();
@@ -319,7 +319,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
PipeDataNodeResourceManager.memory().getFreeMemorySizeInBytes();
final boolean mayInsertNodeMemoryReachDangerousThreshold =
3 * floatingMemoryUsageInByte * pipeCount >= 2 * freeMemorySizeInBytes;
- if (mayInsertNodeMemoryReachDangerousThreshold) {
+ if (mayInsertNodeMemoryReachDangerousThreshold && event.mayExtractorUseTablets(this)) {
LOGGER.info(
"Pipe task {}@{} canNotUseTabletAnyMore7: The shallow memory usage of the insert node {} has reached the dangerous threshold {}",
pipeName,
@@ -379,7 +379,7 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio
return state;
}
- return canNotUseTabletAnyMore()
+ return canNotUseTabletAnyMore(event)
? TsFileEpoch.State.USING_TSFILE
: TsFileEpoch.State.USING_TABLET;
});