This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 7d0565a6ad3 Fix metadata cache and pipe edge cases (#18147) (#18156)
7d0565a6ad3 is described below
commit 7d0565a6ad3442a693b38ed350c50e683b09e50c
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 9 10:12:13 2026 +0800
Fix metadata cache and pipe edge cases (#18147) (#18156)
(cherry picked from commit 7558aa7ef8012f94a082dbb51b55ef2906f60b75)
---
.../realtime/PipeRealtimeDataRegionSource.java | 6 +-
.../db/pipe/source/PipeRealtimeExtractTest.java | 76 ++++++++++++++++++++++
2 files changed, 81 insertions(+), 1 deletion(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index 367c8327421..fa716d5dcca 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -380,7 +380,10 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
if (lastEvent == null || !(lastEvent.getEvent() instanceof
PipeHeartbeatEvent)) {
break;
}
- pendingQueue.pollLast();
+ final PipeRealtimeEvent droppedEvent = (PipeRealtimeEvent)
pendingQueue.pollLast();
+ if (droppedEvent != null) {
+
droppedEvent.decreaseReferenceCount(PipeRealtimeDataRegionSource.class.getName(),
false);
+ }
}
final Event last = pendingQueue.peekLast();
if (last instanceof PipeRealtimeEvent
@@ -391,6 +394,7 @@ public abstract class PipeRealtimeDataRegionSource
implements PipeExtractor {
oldEvent
.getProgressIndex()
.updateToMinimumEqualOrIsAfterProgressIndex(event.getProgressIndex()));
+
event.decreaseReferenceCount(PipeRealtimeDataRegionSource.class.getName(),
false);
return;
}
pendingQueue.offer(event);
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
index 2cba4c8eebc..8cf5eaa8185 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
@@ -26,8 +26,11 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
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.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionHybridSource;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionLogSource;
import
org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
@@ -71,6 +74,7 @@ import java.util.function.Function;
public class PipeRealtimeExtractTest {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeRealtimeExtractTest.class);
+ private static final String TEST_REFERENCE_HOLDER =
PipeRealtimeExtractTest.class.getName();
private final String dataRegion1 = "1";
private final String dataRegion2 = "2";
@@ -317,6 +321,37 @@ public class PipeRealtimeExtractTest {
}
}
+ @Test
+ public void testProgressReportExtractionReleasesDroppedEvents() {
+ final TestRealtimeDataRegionSource source = new
TestRealtimeDataRegionSource();
+
+ final PipeRealtimeEvent heartbeatEvent =
+ PipeRealtimeEventFactory.createRealtimeEvent(dataRegion1, false);
+ heartbeatEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
+ source.extractHeartbeatForTest(heartbeatEvent);
+ Assert.assertEquals(1, heartbeatEvent.getEvent().getReferenceCount());
+
+ final PipeRealtimeEvent progressEvent =
createProgressReportRealtimeEvent();
+ progressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
+ source.extractProgressReportEventForTest(progressEvent);
+
+ Assert.assertEquals(0, heartbeatEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, progressEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, source.getEventCount());
+
+ final PipeRealtimeEvent mergedProgressEvent =
createProgressReportRealtimeEvent();
+ mergedProgressEvent.increaseReferenceCount(TEST_REFERENCE_HOLDER);
+ source.extractProgressReportEventForTest(mergedProgressEvent);
+
+ Assert.assertEquals(0, mergedProgressEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, progressEvent.getEvent().getReferenceCount());
+ Assert.assertEquals(1, source.getEventCount());
+
+ final Event queuedEvent = source.pollForTest();
+ Assert.assertSame(progressEvent, queuedEvent);
+ ((EnrichedEvent) queuedEvent).clearReferenceCount(TEST_REFERENCE_HOLDER);
+ }
+
private Future<?> write2DataRegion(
final int writeNum, final String dataRegionId, final int startNum) {
final File dataRegionDir =
@@ -401,6 +436,47 @@ public class PipeRealtimeExtractTest {
});
}
+ private PipeRealtimeEvent createProgressReportRealtimeEvent() {
+ final ProgressReportEvent progressReportEvent = new
ProgressReportEvent(null, 0, null);
+ progressReportEvent.bindProgressIndex(MinimumProgressIndex.INSTANCE);
+ return PipeRealtimeEventFactory.createRealtimeEvent(progressReportEvent);
+ }
+
+ private static class TestRealtimeDataRegionSource extends
PipeRealtimeDataRegionSource {
+
+ private void extractHeartbeatForTest(final PipeRealtimeEvent event) {
+ extractHeartbeat(event);
+ }
+
+ private void extractProgressReportEventForTest(final PipeRealtimeEvent
event) {
+ extractProgressReportEvent(event);
+ }
+
+ private Event pollForTest() {
+ return pendingQueue.directPoll();
+ }
+
+ @Override
+ protected void doExtract(final PipeRealtimeEvent event) {
+ // Not needed in this reference-counting unit test.
+ }
+
+ @Override
+ public Event supply() {
+ return pendingQueue.directPoll();
+ }
+
+ @Override
+ public boolean isNeedListenToTsFile() {
+ return false;
+ }
+
+ @Override
+ public boolean isNeedListenToInsertNode() {
+ return false;
+ }
+ }
+
private static class NoTsFileRealtimeDataRegionSource extends
PipeRealtimeDataRegionSource {
private final AtomicInteger observedTsFileEventCount = new
AtomicInteger(0);
