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 2025/01/02 08:23:08 UTC
(pulsar) 02/02: Msg delivery is stuck due to items in the collection recentlyJoinedConsumers are out-of-order (#23795)
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 ca535a28d685227134cd3aec6e82de70d04eb836
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Jan 2 15:54:34 2025 +0800
Msg delivery is stuck due to items in the collection recentlyJoinedConsumers are out-of-order (#23795)
(cherry picked from commit 4a01423273c010eebce549139b2fc400bb59555e)
---
...istentStickyKeyDispatcherMultipleConsumers.java | 40 ++++++++++-
...ntStickyKeyDispatcherMultipleConsumersTest.java | 81 ++++++++++++++++++++++
2 files changed, 119 insertions(+), 2 deletions(-)
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 2df9f38531f..27c9eed9709 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
@@ -134,11 +134,44 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
&& consumerList.size() > 1
&& cursor.getNumberOfEntriesSinceFirstNotAckedMessage() > 1) {
recentlyJoinedConsumers.put(consumer, readPositionWhenJoining);
+ sortRecentlyJoinedConsumersIfNeeded();
}
}
});
}
+ private void sortRecentlyJoinedConsumersIfNeeded() {
+ if (recentlyJoinedConsumers.size() == 1) {
+ return;
+ }
+ boolean sortNeeded = false;
+ PositionImpl posPre = null;
+ PositionImpl posAfter = null;
+ for (Map.Entry<Consumer, PositionImpl> entry : recentlyJoinedConsumers.entrySet()) {
+ if (posPre == null) {
+ posPre = entry.getValue();
+ } else {
+ posAfter = entry.getValue();
+ }
+ if (posPre != null && posAfter != null) {
+ if (posPre.compareTo(posAfter) > 0) {
+ sortNeeded = true;
+ break;
+ }
+ posPre = posAfter;
+ }
+ }
+
+ if (sortNeeded) {
+ List<Map.Entry<Consumer, PositionImpl>> sortedList = new ArrayList<>(recentlyJoinedConsumers.entrySet());
+ Collections.sort(sortedList, Map.Entry.comparingByValue());
+ recentlyJoinedConsumers.clear();
+ for (Map.Entry<Consumer, PositionImpl> entry : sortedList) {
+ recentlyJoinedConsumers.put(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+
@Override
public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
// The consumer must be removed from the selector before calling the superclass removeConsumer method.
@@ -548,8 +581,11 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
&& ksm.isAllowOutOfOrderDelivery() == this.allowOutOfOrderDelivery);
}
- public LinkedHashMap<Consumer, PositionImpl> getRecentlyJoinedConsumers() {
- return recentlyJoinedConsumers;
+ public synchronized LinkedHashMap<Consumer, PositionImpl> getRecentlyJoinedConsumers() {
+ if (recentlyJoinedConsumers == null) {
+ return null;
+ }
+ return new LinkedHashMap<>(recentlyJoinedConsumers);
}
public Map<Consumer, List<Range>> getConsumerKeyHashRanges() {
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 f21efb277ba..bf3a79133b0 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
@@ -45,7 +45,9 @@ import io.netty.channel.EventLoopGroup;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
@@ -171,6 +173,85 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
assertTrue(persistentDispatcher.getSelector().getConsumerKeyHashRanges().isEmpty());
}
+ @Test
+ public void testSortRecentlyJoinedConsumersIfNeeded() throws Exception {
+ PersistentStickyKeyDispatcherMultipleConsumers persistentDispatcher =
+ new PersistentStickyKeyDispatcherMultipleConsumers(
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT));
+
+ Consumer consumer0 = mock(Consumer.class);
+ when(consumer0.consumerName()).thenReturn("c0-1");
+ Consumer consumer1 = mock(Consumer.class);
+ when(consumer1.consumerName()).thenReturn("c1");
+ Consumer consumer2 = mock(Consumer.class);
+ when(consumer2.consumerName()).thenReturn("c2");
+ Consumer consumer3 = mock(Consumer.class);
+ when(consumer3.consumerName()).thenReturn("c3");
+ Consumer consumer4 = mock(Consumer.class);
+ when(consumer4.consumerName()).thenReturn("c4");
+ Consumer consumer5 = mock(Consumer.class);
+ when(consumer5.consumerName()).thenReturn("c5");
+ Consumer consumer6 = mock(Consumer.class);
+ when(consumer6.consumerName()).thenReturn("c6");
+
+ when(cursorMock.getNumberOfEntriesSinceFirstNotAckedMessage()).thenReturn(100L);
+ when(cursorMock.getMarkDeletedPosition()).thenReturn(PositionImpl.get(-1, -1));
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(0, 0));
+ persistentDispatcher.addConsumer(consumer0).join();
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(4, 1));
+ persistentDispatcher.addConsumer(consumer1).join();
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(5, 2));
+ persistentDispatcher.addConsumer(consumer2).join();
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(5, 1));
+ persistentDispatcher.addConsumer(consumer3).join();
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(5, 3));
+ persistentDispatcher.addConsumer(consumer4).join();
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(4, 2));
+ persistentDispatcher.addConsumer(consumer5).join();
+
+ when(cursorMock.getReadPosition()).thenReturn(PositionImpl.get(6, 1));
+ persistentDispatcher.addConsumer(consumer6).join();
+
+ assertEquals(persistentDispatcher.getRecentlyJoinedConsumers().size(), 6);
+
+ Iterator<Map.Entry<Consumer, PositionImpl>> itr
+ = persistentDispatcher.getRecentlyJoinedConsumers().entrySet().iterator();
+
+ Map.Entry<Consumer, PositionImpl> entry1 = itr.next();
+ assertEquals(entry1.getValue(), PositionImpl.get(4, 1));
+ assertEquals(entry1.getKey(), consumer1);
+
+ Map.Entry<Consumer, PositionImpl> entry2 = itr.next();
+ assertEquals(entry2.getValue(), PositionImpl.get(4, 2));
+ assertEquals(entry2.getKey(), consumer5);
+
+ Map.Entry<Consumer, PositionImpl> entry3 = itr.next();
+ assertEquals(entry3.getValue(), PositionImpl.get(5, 1));
+ assertEquals(entry3.getKey(), consumer3);
+
+ Map.Entry<Consumer, PositionImpl> entry4 = itr.next();
+ assertEquals(entry4.getValue(), PositionImpl.get(5, 2));
+ assertEquals(entry4.getKey(), consumer2);
+
+ Map.Entry<Consumer, PositionImpl> entry5 = itr.next();
+ assertEquals(entry5.getValue(), PositionImpl.get(5, 3));
+ assertEquals(entry5.getKey(), consumer4);
+
+ Map.Entry<Consumer, PositionImpl> entry6 = itr.next();
+ assertEquals(entry6.getValue(), PositionImpl.get(6, 1));
+ assertEquals(entry6.getKey(), consumer6);
+
+ // cleanup.
+ persistentDispatcher.close();
+ }
+
@Test
public void testSendMarkerMessage() {
try {