Pipe: fix calculations of PipeDataNodeRemainingEventAndTimeOperator (#…

…13876) (#13889)
apache · Oct 23, 2024 · cb0d2a5 · cb0d2a5
1 parent 58e3a69
commit cb0d2a5
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 14 deletions.
diff --git a/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/...java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java
@@ -136,7 +136,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
       PipeDataNodeResourceManager.wal().pin(walEntryHandler);
       if (Objects.nonNull(pipeName)) {
         PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-            .increaseInsertionEventCount(pipeName + "_" + creationTime);
+            .increaseTabletEventCount(pipeName + "_" + creationTime);
       }
       return true;
     } catch (final Exception e) {
@@ -169,7 +169,7 @@ public boolean internallyDecreaseResourceReferenceCount(final String holderMessa
     } finally {
       if (Objects.nonNull(pipeName)) {
         PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-            .decreaseInsertionEventCount(pipeName + "_" + creationTime);
+            .decreaseTabletEventCount(pipeName + "_" + creationTime);
       }
     }
   }

diff --git a/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java b/...c/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java
@@ -118,7 +118,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
                 PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet));
     if (Objects.nonNull(pipeName)) {
       PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-          .increaseInsertionEventCount(pipeName + "_" + creationTime);
+          .increaseTabletEventCount(pipeName + "_" + creationTime);
     }
     return true;
   }
@@ -127,7 +127,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa
   public boolean internallyDecreaseResourceReferenceCount(final String holderMessage) {
     if (Objects.nonNull(pipeName)) {
       PipeDataNodeRemainingEventAndTimeMetrics.getInstance()
-          .decreaseInsertionEventCount(pipeName + "_" + creationTime);
+          .decreaseTabletEventCount(pipeName + "_" + creationTime);
     }
     allocatedMemoryBlock.close();
 

diff --git a/...c/main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeMetrics.java b/...c/main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeMetrics.java
@@ -129,16 +129,16 @@ public void register(final IoTDBSchemaRegionExtractor extractor) {
     }
   }
 
-  public void increaseInsertionEventCount(final String pipeID) {
+  public void increaseTabletEventCount(final String pipeID) {
     remainingEventAndTimeOperatorMap
         .computeIfAbsent(pipeID, k -> new PipeDataNodeRemainingEventAndTimeOperator())
-        .increaseInsertionEventCount();
+        .increaseTabletEventCount();
   }
 
-  public void decreaseInsertionEventCount(final String pipeID) {
+  public void decreaseTabletEventCount(final String pipeID) {
     remainingEventAndTimeOperatorMap
         .computeIfAbsent(pipeID, k -> new PipeDataNodeRemainingEventAndTimeOperator())
-        .decreaseInsertionEventCount();
+        .decreaseTabletEventCount();
   }
 
   public void increaseTsFileEventCount(final String pipeID) {

diff --git a/.../main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeOperator.java b/.../main/java/org/apache/iotdb/db/pipe/metric/PipeDataNodeRemainingEventAndTimeOperator.java
@@ -44,7 +44,7 @@ class PipeDataNodeRemainingEventAndTimeOperator extends PipeRemainingOperator {
   private final Set<IoTDBSchemaRegionExtractor> schemaRegionExtractors =
       Collections.newSetFromMap(new ConcurrentHashMap<>());
 
-  private final AtomicInteger insertionEventCount = new AtomicInteger(0);
+  private final AtomicInteger tabletEventCount = new AtomicInteger(0);
   private final AtomicInteger tsfileEventCount = new AtomicInteger(0);
   private final AtomicInteger heartbeatEventCount = new AtomicInteger(0);
 
@@ -58,12 +58,12 @@ class PipeDataNodeRemainingEventAndTimeOperator extends PipeRemainingOperator {
 
   //////////////////////////// Remaining event & time calculation ////////////////////////////
 
-  void increaseInsertionEventCount() {
-    tsfileEventCount.incrementAndGet();
+  void increaseTabletEventCount() {
+    tabletEventCount.incrementAndGet();
   }
 
-  void decreaseInsertionEventCount() {
-    tsfileEventCount.decrementAndGet();
+  void decreaseTabletEventCount() {
+    tabletEventCount.decrementAndGet();
   }
 
   void increaseTsFileEventCount() {
@@ -84,6 +84,7 @@ void decreaseHeartbeatEventCount() {
 
   long getRemainingEvents() {
     return tsfileEventCount.get()
+        + tabletEventCount.get()
         + heartbeatEventCount.get()
         + schemaRegionExtractors.stream()
             .map(IoTDBSchemaRegionExtractor::getUnTransferredEventCount)
@@ -105,7 +106,7 @@ long getRemainingEvents() {
     final double invocationValue = collectInvocationHistogram.getMean();
     // Do not take heartbeat event into account
     final double totalDataRegionWriteEventCount =
-        tsfileEventCount.get() * Math.max(invocationValue, 1) + insertionEventCount.get();
+        tsfileEventCount.get() * Math.max(invocationValue, 1) + tabletEventCount.get();
 
     dataRegionCommitMeter.updateAndGet(
         meter -> {