ITERATED_COUNT_REPORT_FREQ

apache · Jan 10, 2025 · 78422c3 · 78422c3
1 parent db27e28
commit 78422c3
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 2 deletions.
diff --git a/...core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java b/...core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java
@@ -65,7 +65,7 @@ public class SubscriptionEvent {
   // record file name for file payload
   private String fileName;
 
-  private static final long NACK_COUNT_THRESHOLD = 3;
+  private static final long NACK_COUNT_REPORT_THRESHOLD = 3;
   private final AtomicLong nackCount = new AtomicLong();
 
   /**
@@ -218,7 +218,7 @@ public void nack() {
     lastPolledTimestamp.set(INVALID_TIMESTAMP);
 
     // record nack count
-    if (nackCount.getAndIncrement() > NACK_COUNT_THRESHOLD) {
+    if (nackCount.getAndIncrement() > NACK_COUNT_REPORT_THRESHOLD) {
       LOGGER.warn("{} has been nacked {} times", this, nackCount);
     }
   }

diff --git a/...n/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java b/...n/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
@@ -43,6 +43,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
 
 public class SubscriptionPipeTabletEventBatch extends SubscriptionPipeEventBatch
     implements Iterator<List<Tablet>> {
@@ -62,6 +63,10 @@ public class SubscriptionPipeTabletEventBatch extends SubscriptionPipeEventBatch
 
   private final List<EnrichedEvent> iteratedEnrichedEvents = new ArrayList<>();
 
+  private static final long ITERATED_COUNT_REPORT_FREQ =
+      30000; // based on the full parse of a 128MB tsfile estimate
+  private final AtomicLong iteratedCount = new AtomicLong();
+
   public SubscriptionPipeTabletEventBatch(
       final int regionId,
       final SubscriptionPrefetchingTabletQueue prefetchingQueue,
@@ -230,6 +235,23 @@ public boolean hasNext() {
 
   @Override
   public List<Tablet> next() {
+    final List<Tablet> tablets = nextInternal();
+    if (Objects.isNull(tablets)) {
+      return null;
+    }
+    if (iteratedCount.incrementAndGet() % ITERATED_COUNT_REPORT_FREQ == 0) {
+      LOGGER.info(
+          "{} has been iterated {} times, current TsFileInsertionEvent {}",
+          this,
+          iteratedCount,
+          Objects.isNull(currentTsFileInsertionEvent)
+              ? "<unknown>"
+              : ((EnrichedEvent) currentTsFileInsertionEvent).coreReportMessage());
+    }
+    return tablets;
+  }
+
+  private List<Tablet> nextInternal() {
     if (Objects.nonNull(currentTabletInsertionEventsIterator)) {
       if (currentTabletInsertionEventsIterator.hasNext()) {
         final TabletInsertionEvent tabletInsertionEvent =