This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new c7ac08b [pulsar-broker] Stop to dispatch when skip message temporally
since Key_Shared consumer stuck on delivery (#7553)
c7ac08b is described below
commit c7ac08b13e95cd7bbbcf551d2112e9131153bd83
Author: Yuri Mizushima <yumiz...@yahoo-corp.jp>
AuthorDate: Wed Sep 2 23:36:29 2020 +0900
[pulsar-broker] Stop to dispatch when skip message temporally since
Key_Shared consumer stuck on delivery (#7553)
### Motivation
In some case of Key_Shared consumer, messages ordering was broken.
Here is how to reproduce(I think it is one of case to reproduce this issue).
1. Connect Consumer1 to Key_Shared subscription `sub` and stop to receive
- receiverQueueSize: 500
2. Connect Producer and publish 500 messages with key `(i % 10)`
3. Connect Consumer2 to same subscription and start to receive
- receiverQueueSize: 1
- since https://github.com/apache/pulsar/pull/7106 , Consumer2 can't
receive (expected)
4. Producer publish more 500 messages with same key generation algorithm
5. After that, Consumer1 start to receive
6. Check Consumer2 message ordering
- sometimes message ordering was broken in same key
Consumer1:
```
Connected: Tue Jul 14 09:36:39 JST 2020
[pulsar-client-io-1-1] WARN com.scurrilous.circe.checksum.Crc32cIntChecksum
- Failed to load Circe JNI library. Falling back to Java based CRC32c provider
[pulsar-timer-4-1] INFO
org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl -
[persistent://public/default/key-shared-test] [sub0] [820f0] Prefetched
messages: 499 --- Consume throughput received: 0.02 msgs/s --- 0.00 Mbit/s ---
Ack sent rate: 0.00 ack/s --- Failed messages: 0 --- batch messages: 0
---Failed acks: 0
Received: my-message-0 PublishTime: 1594687006203 Date: Tue Jul 14 09:37:46
JST 2020
Received: my-message-1 PublishTime: 1594687006243 Date: Tue Jul 14 09:37:46
JST 2020
Received: my-message-2 PublishTime: 1594687006247 Date: Tue Jul 14 09:37:46
JST 2020
...
Received: my-message-498 PublishTime: 1594687008727 Date: Tue Jul 14
09:37:46 JST 2020
Received: my-message-499 PublishTime: 1594687008731 Date: Tue Jul 14
09:37:46 JST 2020
Received: my-message-500 PublishTime: 1594687038742 Date: Tue Jul 14
09:37:46 JST 2020
...
Received: my-message-990 PublishTime: 1594687040094 Date: Tue Jul 14
09:37:46 JST 2020
Received: my-message-994 PublishTime: 1594687040103 Date: Tue Jul 14
09:37:46 JST 2020
Received: my-message-995 PublishTime: 1594687040105 Date: Tue Jul 14
09:37:46 JST 2020
Received: my-message-997 PublishTime: 1594687040113 Date: Tue Jul 14
09:37:46 JST 2020
```
Consumer2:
```
Connected: Tue Jul 14 09:37:03 JST 2020
[pulsar-client-io-1-1] WARN com.scurrilous.circe.checksum.Crc32cIntChecksum
- Failed to load Circe JNI library. Falling back to Java based CRC32c provider
Received: my-message-501 MessageId: 4:1501:-1 PublishTime: 1594687038753
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-502 MessageId: 4:1502:-1 PublishTime: 1594687038755
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-503 MessageId: 4:1503:-1 PublishTime: 1594687038759
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-506 MessageId: 4:1506:-1 PublishTime: 1594687038785
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-508 MessageId: 4:1508:-1 PublishTime: 1594687038812
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-901 MessageId: 4:1901:-1 PublishTime: 1594687039871
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-509 MessageId: 4:1509:-1 PublishTime: 1594687038815
Date: Tue Jul 14 09:37:46 JST 2020
ordering was broken, key: 1 oldNum: 901 newNum: 511
Received: my-message-511 MessageId: 4:1511:-1 PublishTime: 1594687038826
Date: Tue Jul 14 09:37:46 JST 2020
Received: my-message-512 MessageId: 4:1512:-1 PublishTime: 1594687038830
Date: Tue Jul 14 09:37:46 JST 2020
...
```
I think this issue is caused by https://github.com/apache/pulsar/pull/7105.
Here is an example.
1. dispatch messages
2. Consumer2 was stuck and `totalMessagesSent=0`
- Consumer2 availablePermits was 0
3. skip redeliver messages temporally
- Consumer2 availablePermits was back to 1
4. dispatch new messages
- new message was dispatched to Consumer2
5. back to redeliver messages
4. dispatch messages
- ordering was broken
### Modifications
Stop to dispatch when skip message temporally since Key_Shared consumer
stuck on delivery.
---
...istentStickyKeyDispatcherMultipleConsumers.java | 26 ++++-
...ntStickyKeyDispatcherMultipleConsumersTest.java | 105 ++++++++++++++++++++-
2 files changed, 127 insertions(+), 4 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 a56f566..095f9ba 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
@@ -25,6 +25,7 @@ import io.netty.util.concurrent.FastThreadLocal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -64,12 +65,17 @@ public class PersistentStickyKeyDispatcherMultipleConsumers
extends PersistentDi
*/
private final Map<Consumer, PositionImpl> recentlyJoinedConsumers;
+ private final Set<Consumer> stuckConsumers;
+ private final Set<Consumer> nextStuckConsumers;
+
PersistentStickyKeyDispatcherMultipleConsumers(PersistentTopic topic,
ManagedCursor cursor,
Subscription subscription, ServiceConfiguration conf,
KeySharedMeta ksm) {
super(topic, cursor, subscription);
this.allowOutOfOrderDelivery = ksm.getAllowOutOfOrderDelivery();
this.recentlyJoinedConsumers = allowOutOfOrderDelivery ?
Collections.emptyMap() : new HashMap<>();
+ this.stuckConsumers = new HashSet<>();
+ this.nextStuckConsumers = new HashSet<>();
switch (ksm.getKeySharedMode()) {
case AUTO_SPLIT:
@@ -143,6 +149,8 @@ public class PersistentStickyKeyDispatcherMultipleConsumers
extends PersistentDi
return;
}
+ nextStuckConsumers.clear();
+
final Map<Consumer, List<Entry>> groupedEntries =
localGroupedEntries.get();
groupedEntries.clear();
@@ -217,11 +225,14 @@ public class
PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
}
}
+ stuckConsumers.clear();
+
if (totalMessagesSent == 0 && recentlyJoinedConsumers.isEmpty()) {
// This means, that all the messages we've just read cannot be
dispatched right now.
// This condition can only happen when:
// 1. We have consumers ready to accept messages (otherwise the
would not haven been triggered)
// 2. All keys in the current set of messages are routing to
consumers that are currently busy
+ // and stuck is not caused by stuckConsumers
//
// The solution here is to move on and read next batch of messages
which might hopefully contain
// also keys meant for other consumers.
@@ -230,18 +241,31 @@ public class
PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
// ahead in the stream while the new consumers are not ready to
accept the new messages,
// therefore would be most likely only increase the distance
between read-position and mark-delete
// position.
- isDispatcherStuckOnReplays = true;
+ if (!nextStuckConsumers.isEmpty()) {
+ isDispatcherStuckOnReplays = true;
+ stuckConsumers.addAll(nextStuckConsumers);
+ }
+ // readMoreEntries should run regardless whether or not stuck is
caused by stuckConsumers for avoid stopping dispatch.
readMoreEntries();
}
}
private int getRestrictedMaxEntriesForConsumer(Consumer consumer,
List<Entry> entries, int maxMessages) {
if (maxMessages == 0) {
+ // the consumer was stuck
+ nextStuckConsumers.add(consumer);
return 0;
}
PositionImpl maxReadPosition = recentlyJoinedConsumers.get(consumer);
if (maxReadPosition == null) {
+ // stop to dispatch by stuckConsumers
+ if (stuckConsumers.contains(consumer)) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] stop to dispatch by stuckConsumers,
consumer: {}", name, consumer);
+ }
+ return 0;
+ }
// The consumer has not recently joined, so we can send all
messages
return maxMessages;
}
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 6325f10..c281400 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
@@ -41,6 +41,7 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.ObjectFactory;
import org.testng.annotations.Test;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@@ -49,6 +50,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import static
org.apache.pulsar.common.protocol.Commands.serializeMetadataAndPayload;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.fail;
@PrepareForTest({ DispatchRateLimiter.class })
@@ -79,6 +81,8 @@ public class
PersistentStickyKeyDispatcherMultipleConsumersTest {
configMock = mock(ServiceConfiguration.class);
doReturn(true).when(configMock).isSubscriptionRedeliveryTrackerEnabled();
doReturn(100).when(configMock).getDispatcherMaxReadBatchSize();
+
doReturn(true).when(configMock).isSubscriptionKeySharedUseConsistentHashing();
+
doReturn(1).when(configMock).getSubscriptionKeySharedConsistentHashingReplicaPoints();
pulsarMock = mock(PulsarService.class);
doReturn(configMock).when(pulsarMock).getConfiguration();
@@ -96,6 +100,7 @@ public class
PersistentStickyKeyDispatcherMultipleConsumersTest {
consumerMock = mock(Consumer.class);
channelMock = mock(ChannelPromise.class);
+ doReturn("consumer1").when(consumerMock).consumerName();
doReturn(1000).when(consumerMock).getAvailablePermits();
doReturn(true).when(consumerMock).isWritable();
doReturn(channelMock).when(consumerMock).sendMessages(
@@ -120,12 +125,17 @@ public class
PersistentStickyKeyDispatcherMultipleConsumersTest {
persistentDispatcher = new
PersistentStickyKeyDispatcherMultipleConsumers(
topicMock, cursorMock, subscriptionMock, configMock,
KeySharedMeta.getDefaultInstance());
- persistentDispatcher.addConsumer(consumerMock);
- persistentDispatcher.consumerFlow(consumerMock, 1000);
}
@Test
public void testSendMarkerMessage() {
+ try {
+ persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.consumerFlow(consumerMock, 1000);
+ } catch (Exception e) {
+ fail("Failed to add mock consumer", e);
+ }
+
List<Entry> entries = new ArrayList<>();
ByteBuf markerMessage =
Markers.newReplicatedSubscriptionsSnapshotRequest("testSnapshotId",
"testSourceCluster");
entries.add(EntryImpl.create(1, 1, markerMessage));
@@ -156,11 +166,100 @@ public class
PersistentStickyKeyDispatcherMultipleConsumersTest {
Assert.assertEquals(allTotalMessagesCaptor.get(0).intValue(), 5);
}
+ @Test
+ public void testSkipRedeliverTemporally() {
+ final Consumer slowConsumerMock = mock(Consumer.class);
+ final ChannelPromise slowChannelMock = mock(ChannelPromise.class);
+ // add entries to redeliver and read target
+ final List<Entry> redeliverEntries = new ArrayList<>();
+ redeliverEntries.add(EntryImpl.create(1, 1, createMessage("message1",
1, "key1")));
+ final List<Entry> readEntries = new ArrayList<>();
+ readEntries.add(EntryImpl.create(1, 2, createMessage("message2", 2,
"key1")));
+ readEntries.add(EntryImpl.create(1, 3, createMessage("message3", 3,
"key2")));
+
+ try {
+ Field totalAvailablePermitsField =
PersistentDispatcherMultipleConsumers.class.getDeclaredField("totalAvailablePermits");
+ totalAvailablePermitsField.setAccessible(true);
+ totalAvailablePermitsField.set(persistentDispatcher, 1000);
+
+ doAnswer(invocationOnMock -> {
+ ((PersistentStickyKeyDispatcherMultipleConsumers)
invocationOnMock.getArgument(2))
+ .readEntriesComplete(readEntries,
PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal);
+ return null;
+ }).when(cursorMock).asyncReadEntriesOrWait(
+ anyInt(), anyLong(),
any(PersistentStickyKeyDispatcherMultipleConsumers.class),
+
eq(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal));
+ } catch (Exception e) {
+ fail("Failed to set to field", e);
+ }
+
+ // Create 2Consumers
+ try {
+ doReturn("consumer2").when(slowConsumerMock).consumerName();
+ // Change slowConsumer availablePermits to 0 and back to normal
+ when(slowConsumerMock.getAvailablePermits())
+ .thenReturn(0)
+ .thenReturn(1);
+ doReturn(true).when(slowConsumerMock).isWritable();
+ doReturn(slowChannelMock).when(slowConsumerMock).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+
+ persistentDispatcher.addConsumer(consumerMock);
+ persistentDispatcher.addConsumer(slowConsumerMock);
+ } catch (Exception e) {
+ fail("Failed to add mock consumer", e);
+ }
+
+ // run
PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers
+ // run readMoreEntries internally (and skip internally)
+ // Change slowConsumer availablePermits to 1
+ // run
PersistentStickyKeyDispatcherMultipleConsumers#sendMessagesToConsumers
internally
+ // and then stop to dispatch to slowConsumer
+
persistentDispatcher.sendMessagesToConsumers(PersistentStickyKeyDispatcherMultipleConsumers.ReadType.Normal,
redeliverEntries);
+
+ verify(consumerMock, times(1)).sendMessages(
+ argThat(arg -> {
+ assertEquals(arg.size(), 1);
+ Entry entry = arg.get(0);
+ assertEquals(entry.getLedgerId(), 1);
+ assertEquals(entry.getEntryId(), 3);
+ return true;
+ }),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+ verify(slowConsumerMock, times(0)).sendMessages(
+ anyList(),
+ any(EntryBatchSizes.class),
+ any(EntryBatchIndexesAcks.class),
+ anyInt(),
+ anyLong(),
+ anyLong(),
+ any(RedeliveryTracker.class)
+ );
+ }
+
private ByteBuf createMessage(String message, int sequenceId) {
+ return createMessage(message, sequenceId, "testKey");
+ }
+
+ private ByteBuf createMessage(String message, int sequenceId, String key) {
PulsarApi.MessageMetadata.Builder messageMetadata =
PulsarApi.MessageMetadata.newBuilder();
messageMetadata.setSequenceId(sequenceId);
messageMetadata.setProducerName("testProducer");
- messageMetadata.setPartitionKey("testKey");
+ messageMetadata.setPartitionKey(key);
+ messageMetadata.setPartitionKeyB64Encoded(false);
messageMetadata.setPublishTime(System.currentTimeMillis());
return serializeMetadataAndPayload(Commands.ChecksumType.Crc32c,
messageMetadata.build(), Unpooled.copiedBuffer(message.getBytes(UTF_8)));
}
