You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ba...@apache.org on 2025/03/07 07:17:48 UTC
(pulsar) 07/15: [fix][broker] fix broker identifying incorrect stuck topic (#24006)
This is an automated email from the ASF dual-hosted git repository.
baodi pushed a commit to branch branch-3.0.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dd8decaede636f8788bb2fb626563fa0f9c4e8aa
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Thu Feb 20 01:46:47 2025 -0800
[fix][broker] fix broker identifying incorrect stuck topic (#24006)
(cherry picked from commit 28f7845c93edc0dcbc068a71bec45ab4b3402d4e)
(cherry picked from commit 95101f301421c047f32f0f2b44c2b5e92ca80562)
---
.../persistent/PersistentDispatcherMultipleConsumers.java | 2 +-
.../PersistentDispatcherSingleActiveConsumer.java | 4 ++--
.../broker/service/persistent/PersistentTopicTest.java | 14 +++++++++++---
3 files changed, 14 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index ae844b57844..245ccaf2ee3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -1227,7 +1227,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return false;
}
// consider dispatch is stuck if : dispatcher has backlog, available-permits and there is no pending read
- if (totalAvailablePermits > 0 && !havePendingReplayRead && !havePendingRead
+ if (isAtleastOneConsumerAvailable() && !havePendingReplayRead && !havePendingRead
&& cursor.getNumberOfEntriesInBacklog(false) > 0) {
log.warn("{}-{} Dispatcher is stuck and unblocking by issuing reads", topic.getName(), name);
readMoreEntries();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index a138fde7a72..e1b17044b9b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -600,9 +600,9 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
if (consumer == null || cursor.checkAndUpdateReadPositionChanged()) {
return false;
}
- int totalAvailablePermits = consumer.getAvailablePermits();
+ boolean isConsumerAvailable = !consumer.isBlocked() && consumer.getAvailablePermits() > 0;
// consider dispatch is stuck if : dispatcher has backlog, available-permits and there is no pending read
- if (totalAvailablePermits > 0 && !havePendingRead && cursor.getNumberOfEntriesInBacklog(false) > 0) {
+ if (isConsumerAvailable && !havePendingRead && cursor.getNumberOfEntriesInBacklog(false) > 0) {
log.warn("{}-{} Dispatcher is stuck and unblocking by issuing reads", topic.getName(), name);
readMoreEntries(consumer);
return true;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
index a1fded261d1..c2097762228 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
@@ -18,7 +18,8 @@
*/
package org.apache.pulsar.broker.service.persistent;
-import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.Metric;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.parseMetrics;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
@@ -35,8 +36,6 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.Sets;
import java.io.ByteArrayOutputStream;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
@@ -74,6 +73,7 @@ import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.TopicPoliciesService;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.Metric;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
@@ -102,6 +102,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+
@Slf4j
@Test(groups = "broker")
public class PersistentTopicTest extends BrokerTestBase {
@@ -216,6 +217,13 @@ public class PersistentTopicTest extends BrokerTestBase {
assertNotNull(msg);
msg = consumer2.receive(5, TimeUnit.SECONDS);
assertNotNull(msg);
+
+ org.apache.pulsar.broker.service.Consumer sharedConsumer = sharedDispatcher.getConsumers().get(0);
+ Field blockField = org.apache.pulsar.broker.service.Consumer.class.getDeclaredField("blockedConsumerOnUnackedMsgs");
+ blockField.setAccessible(true);
+ blockField.set(sharedConsumer, true);
+ producer.newMessage().value("test").eventTime(5).send();
+ assertFalse(sharedSub.checkAndUnblockIfStuck());
}
@Test