You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2025/01/25 07:01:40 UTC
(iotdb) 05/31: Pipe: report linked tsfile size & Subscription: decrease reference count for other enriched events & add logging to observe possible stuck situations (#14668)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c38997c8b9963f29b0a1b9b298da79211150f5ab
Author: VGalaxies <vg...@apache.org>
AuthorDate: Fri Jan 10 17:31:26 2025 +0800
Pipe: report linked tsfile size & Subscription: decrease reference count for other enriched events & add logging to observe possible stuck situations (#14668)
(cherry picked from commit a5e13fd8f7c1624d20f1628405bcca92c9900ed5)
---
.../resource/tsfile/PipeTsFileResourceManager.java | 5 ++++-
.../broker/SubscriptionPrefetchingQueue.java | 2 ++
.../db/subscription/event/SubscriptionEvent.java | 21 +++++++++++++--------
.../batch/SubscriptionPipeTabletEventBatch.java | 22 ++++++++++++++++++++++
4 files changed, 41 insertions(+), 9 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
index 9607c41a94b..2bdfda64ef4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
@@ -98,7 +98,10 @@ public class PipeTsFileResourceManager {
} else {
logBuilder.append(
String.format(
- "<%s , %d times> ", entry.getKey(), entry.getValue().getReferenceCount()));
+ "<%s , %d times, %d bytes> ",
+ entry.getKey(),
+ entry.getValue().getReferenceCount(),
+ entry.getValue().getFileSize()));
}
} catch (final IOException e) {
LOGGER.warn("failed to close PipeTsFileResource when checking TTL: ", e);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
index 20a0d0983f0..5afbf212829 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java
@@ -357,6 +357,8 @@ public abstract class SubscriptionPrefetchingQueue {
"Subscription: SubscriptionPrefetchingQueue {} ignore EnrichedEvent {} when prefetching.",
this,
event);
+ ((EnrichedEvent) event)
+ .decreaseReferenceCount(SubscriptionPrefetchingQueue.class.getName(), false);
if (onEvent()) {
return;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java
index 97fa621faef..2db5bc7a4b0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java
@@ -65,6 +65,9 @@ public class SubscriptionEvent {
// record file name for file payload
private String fileName;
+ private static final long NACK_COUNT_REPORT_THRESHOLD = 3;
+ private final AtomicLong nackCount = new AtomicLong();
+
/**
* Constructs a {@link SubscriptionEvent} with the response type of {@link
* SubscriptionEventSingleResponse}.
@@ -143,11 +146,11 @@ public class SubscriptionEvent {
}
public void ack(final Consumer<SubscriptionEvent> onCommittedHook) {
- // ack response
- response.ack(onCommittedHook);
-
- // ack pipe events
+ // NOTE: we should ack pipe events before ack response since multiple events may reuse the same
+ // batch (as pipe events)
+ // TODO: consider more elegant design for this method
pipeEvents.ack();
+ response.ack(onCommittedHook);
}
/**
@@ -156,11 +159,8 @@ public class SubscriptionEvent {
* SubscriptionPrefetchingQueue} or {@link SubscriptionPrefetchingQueue#cleanUp}.
*/
public void cleanUp() {
- // reset serialized responses
- response.cleanUp();
-
- // clean up pipe events
pipeEvents.cleanUp();
+ response.cleanUp();
// TODO: clean more fields
}
@@ -216,6 +216,11 @@ public class SubscriptionEvent {
// reset lastPolledTimestamp makes this event pollable
lastPolledTimestamp.set(INVALID_TIMESTAMP);
+
+ // record nack count
+ if (nackCount.getAndIncrement() > NACK_COUNT_REPORT_THRESHOLD) {
+ LOGGER.warn("{} has been nacked {} times", this, nackCount);
+ }
}
public void recordLastPolledConsumerId(final String consumerId) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
index 59528125f49..1031e1e2279 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/batch/SubscriptionPipeTabletEventBatch.java
@@ -43,6 +43,7 @@ import java.util.Iterator;
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 class SubscriptionPipeTabletEventBatch extends SubscriptionPipeEventBatch
@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 =