This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 8f1bed8246c [Subscription] Consolidate poll queue fixes (#18771)
8f1bed8246c is described below
commit 8f1bed8246cc03426435839b2c829a9759deed74
Author: Caideyipi <[email protected]>
AuthorDate: Wed Sep 30 10:37:24 2026 +0800
[Subscription] Consolidate poll queue fixes (#18771)
---
.../db/subscription/broker/SubscriptionBroker.java | 7 +-
.../broker/SubscriptionPrefetchingQueue.java | 92 +++++++++++++++++-----
.../broker/SubscriptionPrefetchingQueueStates.java | 28 +++++--
3 files changed, 98 insertions(+), 29 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
index 3b1dec89dd1..3cfb34014d7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java
@@ -129,10 +129,13 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
// Iterate over each sorted topic name and poll the corresponding events
int remainingTopicSize = sortedTopicNames.size();
for (final String topicName : sortedTopicNames) {
- remainingTopicSize -= 1;
// Check pipe-based queue
final SubscriptionPrefetchingQueue prefetchingQueue =
topicNameToPrefetchingQueue.get(topicName);
+ // Include the current topic when splitting the remaining receiver
deadline. Otherwise, the
+ // first topic receives too large a share and later topics are polled
after it has mostly
+ // elapsed.
+ final int currentTopicSize = remainingTopicSize--;
// Recheck
if (Objects.isNull(prefetchingQueue) || prefetchingQueue.isClosed()) {
@@ -146,7 +149,7 @@ public class SubscriptionBroker implements
ISubscriptionBroker {
final PollTimer timer =
new PollTimer(
System.currentTimeMillis(),
- SubscriptionAgent.receiver().remainingMs() / Math.max(1,
remainingTopicSize));
+ SubscriptionAgent.receiver().remainingMs() / Math.max(1,
currentTopicSize));
event = prefetchingQueue.pollV2(consumerId, timer);
} else {
// TODO: migrate poll to pollV2
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 b760ee62f6f..716e7d90f15 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
@@ -265,31 +265,35 @@ public abstract class SubscriptionPrefetchingQueue {
}
public SubscriptionEvent pollV2(final String consumerId, final PollTimer
timer) {
- acquireReadLock();
- try {
- return isClosed() ? null : pollInternalV2(consumerId, timer);
- } finally {
- releaseReadLock();
- }
- }
-
- private SubscriptionEvent pollInternalV2(final String consumerId, final
PollTimer timer) {
states.markPollRequest();
- // do-while ensures at least one poll
- do {
- SubscriptionEvent event;
+ boolean firstPoll = true;
+ while (firstPoll || timer.notExpired()) {
+ firstPoll = false;
+
+ // Keep the read lock only for the queue and in-flight event transition.
Waiting for more
+ // data happens below, outside the lock, so cleanup can acquire the
write lock promptly.
+ acquireReadLock();
try {
- if (prefetchingQueue.isEmpty()) {
- // TODO: concurrent polling of multiple prefetching queues
- Thread.sleep(100);
- onEvent();
+ if (isClosed()) {
+ return null;
}
- event = pollPrefetchedEvent(consumerId);
+ final SubscriptionEvent event = pollInternalV2(consumerId);
if (Objects.nonNull(event)) {
return event;
}
+ } finally {
+ releaseReadLock();
+ }
+
+ timer.update();
+ if (timer.isExpired()) {
+ break;
+ }
+
+ try {
+ Thread.sleep(Math.min(100L, timer.remainingMs()));
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
LOGGER.warn(
@@ -297,9 +301,52 @@ public abstract class SubscriptionPrefetchingQueue {
.PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_INTERRUPTED_WHILE_F8923826,
this,
e);
+ break;
}
timer.update();
- } while (!timer.isExpired());
+ }
+
+ return null;
+ }
+
+ private SubscriptionEvent pollInternalV2(final String consumerId) {
+ if (prefetchingQueue.isEmpty()) {
+ onEvent();
+ }
+
+ final long size = prefetchingQueue.size();
+ long count = 0;
+
+ SubscriptionEvent event;
+ while (count++ < size && Objects.nonNull(event = prefetchingQueue.poll()))
{
+ if (event.isCommitted()) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_COMMITTED_8684FF17,
+ this,
+ event);
+ // no need to update inFlightEvents
+ continue;
+ }
+
+ if (!event.pollable()) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_NON_POLLABLE_644D5D6B,
+ this,
+ event);
+ nackAndRequeue(event);
+ continue;
+ }
+
+ // This operation should be performed before updating inFlightEvents to
prevent multiple
+ // consumers from consuming the same event.
+ event.recordLastPolledTimestamp(); // now non-pollable
+
+ inFlightEvents.put(new Pair<>(consumerId, event.getCommitContext()),
event);
+ event.recordLastPolledConsumerId(consumerId);
+ return event;
+ }
return null;
}
@@ -332,8 +379,7 @@ public abstract class SubscriptionPrefetchingQueue {
.PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_NON_POLLABLE_644D5D6B,
this,
event);
- event.nack(); // now pollable
- // no need to update inFlightEvents and prefetchingQueue
+ nackAndRequeue(event);
continue;
}
@@ -349,6 +395,12 @@ public abstract class SubscriptionPrefetchingQueue {
return null;
}
+ private void nackAndRequeue(final SubscriptionEvent event) {
+ event.nack(); // now pollable
+ // The event has already been removed from this queue, so put it back
after resetting its state.
+ prefetchingQueue.offer(event);
+ }
+
/////////////////////////////// prefetch ///////////////////////////////
public boolean executePrefetch() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
index 70eacd2eaaa..a9167b7b475 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueStates.java
@@ -25,11 +25,12 @@ import
org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.metrics.core.utils.IoTDBMovingAverage;
import com.codahale.metrics.Clock;
-import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.atomic.AtomicLong;
+
import static com.google.common.base.MoreObjects.toStringHelper;
/**
@@ -58,7 +59,7 @@ public class SubscriptionPrefetchingQueueStates {
private volatile long lastPollRequestTimestamp;
private final Meter pollRequestMeter;
private final Meter missingPrefechMeter;
- private final Counter disorderCauseCounter; // TODO: use meter
+ private final AtomicLong disorderCauseCounter;
public SubscriptionPrefetchingQueueStates(final SubscriptionPrefetchingQueue
prefetchingQueue) {
this.prefetchingQueue = prefetchingQueue;
@@ -66,7 +67,7 @@ public class SubscriptionPrefetchingQueueStates {
this.lastPollRequestTimestamp = -1;
this.pollRequestMeter = new Meter(new IoTDBMovingAverage(),
Clock.defaultClock());
this.missingPrefechMeter = new Meter(new IoTDBMovingAverage(),
Clock.defaultClock());
- this.disorderCauseCounter = new Counter();
+ this.disorderCauseCounter = new AtomicLong();
}
public void markPollRequest() {
@@ -79,7 +80,9 @@ public class SubscriptionPrefetchingQueueStates {
}
public void markDisorderCause() {
- disorderCauseCounter.inc();
+ // Disable prefetching for the current sequence. The guard is cleared
after the prefetched and
+ // in-flight events have drained so a single disorder does not permanently
disable prefetching.
+ disorderCauseCounter.incrementAndGet();
}
private double pollRate() {
@@ -117,7 +120,14 @@ public class SubscriptionPrefetchingQueueStates {
// 1.5. disorder history
if (hasDisorderCause()) {
- return false;
+ // Keep prefetching disabled while events from the disordered sequence
are still visible to
+ // consumers. Once both queues are drained, the next sequence starts
from a clean ordering
+ // boundary and prefetching can recover.
+ if (prefetchingQueue.getPrefetchedEventCount() > 0
+ || prefetchingQueue.getSubscriptionUncommittedEventCount() > 0) {
+ return false;
+ }
+ clearDisorderCause();
}
// 2. permissive conditions
@@ -155,7 +165,11 @@ public class SubscriptionPrefetchingQueueStates {
}
private boolean hasDisorderCause() {
- return disorderCauseCounter.getCount() > 0;
+ return disorderCauseCounter.get() > 0;
+ }
+
+ private void clearDisorderCause() {
+ disorderCauseCounter.set(0);
}
private static boolean isApproximatelyZero(final double value) {
@@ -168,7 +182,7 @@ public class SubscriptionPrefetchingQueueStates {
.add("lastPollRequestTimestamp", lastPollRequestTimestamp)
.add("pollRate", pollRate())
.add("missingRate", missingRate())
- .add("disorderCause", disorderCauseCounter.getCount())
+ .add("disorderCause", disorderCauseCounter.get())
.toString();
}
}
