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 35ffb5128db [IOTDB-6135] Pipe: Fix a bug which keeps generating
PipeHeartbeatEvent unnecessarily (#11008)
35ffb5128db is described below
commit 35ffb5128db06d10ed1715199b217c217c60a42c
Author: 马子坤 <[email protected]>
AuthorDate: Mon Sep 4 10:21:45 2023 +0800
[IOTDB-6135] Pipe: Fix a bug which keeps generating PipeHeartbeatEvent
unnecessarily (#11008)
---
.../extractor/realtime/PipeRealtimeDataRegionHybridExtractor.java | 4 +++-
.../pipe/extractor/realtime/PipeRealtimeDataRegionLogExtractor.java | 4 +++-
.../extractor/realtime/PipeRealtimeDataRegionTsFileExtractor.java | 4 +++-
3 files changed, 9 insertions(+), 3 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 edbc14e26d4..cbd6d77f189 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,7 +155,9 @@ public class PipeRealtimeDataRegionHybridExtractor extends
PipeRealtimeDataRegio
}
private void extractHeartbeat(PipeRealtimeEvent event) {
- if (pendingQueue.peekLast() instanceof PipeHeartbeatEvent) {
+ Event lastEvent = pendingQueue.peekLast();
+ if (lastEvent instanceof PipeRealtimeEvent
+ && ((PipeRealtimeEvent) lastEvent).getEvent() 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());
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 fb3783dc224..52970f2957d 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,7 +75,9 @@ public class PipeRealtimeDataRegionLogExtractor extends
PipeRealtimeDataRegionEx
}
private void extractHeartbeat(PipeRealtimeEvent event) {
- if (pendingQueue.peekLast() instanceof PipeHeartbeatEvent) {
+ Event lastEvent = pendingQueue.peekLast();
+ if (lastEvent instanceof PipeRealtimeEvent
+ && ((PipeRealtimeEvent) lastEvent).getEvent() 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());
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 5967b1bc87b..816e0249b47 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,7 +75,9 @@ public class PipeRealtimeDataRegionTsFileExtractor extends
PipeRealtimeDataRegio
}
private void extractHeartbeat(PipeRealtimeEvent event) {
- if (pendingQueue.peekLast() instanceof PipeHeartbeatEvent) {
+ Event lastEvent = pendingQueue.peekLast();
+ if (lastEvent instanceof PipeRealtimeEvent
+ && ((PipeRealtimeEvent) lastEvent).getEvent() 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());
