You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/09/24 15:14:24 UTC
(pulsar) branch branch-3.0 updated (e1bd313e5be -> 0fd9f1746ba)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from e1bd313e5be [fix] Key_Shared mode consumption latency when low traffic (#23340)
new ca75892004a Revert "[fix] Key_Shared mode consumption latency when low traffic (#23340)"
new 9b9e7a8ef4d Revert "[fix][broker] Fix retry backoff for PersistentDispatcherMultipleConsumers (#23284)"
new 0fd9f1746ba Revert "[improve][broker] Reschedule reads with increasing backoff when no messages are dispatched (#23226)"
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
conf/broker.conf | 10 --
conf/standalone.conf | 10 --
.../apache/pulsar/broker/ServiceConfiguration.java | 14 --
.../PersistentDispatcherMultipleConsumers.java | 74 +++------
...istentStickyKeyDispatcherMultipleConsumers.java | 3 -
.../broker/auth/MockedPulsarServiceBaseTest.java | 3 -
...ntStickyKeyDispatcherMultipleConsumersTest.java | 177 +--------------------
.../broker/transaction/TransactionTestBase.java | 3 -
8 files changed, 25 insertions(+), 269 deletions(-)
(pulsar) 01/03: Revert "[fix] Key_Shared mode consumption latency when low traffic (#23340)"
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ca75892004ab9c38b9ca2f2fc382b7f4052bfd33
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Tue Sep 24 18:01:45 2024 +0300
Revert "[fix] Key_Shared mode consumption latency when low traffic (#23340)"
This reverts commit e1bd313e5beb56d23193e697d498e85d83a9750d.
---
conf/broker.conf | 4 ++--
conf/standalone.conf | 4 ++--
.../main/java/org/apache/pulsar/broker/ServiceConfiguration.java | 4 ++--
.../service/persistent/PersistentDispatcherMultipleConsumers.java | 6 ++----
4 files changed, 8 insertions(+), 10 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 81cb00b29bc..7df126ce56d 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -465,12 +465,12 @@ dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
# delay. This parameter sets the initial backoff delay in milliseconds.
-dispatcherRetryBackoffInitialTimeInMs=1
+dispatcherRetryBackoffInitialTimeInMs=100
# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
# delay. This parameter sets the maximum backoff delay in milliseconds.
-dispatcherRetryBackoffMaxTimeInMs=10
+dispatcherRetryBackoffMaxTimeInMs=1000
# Precise dispathcer flow control according to history message number of each entry
preciseDispatcherFlowControl=false
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 66bb5cc1d63..ec2c87369e2 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -281,12 +281,12 @@ dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
# delay. This parameter sets the initial backoff delay in milliseconds.
-dispatcherRetryBackoffInitialTimeInMs=1
+dispatcherRetryBackoffInitialTimeInMs=100
# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
# delay. This parameter sets the maximum backoff delay in milliseconds.
-dispatcherRetryBackoffMaxTimeInMs=10
+dispatcherRetryBackoffMaxTimeInMs=1000
# Precise dispathcer flow control according to history message number of each entry
preciseDispatcherFlowControl=false
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index ce6e9acaea9..0068a7f1107 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1177,14 +1177,14 @@ public class ServiceConfiguration implements PulsarConfiguration {
doc = "On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered "
+ "out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff "
+ "delay. This parameter sets the initial backoff delay in milliseconds.")
- private int dispatcherRetryBackoffInitialTimeInMs = 1;
+ private int dispatcherRetryBackoffInitialTimeInMs = 100;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered "
+ "out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff "
+ "delay. This parameter sets the maximum backoff delay in milliseconds.")
- private int dispatcherRetryBackoffMaxTimeInMs = 10;
+ private int dispatcherRetryBackoffMaxTimeInMs = 1000;
@FieldContext(
dynamic = true,
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 ba3cc280a34..cd5acd069e7 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
@@ -649,12 +649,10 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
boolean triggerReadingMore = sendMessagesToConsumers(readType, entries, needAcquireSendInProgress);
int entriesDispatched = lastNumberOfEntriesDispatched;
updatePendingBytesToDispatch(-totalBytesSize);
- if (entriesDispatched > 0) {
- // Reset the backoff when we successfully dispatched messages
- retryBackoff.reset();
- }
if (triggerReadingMore) {
if (entriesDispatched > 0) {
+ // Reset the backoff when we successfully dispatched messages
+ retryBackoff.reset();
// Call readMoreEntries in the same thread to trigger the next read
readMoreEntries();
} else if (entriesDispatched == 0) {
(pulsar) 02/03: Revert "[fix][broker] Fix retry backoff for PersistentDispatcherMultipleConsumers (#23284)"
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9b9e7a8ef4d31dc6fde6482b5dfe0ba928bc9ae9
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Tue Sep 24 18:02:04 2024 +0300
Revert "[fix][broker] Fix retry backoff for PersistentDispatcherMultipleConsumers (#23284)"
This reverts commit d9598b501b234a986e93a6609c5e62f6d05bd234.
---
.../PersistentDispatcherMultipleConsumers.java | 19 ++-
...ntStickyKeyDispatcherMultipleConsumersTest.java | 140 ++++-----------------
2 files changed, 30 insertions(+), 129 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 cd5acd069e7..23c4cdd84c2 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
@@ -406,15 +406,12 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Reschedule message read in {} ms", topic.getName(), name, readAfterMs);
}
- Runnable runnable = () -> {
- isRescheduleReadInProgress.set(false);
- readMoreEntries();
- };
- if (readAfterMs > 0) {
- topic.getBrokerService().executor().schedule(runnable, readAfterMs, TimeUnit.MILLISECONDS);
- } else {
- topic.getBrokerService().executor().execute(runnable);
- }
+ topic.getBrokerService().executor().schedule(
+ () -> {
+ isRescheduleReadInProgress.set(false);
+ readMoreEntries();
+ },
+ readAfterMs, TimeUnit.MILLISECONDS);
}
}
@@ -798,7 +795,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
totalBytesSent += sendMessageInfo.getTotalBytes();
}
- lastNumberOfEntriesDispatched = (int) totalEntries;
acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
if (entriesToDispatch > 0) {
@@ -811,8 +807,9 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
entry.release();
});
- }
+ lastNumberOfEntriesDispatched = entriesToDispatch;
+ }
return true;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
index f7326734eaa..f6431b38fb6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
@@ -98,8 +98,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
final String topicName = "persistent://public/default/testTopic";
final String subscriptionName = "testSubscription";
private AtomicInteger consumerMockAvailablePermits;
- int retryBackoffInitialTimeInMs = 10;
- int retryBackoffMaxTimeInMs = 50;
@BeforeMethod
public void setup() throws Exception {
@@ -109,8 +107,8 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
doReturn(true).when(configMock).isSubscriptionKeySharedUseConsistentHashing();
doReturn(1).when(configMock).getSubscriptionKeySharedConsistentHashingReplicaPoints();
doReturn(true).when(configMock).isDispatcherDispatchMessagesInSubscriptionThread();
- doAnswer(invocation -> retryBackoffInitialTimeInMs).when(configMock).getDispatcherRetryBackoffInitialTimeInMs();
- doAnswer(invocation -> retryBackoffMaxTimeInMs).when(configMock).getDispatcherRetryBackoffMaxTimeInMs();
+ doReturn(10).when(configMock).getDispatcherRetryBackoffInitialTimeInMs();
+ doReturn(50).when(configMock).getDispatcherRetryBackoffMaxTimeInMs();
pulsarMock = mock(PulsarService.class);
doReturn(configMock).when(pulsarMock).getConfiguration();
@@ -461,45 +459,34 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
allEntries.forEach(entry -> entry.release());
}
- @DataProvider(name = "testBackoffDelayWhenNoMessagesDispatched")
- private Object[][] testBackoffDelayWhenNoMessagesDispatchedParams() {
- return new Object[][] { { false, true }, { true, true }, { true, false }, { false, false } };
+ @DataProvider(name = "dispatchMessagesInSubscriptionThread")
+ private Object[][] dispatchMessagesInSubscriptionThread() {
+ return new Object[][] { { false }, { true } };
}
- @Test(dataProvider = "testBackoffDelayWhenNoMessagesDispatched")
- public void testBackoffDelayWhenNoMessagesDispatched(boolean dispatchMessagesInSubscriptionThread, boolean isKeyShared)
+ @Test(dataProvider = "dispatchMessagesInSubscriptionThread")
+ public void testBackoffDelayWhenNoMessagesDispatched(boolean dispatchMessagesInSubscriptionThread)
throws Exception {
persistentDispatcher.close();
List<Long> retryDelays = new CopyOnWriteArrayList<>();
doReturn(dispatchMessagesInSubscriptionThread).when(configMock).isDispatcherDispatchMessagesInSubscriptionThread();
-
- PersistentDispatcherMultipleConsumers dispatcher;
- if (isKeyShared) {
- dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
- topicMock, cursorMock, subscriptionMock, configMock,
- new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT)) {
- @Override
- protected void reScheduleReadInMs(long readAfterMs) {
- retryDelays.add(readAfterMs);
- }
- };
- } else {
- dispatcher = new PersistentDispatcherMultipleConsumers(topicMock, cursorMock, subscriptionMock) {
- @Override
- protected void reScheduleReadInMs(long readAfterMs) {
- retryDelays.add(readAfterMs);
- }
- };
- }
+ persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT)) {
+ @Override
+ protected void reScheduleReadInMs(long readAfterMs) {
+ retryDelays.add(readAfterMs);
+ }
+ };
// add a consumer without permits to trigger the retry behavior
consumerMockAvailablePermits.set(0);
- dispatcher.addConsumer(consumerMock);
+ persistentDispatcher.addConsumer(consumerMock);
// call "readEntriesComplete" directly to test the retry behavior
List<Entry> entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 1);
assertEquals(retryDelays.get(0), 10, "Initial retry delay should be 10ms");
@@ -507,7 +494,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
);
// test the second retry delay
entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 2);
double delay = retryDelays.get(1);
@@ -517,7 +504,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// verify the max retry delay
for (int i = 0; i < 100; i++) {
entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
}
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 102);
@@ -528,14 +515,14 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
// unblock to check that the retry delay is reset
consumerMockAvailablePermits.set(1000);
entries = List.of(EntryImpl.create(1, 2, createMessage("message2", 1, "key2")));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
// wait that the possibly async handling has completed
- Awaitility.await().untilAsserted(() -> assertFalse(dispatcher.isSendInProgress()));
+ Awaitility.await().untilAsserted(() -> assertFalse(persistentDispatcher.isSendInProgress()));
// now block again to check the next retry delay so verify it was reset
consumerMockAvailablePermits.set(0);
entries = List.of(EntryImpl.create(1, 3, createMessage("message3", 1, "key3")));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
+ persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
Awaitility.await().untilAsserted(() -> {
assertEquals(retryDelays.size(), 103);
assertEquals(retryDelays.get(0), 10, "Resetted retry delay should be 10ms");
@@ -543,89 +530,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
);
}
- @Test(dataProvider = "testBackoffDelayWhenNoMessagesDispatched")
- public void testBackoffDelayWhenRetryDelayDisabled(boolean dispatchMessagesInSubscriptionThread, boolean isKeyShared)
- throws Exception {
- persistentDispatcher.close();
-
- // it should be possible to disable the retry delay
- // by setting retryBackoffInitialTimeInMs and retryBackoffMaxTimeInMs to 0
- retryBackoffInitialTimeInMs=0;
- retryBackoffMaxTimeInMs=0;
-
- List<Long> retryDelays = new CopyOnWriteArrayList<>();
- doReturn(dispatchMessagesInSubscriptionThread).when(configMock)
- .isDispatcherDispatchMessagesInSubscriptionThread();
-
- PersistentDispatcherMultipleConsumers dispatcher;
- if (isKeyShared) {
- dispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
- topicMock, cursorMock, subscriptionMock, configMock,
- new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT)) {
- @Override
- protected void reScheduleReadInMs(long readAfterMs) {
- retryDelays.add(readAfterMs);
- }
- };
- } else {
- dispatcher = new PersistentDispatcherMultipleConsumers(topicMock, cursorMock, subscriptionMock) {
- @Override
- protected void reScheduleReadInMs(long readAfterMs) {
- retryDelays.add(readAfterMs);
- }
- };
- }
-
- // add a consumer without permits to trigger the retry behavior
- consumerMockAvailablePermits.set(0);
- dispatcher.addConsumer(consumerMock);
-
- // call "readEntriesComplete" directly to test the retry behavior
- List<Entry> entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 1);
- assertEquals(retryDelays.get(0), 0, "Initial retry delay should be 0ms");
- }
- );
- // test the second retry delay
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 2);
- double delay = retryDelays.get(1);
- assertEquals(delay, 0, 0, "Second retry delay should be 0ms");
- }
- );
- // verify the max retry delay
- for (int i = 0; i < 100; i++) {
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- }
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 102);
- double delay = retryDelays.get(101);
- assertEquals(delay, 0, 0, "Max delay should be 0ms");
- }
- );
- // unblock to check that the retry delay is reset
- consumerMockAvailablePermits.set(1000);
- entries = List.of(EntryImpl.create(1, 2, createMessage("message2", 1, "key2")));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- // wait that the possibly async handling has completed
- Awaitility.await().untilAsserted(() -> assertFalse(dispatcher.isSendInProgress()));
-
- // now block again to check the next retry delay so verify it was reset
- consumerMockAvailablePermits.set(0);
- entries = List.of(EntryImpl.create(1, 3, createMessage("message3", 1, "key3")));
- dispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 103);
- assertEquals(retryDelays.get(0), 0, "Resetted retry delay should be 0ms");
- }
- );
- }
-
private ByteBuf createMessage(String message, int sequenceId) {
return createMessage(message, sequenceId, "testKey");
}
(pulsar) 03/03: Revert "[improve][broker] Reschedule reads with increasing backoff when no messages are dispatched (#23226)"
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0fd9f1746baedd3825e17a33f7d4b9aadd0a3526
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Tue Sep 24 18:03:02 2024 +0300
Revert "[improve][broker] Reschedule reads with increasing backoff when no messages are dispatched (#23226)"
This reverts commit f990e28c8bf1abfbbf5539758156fffd578232ef.
---
conf/broker.conf | 10 ---
conf/standalone.conf | 10 ---
.../apache/pulsar/broker/ServiceConfiguration.java | 14 ----
.../PersistentDispatcherMultipleConsumers.java | 59 +++++-----------
...istentStickyKeyDispatcherMultipleConsumers.java | 3 -
.../broker/auth/MockedPulsarServiceBaseTest.java | 3 -
...ntStickyKeyDispatcherMultipleConsumersTest.java | 81 +---------------------
.../broker/transaction/TransactionTestBase.java | 3 -
8 files changed, 20 insertions(+), 163 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 7df126ce56d..3689721e267 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -462,16 +462,6 @@ dispatcherReadFailureBackoffMaxTimeInMs=60000
# The read failure backoff mandatory stop time in milliseconds. By default it is 0s.
dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
-# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
-# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
-# delay. This parameter sets the initial backoff delay in milliseconds.
-dispatcherRetryBackoffInitialTimeInMs=100
-
-# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
-# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
-# delay. This parameter sets the maximum backoff delay in milliseconds.
-dispatcherRetryBackoffMaxTimeInMs=1000
-
# Precise dispathcer flow control according to history message number of each entry
preciseDispatcherFlowControl=false
diff --git a/conf/standalone.conf b/conf/standalone.conf
index ec2c87369e2..da8c9759960 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -278,16 +278,6 @@ dispatcherReadFailureBackoffMaxTimeInMs=60000
# The read failure backoff mandatory stop time in milliseconds. By default it is 0s.
dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
-# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
-# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
-# delay. This parameter sets the initial backoff delay in milliseconds.
-dispatcherRetryBackoffInitialTimeInMs=100
-
-# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
-# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
-# delay. This parameter sets the maximum backoff delay in milliseconds.
-dispatcherRetryBackoffMaxTimeInMs=1000
-
# Precise dispathcer flow control according to history message number of each entry
preciseDispatcherFlowControl=false
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 0068a7f1107..ff401fdb2b7 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1172,20 +1172,6 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private int dispatcherReadFailureBackoffMandatoryStopTimeInMs = 0;
- @FieldContext(
- category = CATEGORY_POLICIES,
- doc = "On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered "
- + "out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff "
- + "delay. This parameter sets the initial backoff delay in milliseconds.")
- private int dispatcherRetryBackoffInitialTimeInMs = 100;
-
- @FieldContext(
- category = CATEGORY_POLICIES,
- doc = "On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered "
- + "out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff "
- + "delay. This parameter sets the maximum backoff delay in milliseconds.")
- private int dispatcherRetryBackoffMaxTimeInMs = 1000;
-
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
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 23c4cdd84c2..ae844b57844 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
@@ -47,7 +47,6 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.TooManyRequestsExcep
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory;
import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker;
import org.apache.pulsar.broker.delayed.DelayedDeliveryTrackerFactory;
@@ -85,6 +84,7 @@ import org.slf4j.LoggerFactory;
*/
public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMultipleConsumers
implements Dispatcher, ReadEntriesCallback {
+
protected final PersistentTopic topic;
protected final ManagedCursor cursor;
protected volatile Range<PositionImpl> lastIndividualDeletedRangeFromCursorRecovery;
@@ -122,8 +122,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
private AtomicBoolean isRescheduleReadInProgress = new AtomicBoolean(false);
protected final ExecutorService dispatchMessagesThread;
private final SharedConsumerAssignor assignor;
- protected int lastNumberOfEntriesDispatched;
- private final Backoff retryBackoff;
+
protected enum ReadType {
Normal, Replay
}
@@ -148,15 +147,10 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
this.readBatchSize = serviceConfig.getDispatcherMaxReadBatchSize();
this.initializeDispatchRateLimiterIfNeeded();
this.assignor = new SharedConsumerAssignor(this::getNextConsumer, this::addMessageToReplay);
- ServiceConfiguration serviceConfiguration = topic.getBrokerService().pulsar().getConfiguration();
this.readFailureBackoff = new Backoff(
- serviceConfiguration.getDispatcherReadFailureBackoffInitialTimeInMs(),
+ topic.getBrokerService().pulsar().getConfiguration().getDispatcherReadFailureBackoffInitialTimeInMs(),
TimeUnit.MILLISECONDS,
1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
- retryBackoff = new Backoff(
- serviceConfiguration.getDispatcherRetryBackoffInitialTimeInMs(), TimeUnit.MILLISECONDS,
- serviceConfiguration.getDispatcherRetryBackoffMaxTimeInMs(), TimeUnit.MILLISECONDS,
- 0, TimeUnit.MILLISECONDS);
}
@Override
@@ -398,20 +392,16 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
@Override
protected void reScheduleRead() {
- reScheduleReadInMs(MESSAGE_RATE_BACKOFF_MS);
- }
-
- protected void reScheduleReadInMs(long readAfterMs) {
if (isRescheduleReadInProgress.compareAndSet(false, true)) {
if (log.isDebugEnabled()) {
- log.debug("[{}] [{}] Reschedule message read in {} ms", topic.getName(), name, readAfterMs);
+ log.debug("[{}] [{}] Reschedule message read in {} ms", topic.getName(), name, MESSAGE_RATE_BACKOFF_MS);
}
topic.getBrokerService().executor().schedule(
() -> {
isRescheduleReadInProgress.set(false);
readMoreEntries();
},
- readAfterMs, TimeUnit.MILLISECONDS);
+ MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
}
}
@@ -622,8 +612,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
log.debug("[{}] Distributing {} messages to {} consumers", name, entries.size(), consumerList.size());
}
- long totalBytesSize = entries.stream().mapToLong(Entry::getLength).sum();
- updatePendingBytesToDispatch(totalBytesSize);
+ long size = entries.stream().mapToLong(Entry::getLength).sum();
+ updatePendingBytesToDispatch(size);
// dispatch messages to a separate thread, but still in order for this subscription
// sendMessagesToConsumers is responsible for running broker-side filters
@@ -633,28 +623,19 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
// in a separate thread, and we want to prevent more reads
acquireSendInProgress();
dispatchMessagesThread.execute(() -> {
- handleSendingMessagesAndReadingMore(readType, entries, false, totalBytesSize);
+ if (sendMessagesToConsumers(readType, entries, false)) {
+ updatePendingBytesToDispatch(-size);
+ readMoreEntries();
+ } else {
+ updatePendingBytesToDispatch(-size);
+ }
});
} else {
- handleSendingMessagesAndReadingMore(readType, entries, true, totalBytesSize);
- }
- }
-
- private synchronized void handleSendingMessagesAndReadingMore(ReadType readType, List<Entry> entries,
- boolean needAcquireSendInProgress,
- long totalBytesSize) {
- boolean triggerReadingMore = sendMessagesToConsumers(readType, entries, needAcquireSendInProgress);
- int entriesDispatched = lastNumberOfEntriesDispatched;
- updatePendingBytesToDispatch(-totalBytesSize);
- if (triggerReadingMore) {
- if (entriesDispatched > 0) {
- // Reset the backoff when we successfully dispatched messages
- retryBackoff.reset();
- // Call readMoreEntries in the same thread to trigger the next read
- readMoreEntries();
- } else if (entriesDispatched == 0) {
- // If no messages were dispatched, we need to reschedule a new read with an increasing backoff delay
- reScheduleReadInMs(retryBackoff.next());
+ if (sendMessagesToConsumers(readType, entries, true)) {
+ updatePendingBytesToDispatch(-size);
+ readMoreEntriesAsync();
+ } else {
+ updatePendingBytesToDispatch(-size);
}
}
}
@@ -693,7 +674,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
if (needTrimAckedMessages()) {
cursor.trimDeletedEntries(entries);
}
- lastNumberOfEntriesDispatched = 0;
int entriesToDispatch = entries.size();
// Trigger read more messages
@@ -807,8 +787,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
entry.release();
});
-
- lastNumberOfEntriesDispatched = entriesToDispatch;
}
return true;
}
@@ -871,7 +849,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
totalBytesSent += sendMessageInfo.getTotalBytes();
}
- lastNumberOfEntriesDispatched = (int) totalEntries;
acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
return numConsumers.get() == 0; // trigger a new readMoreEntries() call
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 397cb7226b7..2df9f38531f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -178,7 +178,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
@Override
protected synchronized boolean trySendMessagesToConsumers(ReadType readType, List<Entry> entries) {
- lastNumberOfEntriesDispatched = 0;
long totalMessagesSent = 0;
long totalBytesSent = 0;
long totalEntries = 0;
@@ -313,8 +312,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
}
}
-
- lastNumberOfEntriesDispatched = (int) totalEntries;
// acquire message-dispatch permits for already delivered messages
acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index e7ec93a5d6a..df017eaf3a8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -234,9 +234,6 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
this.conf.setWebServicePort(Optional.of(0));
this.conf.setNumExecutorThreadPoolSize(5);
this.conf.setExposeBundlesMetricsInPrometheus(true);
- // Disable the dispatcher retry backoff in tests by default
- this.conf.setDispatcherRetryBackoffInitialTimeInMs(0);
- this.conf.setDispatcherRetryBackoffMaxTimeInMs(0);
}
protected final void init() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
index f6431b38fb6..7e1b5f8c71e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
@@ -35,7 +35,6 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import io.netty.buffer.ByteBuf;
@@ -49,7 +48,6 @@ import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -77,7 +75,6 @@ import org.awaitility.Awaitility;
import org.mockito.ArgumentCaptor;
import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
-import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker")
@@ -97,7 +94,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
final String topicName = "persistent://public/default/testTopic";
final String subscriptionName = "testSubscription";
- private AtomicInteger consumerMockAvailablePermits;
@BeforeMethod
public void setup() throws Exception {
@@ -107,8 +103,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
doReturn(true).when(configMock).isSubscriptionKeySharedUseConsistentHashing();
doReturn(1).when(configMock).getSubscriptionKeySharedConsistentHashingReplicaPoints();
doReturn(true).when(configMock).isDispatcherDispatchMessagesInSubscriptionThread();
- doReturn(10).when(configMock).getDispatcherRetryBackoffInitialTimeInMs();
- doReturn(50).when(configMock).getDispatcherRetryBackoffMaxTimeInMs();
+
pulsarMock = mock(PulsarService.class);
doReturn(configMock).when(pulsarMock).getConfiguration();
@@ -140,8 +135,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
consumerMock = mock(Consumer.class);
channelMock = mock(ChannelPromise.class);
doReturn("consumer1").when(consumerMock).consumerName();
- consumerMockAvailablePermits = new AtomicInteger(1000);
- doAnswer(invocation -> consumerMockAvailablePermits.get()).when(consumerMock).getAvailablePermits();
+ doReturn(1000).when(consumerMock).getAvailablePermits();
doReturn(true).when(consumerMock).isWritable();
doReturn(channelMock).when(consumerMock).sendMessages(
anyList(),
@@ -459,77 +453,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
allEntries.forEach(entry -> entry.release());
}
- @DataProvider(name = "dispatchMessagesInSubscriptionThread")
- private Object[][] dispatchMessagesInSubscriptionThread() {
- return new Object[][] { { false }, { true } };
- }
-
- @Test(dataProvider = "dispatchMessagesInSubscriptionThread")
- public void testBackoffDelayWhenNoMessagesDispatched(boolean dispatchMessagesInSubscriptionThread)
- throws Exception {
- persistentDispatcher.close();
-
- List<Long> retryDelays = new CopyOnWriteArrayList<>();
- doReturn(dispatchMessagesInSubscriptionThread).when(configMock).isDispatcherDispatchMessagesInSubscriptionThread();
- persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
- topicMock, cursorMock, subscriptionMock, configMock,
- new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT)) {
- @Override
- protected void reScheduleReadInMs(long readAfterMs) {
- retryDelays.add(readAfterMs);
- }
- };
-
- // add a consumer without permits to trigger the retry behavior
- consumerMockAvailablePermits.set(0);
- persistentDispatcher.addConsumer(consumerMock);
-
- // call "readEntriesComplete" directly to test the retry behavior
- List<Entry> entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 1);
- assertEquals(retryDelays.get(0), 10, "Initial retry delay should be 10ms");
- }
- );
- // test the second retry delay
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 2);
- double delay = retryDelays.get(1);
- assertEquals(delay, 20.0, 2.0, "Second retry delay should be 20ms (jitter <-10%)");
- }
- );
- // verify the max retry delay
- for (int i = 0; i < 100; i++) {
- entries = List.of(EntryImpl.create(1, 1, createMessage("message1", 1)));
- persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- }
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 102);
- double delay = retryDelays.get(101);
- assertEquals(delay, 50.0, 5.0, "Max delay should be 50ms (jitter <-10%)");
- }
- );
- // unblock to check that the retry delay is reset
- consumerMockAvailablePermits.set(1000);
- entries = List.of(EntryImpl.create(1, 2, createMessage("message2", 1, "key2")));
- persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- // wait that the possibly async handling has completed
- Awaitility.await().untilAsserted(() -> assertFalse(persistentDispatcher.isSendInProgress()));
-
- // now block again to check the next retry delay so verify it was reset
- consumerMockAvailablePermits.set(0);
- entries = List.of(EntryImpl.create(1, 3, createMessage("message3", 1, "key3")));
- persistentDispatcher.readEntriesComplete(entries, PersistentDispatcherMultipleConsumers.ReadType.Normal);
- Awaitility.await().untilAsserted(() -> {
- assertEquals(retryDelays.size(), 103);
- assertEquals(retryDelays.get(0), 10, "Resetted retry delay should be 10ms");
- }
- );
- }
-
private ByteBuf createMessage(String message, int sequenceId) {
return createMessage(message, sequenceId, "testKey");
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java
index 26f1686db67..3a83d2f95fe 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTestBase.java
@@ -162,9 +162,6 @@ public abstract class TransactionTestBase extends TestRetrySupport {
conf.setBrokerDeduplicationEnabled(true);
conf.setTransactionBufferSnapshotMaxTransactionCount(2);
conf.setTransactionBufferSnapshotMinTimeInMillis(2000);
- // Disable the dispatcher retry backoff in tests by default
- conf.setDispatcherRetryBackoffInitialTimeInMs(0);
- conf.setDispatcherRetryBackoffMaxTimeInMs(0);
serviceConfigurationList.add(conf);
PulsarTestContext.Builder testContextBuilder =