This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch pipe-he-reduce in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5059ebb0617afb39baa160b655e44ffeafe787b5 Author: Steve Yurong Su <[email protected]> AuthorDate: Mon Aug 28 23:24:25 2023 +0800 Pipe: drop HeartbeatEvent when pipe is stopped to avoid OOM (#10977) (cherry picked from commit f1a4012424fef30509ed13e33e1b641ffc5d4ba9) --- .../realtime/PipeRealtimeDataRegionHybridExtractor.java | 9 ++++++++- .../realtime/PipeRealtimeDataRegionLogExtractor.java | 9 ++++++++- .../realtime/PipeRealtimeDataRegionTsFileExtractor.java | 7 +++++++ .../iotdb/db/pipe/task/connection/BlockingPendingQueue.java | 2 +- .../iotdb/db/pipe/task/connection/PipeEventCollector.java | 13 ++++++++++--- .../pipe/task/connection/UnboundedBlockingPendingQueue.java | 12 ++++++++++-- 6 files changed, 44 insertions(+), 8 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java index b9e8f275e28..edbc14e26d4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java @@ -155,11 +155,18 @@ public class PipeRealtimeDataRegionHybridExtractor extends PipeRealtimeDataRegio } private void extractHeartbeat(PipeRealtimeEvent event) { + if (pendingQueue.peekLast() instanceof PipeHeartbeatEvent) { + // if the last event in the pending queue is a heartbeat event, we should not extract any more + // heartbeat events to avoid OOM when the pipe is stopped. + event.decreaseReferenceCount(PipeRealtimeDataRegionHybridExtractor.class.getName()); + return; + } + if (!pendingQueue.waitedOffer(event)) { // this would not happen, but just in case. // pendingQueue is unbounded, so it should never reach capacity. LOGGER.error( - "extract: pending queue of PipeRealtimeDataRegionTsFileExtractor {} " + "extract: pending queue of PipeRealtimeDataRegionHybridExtractor {} " + "has reached capacity, discard heartbeat event {}", this, event); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java index 14e899300b8..fb3783dc224 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java @@ -75,11 +75,18 @@ public class PipeRealtimeDataRegionLogExtractor extends PipeRealtimeDataRegionEx } private void extractHeartbeat(PipeRealtimeEvent event) { + if (pendingQueue.peekLast() instanceof PipeHeartbeatEvent) { + // if the last event in the pending queue is a heartbeat event, we should not extract any more + // heartbeat events to avoid OOM when the pipe is stopped. + event.decreaseReferenceCount(PipeRealtimeDataRegionLogExtractor.class.getName()); + return; + } + if (!pendingQueue.waitedOffer(event)) { // this would not happen, but just in case. // pendingQueue is unbounded, so it should never reach capacity. LOGGER.error( - "extract: pending queue of PipeRealtimeDataRegionTsFileExtractor {} " + "extract: pending queue of PipeRealtimeDataRegionLogExtractor {} " + "has reached capacity, discard heartbeat event {}", this, event); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java index 9546e35906d..5967b1bc87b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java @@ -75,6 +75,13 @@ public class PipeRealtimeDataRegionTsFileExtractor extends PipeRealtimeDataRegio } private void extractHeartbeat(PipeRealtimeEvent event) { + if (pendingQueue.peekLast() instanceof PipeHeartbeatEvent) { + // if the last event in the pending queue is a heartbeat event, we should not extract any more + // heartbeat events to avoid OOM when the pipe is stopped. + event.decreaseReferenceCount(PipeRealtimeDataRegionTsFileExtractor.class.getName()); + return; + } + if (!pendingQueue.waitedOffer(event)) { // This would not happen, but just in case. // Pending is unbounded, so it should never reach capacity. diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java index f81d301b6f2..fc7e2cd6f5c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/BlockingPendingQueue.java @@ -35,7 +35,7 @@ public abstract class BlockingPendingQueue<E extends Event> { private static final long MAX_BLOCKING_TIME_MS = PipeConfig.getInstance().getPipeSubtaskExecutorPendingQueueMaxBlockingTimeMs(); - private final BlockingQueue<E> pendingQueue; + protected final BlockingQueue<E> pendingQueue; protected BlockingPendingQueue(BlockingQueue<E> pendingQueue) { this.pendingQueue = pendingQueue; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java index bf57908c71b..772851305ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/PipeEventCollector.java @@ -20,17 +20,18 @@ package org.apache.iotdb.db.pipe.task.connection; import org.apache.iotdb.db.pipe.event.EnrichedEvent; +import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.pipe.api.collector.EventCollector; import org.apache.iotdb.pipe.api.event.Event; +import java.util.Deque; import java.util.LinkedList; -import java.util.Queue; public class PipeEventCollector implements EventCollector { private final BoundedBlockingPendingQueue<Event> pendingQueue; - private final Queue<Event> bufferQueue; + private final Deque<Event> bufferQueue; public PipeEventCollector(BoundedBlockingPendingQueue<Event> pendingQueue) { this.pendingQueue = pendingQueue; @@ -50,7 +51,13 @@ public class PipeEventCollector implements EventCollector { if (pendingQueue.waitedOffer(bufferedEvent)) { bufferQueue.poll(); } else { - bufferQueue.offer(event); + // We can NOT keep too many PipeHeartbeatEvent in bufferQueue because they may cause OOM. + if (event instanceof PipeHeartbeatEvent + && bufferQueue.peekLast() instanceof PipeHeartbeatEvent) { + ((EnrichedEvent) event).decreaseReferenceCount(PipeEventCollector.class.getName()); + } else { + bufferQueue.offer(event); + } return; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java index dafb567e902..343621bbb4a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/connection/UnboundedBlockingPendingQueue.java @@ -21,11 +21,19 @@ package org.apache.iotdb.db.pipe.task.connection; import org.apache.iotdb.pipe.api.event.Event; -import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.BlockingDeque; +import java.util.concurrent.LinkedBlockingDeque; public class UnboundedBlockingPendingQueue<E extends Event> extends BlockingPendingQueue<E> { + private final BlockingDeque<E> pendingDeque; + public UnboundedBlockingPendingQueue() { - super(new LinkedBlockingQueue<>()); + super(new LinkedBlockingDeque<>()); + pendingDeque = (BlockingDeque<E>) pendingQueue; + } + + public E peekLast() { + return pendingDeque.peekLast(); } }
