This is an automated email from the ASF dual-hosted git repository. technoboy pushed a commit to branch revert-21953-pip_282_to_master in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2ca3d65db30e8ea7f5b45d845c17d6fba2184de4 Author: Jiwei Guo <[email protected]> AuthorDate: Sat Jul 20 10:07:25 2024 +0800 Revert "[fix][broker] Introduce the last sent position to fix message orderin…" This reverts commit d7e8ea16e6682df9a9354cda25cf4f1f9cb54429. --- .../bookkeeper/mledger/impl/ManagedCursorImpl.java | 13 - .../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +- .../org/apache/pulsar/broker/service/Consumer.java | 10 +- ...istentStickyKeyDispatcherMultipleConsumers.java | 195 ++-------- .../service/persistent/PersistentSubscription.java | 19 +- .../apache/pulsar/broker/admin/AdminApiTest.java | 196 +--------- ...ntStickyKeyDispatcherMultipleConsumersTest.java | 357 ------------------ .../pulsar/broker/stats/ConsumerStatsTest.java | 2 +- .../client/api/KeySharedSubscriptionTest.java | 399 +-------------------- .../pulsar/common/policies/data/ConsumerStats.java | 4 +- .../common/policies/data/SubscriptionStats.java | 6 - .../policies/data/stats/ConsumerStatsImpl.java | 10 +- .../policies/data/stats/SubscriptionStatsImpl.java | 6 - 13 files changed, 61 insertions(+), 1158 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index f99ee957e02..4ef9678f3e1 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java @@ -3472,19 +3472,6 @@ public class ManagedCursorImpl implements ManagedCursor { return individualDeletedMessages; } - public Position processIndividuallyDeletedMessagesAndGetMarkDeletedPosition( - LongPairRangeSet.RangeProcessor<Position> processor) { - final Position mdp; - lock.readLock().lock(); - try { - mdp = markDeletePosition; - individualDeletedMessages.forEach(processor); - } finally { - lock.readLock().unlock(); - } - return mdp; - } - public boolean isMessageDeleted(Position position) { lock.readLock().lock(); try { diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 209bf57b24f..b7734906f75 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -3497,7 +3497,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback { * the position range * @return the count of entries */ - public long getNumberOfEntries(Range<Position> range) { + long getNumberOfEntries(Range<Position> range) { Position fromPosition = range.lowerEndpoint(); boolean fromIncluded = range.lowerBoundType() == BoundType.CLOSED; Position toPosition = range.upperEndpoint(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index dca64395d86..02e21c44c91 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -145,7 +145,7 @@ public class Consumer { private static final double avgPercent = 0.9; private boolean preciseDispatcherFlowControl; - private Position lastSentPositionWhenJoining; + private Position readPositionWhenJoining; private final String clientAddress; // IP address only, no port number included private final MessageId startMessageId; private final boolean isAcknowledgmentAtBatchIndexLevelEnabled; @@ -931,8 +931,8 @@ public class Consumer { stats.unackedMessages = unackedMessages; stats.blockedConsumerOnUnackedMsgs = blockedConsumerOnUnackedMsgs; stats.avgMessagesPerEntry = getAvgMessagesPerEntry(); - if (lastSentPositionWhenJoining != null) { - stats.lastSentPositionWhenJoining = lastSentPositionWhenJoining.toString(); + if (readPositionWhenJoining != null) { + stats.readPositionWhenJoining = readPositionWhenJoining.toString(); } return stats; } @@ -1166,8 +1166,8 @@ public class Consumer { return preciseDispatcherFlowControl; } - public void setLastSentPositionWhenJoining(Position lastSentPositionWhenJoining) { - this.lastSentPositionWhenJoining = lastSentPositionWhenJoining; + public void setReadPositionWhenJoining(Position readPositionWhenJoining) { + this.readPositionWhenJoining = readPositionWhenJoining; } public int getMaxUnackedMessages() { 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 91cec1f8e90..766f45ad990 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 @@ -34,12 +34,9 @@ import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; -import javax.annotation.Nullable; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.Position; -import org.apache.bookkeeper.mledger.PositionFactory; -import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.commons.collections4.MapUtils; import org.apache.pulsar.broker.ServiceConfiguration; @@ -58,8 +55,6 @@ import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.KeySharedMeta; import org.apache.pulsar.common.api.proto.KeySharedMode; import org.apache.pulsar.common.util.FutureUtil; -import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet; -import org.apache.pulsar.common.util.collections.LongPairRangeSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -78,22 +73,12 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi */ private final LinkedHashMap<Consumer, Position> recentlyJoinedConsumers; - /** - * The lastSentPosition and the individuallySentPositions are not thread safe. - */ - @Nullable - private Position lastSentPosition; - private final LongPairRangeSet<Position> individuallySentPositions; - private static final LongPairRangeSet.LongPairConsumer<Position> positionRangeConverter = PositionFactory::create; - PersistentStickyKeyDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor, Subscription subscription, ServiceConfiguration conf, KeySharedMeta ksm) { super(topic, cursor, subscription, ksm.isAllowOutOfOrderDelivery()); this.allowOutOfOrderDelivery = ksm.isAllowOutOfOrderDelivery(); this.recentlyJoinedConsumers = allowOutOfOrderDelivery ? null : new LinkedHashMap<>(); - this.individuallySentPositions = - allowOutOfOrderDelivery ? null : new ConcurrentOpenLongPairRangeSet<>(4096, positionRangeConverter); this.keySharedMode = ksm.getKeySharedMode(); switch (this.keySharedMode) { case AUTO_SPLIT: @@ -139,18 +124,15 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi }) ).thenRun(() -> { synchronized (PersistentStickyKeyDispatcherMultipleConsumers.this) { - if (!allowOutOfOrderDelivery) { - final Position lastSentPositionWhenJoining = updateIfNeededAndGetLastSentPosition(); - if (lastSentPositionWhenJoining != null) { - consumer.setLastSentPositionWhenJoining(lastSentPositionWhenJoining); - // If this was the 1st consumer, or if all the messages are already acked, then we - // don't need to do anything special - if (recentlyJoinedConsumers != null - && consumerList.size() > 1 - && cursor.getNumberOfEntriesSinceFirstNotAckedMessage() > 1) { - recentlyJoinedConsumers.put(consumer, lastSentPositionWhenJoining); - } - } + Position readPositionWhenJoining = cursor.getReadPosition(); + consumer.setReadPositionWhenJoining(readPositionWhenJoining); + // If this was the 1st consumer, or if all the messages are already acked, then we + // don't need to do anything special + if (!allowOutOfOrderDelivery + && recentlyJoinedConsumers != null + && consumerList.size() > 1 + && cursor.getNumberOfEntriesSinceFirstNotAckedMessage() > 1) { + recentlyJoinedConsumers.put(consumer, readPositionWhenJoining); } } }); @@ -166,16 +148,10 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi // eventually causing all consumers to get stuck. selector.removeConsumer(consumer); super.removeConsumer(consumer); - if (!allowOutOfOrderDelivery && recentlyJoinedConsumers != null) { + if (recentlyJoinedConsumers != null) { recentlyJoinedConsumers.remove(consumer); if (consumerList.size() == 1) { recentlyJoinedConsumers.clear(); - } else if (consumerList.isEmpty()) { - // The subscription removes consumers if rewind or reset cursor operations are called. - // The dispatcher must clear lastSentPosition and individuallySentPositions because - // these operations trigger re-sending messages. - lastSentPosition = null; - individuallySentPositions.clear(); } if (removeConsumersFromRecentJoinedConsumers() || !redeliveryMessages.isEmpty()) { readMoreEntries(); @@ -217,9 +193,9 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi return false; } + // A corner case that we have to retry a readMoreEntries in order to preserver order delivery. + // This may happen when consumer closed. See issue #12885 for details. if (!allowOutOfOrderDelivery) { - // A corner case that we have to retry a readMoreEntries in order to preserver order delivery. - // This may happen when consumer closed. See issue #12885 for details. NavigableSet<Position> messagesToReplayNow = this.getMessagesToReplayNow(1); if (messagesToReplayNow != null && !messagesToReplayNow.isEmpty()) { Position replayPosition = messagesToReplayNow.first(); @@ -253,24 +229,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi } } } - - // Update if the markDeletePosition move forward - updateIfNeededAndGetLastSentPosition(); - - // Should not access to individualDeletedMessages from outside managed cursor - // because it doesn't guarantee thread safety. - if (lastSentPosition == null) { - if (cursor.getMarkDeletedPosition() != null) { - lastSentPosition = ((ManagedCursorImpl) cursor) - .processIndividuallyDeletedMessagesAndGetMarkDeletedPosition(range -> { - final Position lower = range.lowerEndpoint(); - final Position upper = range.upperEndpoint(); - individuallySentPositions.addOpenClosed(lower.getLedgerId(), lower.getEntryId(), - upper.getLedgerId(), upper.getEntryId()); - return true; - }); - } - } } final Map<Consumer, List<Entry>> groupedEntries = localGroupedEntries.get(); @@ -322,24 +280,12 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi } if (messagesForC > 0) { - final ManagedLedgerImpl managedLedger = ((ManagedLedgerImpl) cursor.getManagedLedger()); - for (int i = 0; i < messagesForC; i++) { - final Entry entry = entriesWithSameKey.get(i); - // remove positions first from replay list first : sendMessages recycles entries - if (readType == ReadType.Replay) { + // remove positions first from replay list first : sendMessages recycles entries + if (readType == ReadType.Replay) { + for (int i = 0; i < messagesForC; i++) { + Entry entry = entriesWithSameKey.get(i); redeliveryMessages.remove(entry.getLedgerId(), entry.getEntryId()); } - // Add positions to individuallySentPositions if necessary - if (!allowOutOfOrderDelivery) { - final Position position = entry.getPosition(); - // Store to individuallySentPositions even if lastSentPosition is null - if ((lastSentPosition == null || position.compareTo(lastSentPosition) > 0) - && !individuallySentPositions.contains(position.getLedgerId(), position.getEntryId())) { - final Position previousPosition = managedLedger.getPreviousPosition(position); - individuallySentPositions.addOpenClosed(previousPosition.getLedgerId(), - previousPosition.getEntryId(), position.getLedgerId(), position.getEntryId()); - } - } } SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal(); @@ -365,61 +311,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi } } - // Update the last sent position and remove ranges from individuallySentPositions if necessary - if (!allowOutOfOrderDelivery && lastSentPosition != null) { - final ManagedLedgerImpl managedLedger = ((ManagedLedgerImpl) cursor.getManagedLedger()); - com.google.common.collect.Range<Position> range = individuallySentPositions.firstRange(); - - // If the upper bound is before the last sent position, we need to move ahead as these - // individuallySentPositions are now irrelevant. - if (range != null && range.upperEndpoint().compareTo(lastSentPosition) <= 0) { - individuallySentPositions.removeAtMost(lastSentPosition.getLedgerId(), - lastSentPosition.getEntryId()); - range = individuallySentPositions.firstRange(); - } - - if (range != null) { - // If the lowerBound is ahead of the last sent position, - // verify if there are any entries in-between. - if (range.lowerEndpoint().compareTo(lastSentPosition) <= 0 || managedLedger - .getNumberOfEntries(com.google.common.collect.Range.openClosed(lastSentPosition, - range.lowerEndpoint())) <= 0) { - if (log.isDebugEnabled()) { - log.debug("[{}] Found a position range to last sent: {}", name, range); - } - Position newLastSentPosition = range.upperEndpoint(); - Position positionAfterNewLastSent = managedLedger - .getNextValidPosition(newLastSentPosition); - // sometime ranges are connected but belongs to different ledgers - // so, they are placed sequentially - // eg: (2:10..3:15] can be returned as (2:10..2:15],[3:0..3:15]. - // So, try to iterate over connected range and found the last non-connected range - // which gives new last sent position. - final Position lastConfirmedEntrySnapshot = managedLedger.getLastConfirmedEntry(); - if (lastConfirmedEntrySnapshot != null) { - while (positionAfterNewLastSent.compareTo(lastConfirmedEntrySnapshot) <= 0) { - if (individuallySentPositions.contains(positionAfterNewLastSent.getLedgerId(), - positionAfterNewLastSent.getEntryId())) { - range = individuallySentPositions.rangeContaining( - positionAfterNewLastSent.getLedgerId(), positionAfterNewLastSent.getEntryId()); - newLastSentPosition = range.upperEndpoint(); - positionAfterNewLastSent = managedLedger.getNextValidPosition(newLastSentPosition); - // check if next valid position is also deleted and part of the deleted-range - continue; - } - break; - } - } - - if (lastSentPosition.compareTo(newLastSentPosition) < 0) { - lastSentPosition = newLastSentPosition; - } - individuallySentPositions.removeAtMost(lastSentPosition.getLedgerId(), - lastSentPosition.getEntryId()); - } - } - } - // acquire message-dispatch permits for already delivered messages acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent); @@ -460,10 +351,10 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi return maxMessages; } removeConsumersFromRecentJoinedConsumers(); - Position maxLastSentPosition = recentlyJoinedConsumers.get(consumer); + Position maxReadPosition = recentlyJoinedConsumers.get(consumer); // At this point, all the old messages were already consumed and this consumer // is now ready to receive any message - if (maxLastSentPosition == null) { + if (maxReadPosition == null) { // The consumer has not recently joined, so we can send all messages return maxMessages; } @@ -482,16 +373,16 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi // But the message [2,3] should not dispatch to consumer2. if (readType == ReadType.Replay) { - Position minLastSentPositionForRecentJoinedConsumer = recentlyJoinedConsumers.values().iterator().next(); - if (minLastSentPositionForRecentJoinedConsumer != null - && minLastSentPositionForRecentJoinedConsumer.compareTo(maxLastSentPosition) < 0) { - maxLastSentPosition = minLastSentPositionForRecentJoinedConsumer; + Position minReadPositionForRecentJoinedConsumer = recentlyJoinedConsumers.values().iterator().next(); + if (minReadPositionForRecentJoinedConsumer != null + && minReadPositionForRecentJoinedConsumer.compareTo(maxReadPosition) < 0) { + maxReadPosition = minReadPositionForRecentJoinedConsumer; } } // Here, the consumer is one that has recently joined, so we can only send messages that were // published before it has joined. for (int i = 0; i < maxMessages; i++) { - if ((entries.get(i)).compareTo(maxLastSentPosition) > 0) { + if ((entries.get(i)).compareTo(maxReadPosition) >= 0) { // We have already crossed the divider line. All messages in the list are now // newer than what we can currently dispatch to this consumer return i; @@ -525,9 +416,11 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi boolean hasConsumerRemovedFromTheRecentJoinedConsumers = false; Position mdp = cursor.getMarkDeletedPosition(); if (mdp != null) { + Position nextPositionOfTheMarkDeletePosition = + ((ManagedLedgerImpl) cursor.getManagedLedger()).getNextValidPosition(mdp); while (itr.hasNext()) { Map.Entry<Consumer, Position> entry = itr.next(); - if (entry.getValue().compareTo(mdp) <= 0) { + if (entry.getValue().compareTo(nextPositionOfTheMarkDeletePosition) <= 0) { itr.remove(); hasConsumerRemovedFromTheRecentJoinedConsumers = true; } else { @@ -538,18 +431,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi return hasConsumerRemovedFromTheRecentJoinedConsumers; } - @Nullable - private synchronized Position updateIfNeededAndGetLastSentPosition() { - if (lastSentPosition == null) { - return null; - } - final Position mdp = cursor.getMarkDeletedPosition(); - if (mdp != null && mdp.compareTo(lastSentPosition) > 0) { - lastSentPosition = mdp; - } - return lastSentPosition; - } - @Override protected synchronized NavigableSet<Position> getMessagesToReplayNow(int maxMessagesToRead) { if (isDispatcherStuckOnReplays) { @@ -670,30 +551,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi return recentlyJoinedConsumers; } - public synchronized String getLastSentPosition() { - if (lastSentPosition == null) { - return null; - } - return lastSentPosition.toString(); - } - - @VisibleForTesting - public Position getLastSentPositionField() { - return lastSentPosition; - } - - public synchronized String getIndividuallySentPositions() { - if (individuallySentPositions == null) { - return null; - } - return individuallySentPositions.toString(); - } - - @VisibleForTesting - public LongPairRangeSet<Position> getIndividuallySentPositionsField() { - return individuallySentPositions; - } - public Map<Consumer, List<Range>> getConsumerKeyHashRanges() { return selector.getConsumerKeyHashRanges(); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 77aa5f82c39..a1d51668ca8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -1305,26 +1305,9 @@ public class PersistentSubscription extends AbstractSubscription { .getRecentlyJoinedConsumers(); if (recentlyJoinedConsumers != null && recentlyJoinedConsumers.size() > 0) { recentlyJoinedConsumers.forEach((k, v) -> { - // The dispatcher allows same name consumers - final StringBuilder stringBuilder = new StringBuilder(); - stringBuilder.append("consumerName=").append(k.consumerName()) - .append(", consumerId=").append(k.consumerId()); - if (k.cnx() != null) { - stringBuilder.append(", address=").append(k.cnx().clientAddress()); - } - subStats.consumersAfterMarkDeletePosition.put(stringBuilder.toString(), v.toString()); + subStats.consumersAfterMarkDeletePosition.put(k.consumerName(), v.toString()); }); } - final String lastSentPosition = ((PersistentStickyKeyDispatcherMultipleConsumers) dispatcher) - .getLastSentPosition(); - if (lastSentPosition != null) { - subStats.lastSentPosition = lastSentPosition; - } - final String individuallySentPositions = ((PersistentStickyKeyDispatcherMultipleConsumers) dispatcher) - .getIndividuallySentPositions(); - if (individuallySentPositions != null) { - subStats.individuallySentPositions = individuallySentPositions; - } } subStats.nonContiguousDeletedMessagesRanges = cursor.getTotalNonContiguousDeletedMessagesRange(); subStats.nonContiguousDeletedMessagesRangesSerializedSize = diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java index 5432b8a430d..1c83941d6e7 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java @@ -18,8 +18,6 @@ */ package org.apache.pulsar.broker.admin; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -58,7 +56,6 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Function; import javax.ws.rs.client.InvocationCallback; import javax.ws.rs.client.WebTarget; import javax.ws.rs.core.Response.Status; @@ -68,7 +65,6 @@ import lombok.SneakyThrows; import lombok.Value; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.ManagedLedgerInfo; -import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.PulsarServerException; @@ -79,8 +75,6 @@ import org.apache.pulsar.broker.loadbalance.LeaderBroker; import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; import org.apache.pulsar.broker.namespace.NamespaceEphemeralData; import org.apache.pulsar.broker.namespace.NamespaceService; -import org.apache.pulsar.broker.service.StickyKeyConsumerSelector; -import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers; import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.broker.testcontext.SpyConfig; import org.apache.pulsar.client.admin.GetStatsOptions; @@ -145,10 +139,7 @@ import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.policies.data.TopicHashPositions; import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.common.util.Codec; -import org.apache.pulsar.common.util.Murmur3_32Hash; import org.apache.pulsar.common.util.ObjectMapperFactory; -import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet; -import org.apache.pulsar.common.util.collections.LongPairRangeSet; import org.apache.pulsar.compaction.Compactor; import org.apache.pulsar.compaction.PulsarCompactionServiceFactory; import org.awaitility.Awaitility; @@ -3458,8 +3449,8 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { } @Test - public void testGetLastSentPositionWhenJoining() throws Exception { - final String topic = "persistent://prop-xyz/ns1/testGetLastSentPositionWhenJoining-" + UUID.randomUUID().toString(); + public void testGetReadPositionWhenJoining() throws Exception { + final String topic = "persistent://prop-xyz/ns1/testGetReadPositionWhenJoining-" + UUID.randomUUID().toString(); final String subName = "my-sub"; @Cleanup Producer<byte[]> producer = pulsarClient.newProducer() @@ -3467,189 +3458,34 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest { .enableBatching(false) .create(); - @Cleanup - final Consumer<byte[]> consumer1 = pulsarClient.newConsumer() - .topic(topic) - .subscriptionType(SubscriptionType.Key_Shared) - .subscriptionName(subName) - .subscribe(); - final int messages = 10; MessageIdImpl messageId = null; for (int i = 0; i < messages; i++) { messageId = (MessageIdImpl) producer.send(("Hello Pulsar - " + i).getBytes()); - consumer1.receive(); } - @Cleanup - final Consumer<byte[]> consumer2 = pulsarClient.newConsumer() - .topic(topic) - .subscriptionType(SubscriptionType.Key_Shared) - .subscriptionName(subName) - .subscribe(); + List<Consumer<byte[]>> consumers = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + Consumer<byte[]> consumer = pulsarClient.newConsumer() + .topic(topic) + .subscriptionType(SubscriptionType.Key_Shared) + .subscriptionName(subName) + .subscribe(); + consumers.add(consumer); + } TopicStats stats = admin.topics().getStats(topic); Assert.assertEquals(stats.getSubscriptions().size(), 1); SubscriptionStats subStats = stats.getSubscriptions().get(subName); Assert.assertNotNull(subStats); Assert.assertEquals(subStats.getConsumers().size(), 2); - ConsumerStats consumerStats = subStats.getConsumers().stream() - .filter(s -> s.getConsumerName().equals(consumer2.getConsumerName())).findFirst().get(); - Assert.assertEquals(consumerStats.getLastSentPositionWhenJoining(), - PositionFactory.create(messageId.getLedgerId(), messageId.getEntryId()).toString()); - } - - @Test - public void testGetLastSentPosition() throws Exception { - final String topic = "persistent://prop-xyz/ns1/testGetLastSentPosition-" + UUID.randomUUID().toString(); - final String subName = "my-sub"; - @Cleanup - final Producer<byte[]> producer = pulsarClient.newProducer() - .topic(topic) - .enableBatching(false) - .create(); - final AtomicInteger counter = new AtomicInteger(); - @Cleanup - final Consumer<byte[]> consumer = pulsarClient.newConsumer() - .topic(topic) - .subscriptionType(SubscriptionType.Key_Shared) - .subscriptionName(subName) - .messageListener((c, msg) -> { - try { - c.acknowledge(msg); - counter.getAndIncrement(); - } catch (Exception e) { - throw new RuntimeException(e); - } - }) - .subscribe(); - - TopicStats stats = admin.topics().getStats(topic); - Assert.assertEquals(stats.getSubscriptions().size(), 1); - SubscriptionStats subStats = stats.getSubscriptions().get(subName); - Assert.assertNotNull(subStats); - Assert.assertNull(subStats.getLastSentPosition()); + ConsumerStats consumerStats = subStats.getConsumers().get(0); + Assert.assertEquals(consumerStats.getReadPositionWhenJoining(), + PositionFactory.create(messageId.getLedgerId(), messageId.getEntryId() + 1).toString()); - final int messages = 10; - MessageIdImpl messageId = null; - for (int i = 0; i < messages; i++) { - messageId = (MessageIdImpl) producer.send(("Hello Pulsar - " + i).getBytes()); + for (Consumer<byte[]> consumer : consumers) { + consumer.close(); } - - Awaitility.await().untilAsserted(() -> assertEquals(counter.get(), messages)); - - stats = admin.topics().getStats(topic); - Assert.assertEquals(stats.getSubscriptions().size(), 1); - subStats = stats.getSubscriptions().get(subName); - Assert.assertNotNull(subStats); - Assert.assertEquals(subStats.getLastSentPosition(), PositionFactory.create(messageId.getLedgerId(), messageId.getEntryId()).toString()); - } - - @Test - public void testGetIndividuallySentPositions() throws Exception { - // The producer sends messages with two types of keys. - // The dispatcher sends keyA messages to consumer1. - // Consumer1 will not receive any messages. Its receiver queue size is 1. - // Consumer2 will receive and ack any messages immediately. - - final String topic = "persistent://prop-xyz/ns1/testGetIndividuallySentPositions-" + UUID.randomUUID().toString(); - final String subName = "my-sub"; - @Cleanup - final Producer<byte[]> producer = pulsarClient.newProducer() - .topic(topic) - .enableBatching(false) - .create(); - - final String consumer1Name = "c1"; - final String consumer2Name = "c2"; - - @Cleanup - final Consumer<byte[]> consumer1 = pulsarClient.newConsumer() - .topic(topic) - .consumerName(consumer1Name) - .receiverQueueSize(1) - .subscriptionType(SubscriptionType.Key_Shared) - .subscriptionName(subName) - .subscribe(); - - final PersistentStickyKeyDispatcherMultipleConsumers dispatcher = - (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topic).get().get().getSubscription(subName).getDispatcher(); - final String keyA = "key-a"; - final String keyB = "key-b"; - final int hashA = Murmur3_32Hash.getInstance().makeHash(keyA.getBytes()); - - final Field selectorField = PersistentStickyKeyDispatcherMultipleConsumers.class.getDeclaredField("selector"); - selectorField.setAccessible(true); - final StickyKeyConsumerSelector selector = spy((StickyKeyConsumerSelector) selectorField.get(dispatcher)); - selectorField.set(dispatcher, selector); - - // the selector returns consumer1 if keyA - doAnswer((invocationOnMock -> { - final int hash = invocationOnMock.getArgument(0); - - final String consumerName = hash == hashA ? consumer1Name : consumer2Name; - return dispatcher.getConsumers().stream().filter(consumer -> consumer.consumerName().equals(consumerName)).findFirst().get(); - })).when(selector).select(anyInt()); - - final AtomicInteger consumer2AckCounter = new AtomicInteger(); - @Cleanup - final Consumer<byte[]> consumer2 = pulsarClient.newConsumer() - .topic(topic) - .consumerName(consumer2Name) - .subscriptionType(SubscriptionType.Key_Shared) - .subscriptionName(subName) - .messageListener((c, msg) -> { - try { - c.acknowledge(msg); - consumer2AckCounter.getAndIncrement(); - } catch (Exception e) { - throw new RuntimeException(e); - } - }) - .subscribe(); - - final LongPairRangeSet.LongPairConsumer<Position> positionRangeConverter = PositionFactory::create; - final LongPairRangeSet<Position> expectedIndividuallySentPositions = new ConcurrentOpenLongPairRangeSet<>(4096, positionRangeConverter); - - TopicStats stats = admin.topics().getStats(topic); - Assert.assertEquals(stats.getSubscriptions().size(), 1); - SubscriptionStats subStats = stats.getSubscriptions().get(subName); - Assert.assertNotNull(subStats); - Assert.assertEquals(subStats.getIndividuallySentPositions(), expectedIndividuallySentPositions.toString()); - - final Function<String, MessageIdImpl> sendFn = (key) -> { - try { - return (MessageIdImpl) producer.newMessage().key(key).value(("msg").getBytes()).send(); - } catch (PulsarClientException e) { - throw new RuntimeException(e); - } - }; - final List<MessageIdImpl> messageIdList = new ArrayList<>(); - - // the dispatcher can send keyA message, but then consumer1's receiver queue will be full - messageIdList.add(sendFn.apply(keyA)); - - // the dispatcher can send messages other than keyA - messageIdList.add(sendFn.apply(keyA)); - messageIdList.add(sendFn.apply(keyB)); - messageIdList.add(sendFn.apply(keyA)); - messageIdList.add(sendFn.apply(keyB)); - messageIdList.add(sendFn.apply(keyB)); - - assertEquals(messageIdList.size(), 6); - Awaitility.await().untilAsserted(() -> assertEquals(consumer2AckCounter.get(), 3)); - - // set expected value - expectedIndividuallySentPositions.addOpenClosed(messageIdList.get(1).getLedgerId(), messageIdList.get(1).getEntryId(), - messageIdList.get(2).getLedgerId(), messageIdList.get(2).getEntryId()); - expectedIndividuallySentPositions.addOpenClosed(messageIdList.get(3).getLedgerId(), messageIdList.get(3).getEntryId(), - messageIdList.get(5).getLedgerId(), messageIdList.get(5).getEntryId()); - - stats = admin.topics().getStats(topic); - Assert.assertEquals(stats.getSubscriptions().size(), 1); - subStats = stats.getSubscriptions().get(subName); - Assert.assertNotNull(subStats); - Assert.assertEquals(subStats.getIndividuallySentPositions(), expectedIndividuallySentPositions.toString()); } @Test 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 1a205d0f686..a70b3ce7a42 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,19 +35,14 @@ 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.assertNull; -import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; -import com.google.common.collect.BoundType; -import com.google.common.collect.Range; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelPromise; import io.netty.channel.EventLoopGroup; import java.lang.reflect.Field; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Queue; @@ -55,14 +50,12 @@ import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Supplier; import java.util.stream.Collectors; import org.apache.bookkeeper.common.util.OrderedExecutor; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.EntryImpl; import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; -import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; @@ -79,14 +72,11 @@ import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.policies.data.HierarchyTopicPolicies; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.Markers; -import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet; -import org.apache.pulsar.common.util.collections.LongPairRangeSet; import org.awaitility.Awaitility; import org.mockito.ArgumentCaptor; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; -import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @Test(groups = "broker") @@ -94,7 +84,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest { private PulsarService pulsarMock; private BrokerService brokerMock; - private ManagedLedgerImpl ledgerMock; private ManagedCursorImpl cursorMock; private Consumer consumerMock; private PersistentTopic topicMock; @@ -146,44 +135,9 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest { doReturn(topicName).when(topicMock).getName(); doReturn(topicPolicies).when(topicMock).getHierarchyTopicPolicies(); - ledgerMock = mock(ManagedLedgerImpl.class); - doAnswer((invocationOnMock -> { - final Position position = invocationOnMock.getArgument(0); - if (position.getEntryId() > 0) { - return PositionFactory.create(position.getLedgerId(), position.getEntryId() - 1); - } else { - fail("Undefined behavior on mock"); - return PositionFactory.EARLIEST; - } - })).when(ledgerMock).getPreviousPosition(any(Position.class)); - doAnswer((invocationOnMock -> { - final Position position = invocationOnMock.getArgument(0); - return PositionFactory.create(position.getLedgerId(), position.getEntryId() < 0 ? 0 : position.getEntryId() + 1); - })).when(ledgerMock).getNextValidPosition(any(Position.class)); - doAnswer((invocationOnMock -> { - final Range<Position> range = invocationOnMock.getArgument(0); - Position fromPosition = range.lowerEndpoint(); - boolean fromIncluded = range.lowerBoundType() == BoundType.CLOSED; - Position toPosition = range.upperEndpoint(); - boolean toIncluded = range.upperBoundType() == BoundType.CLOSED; - - long count = 0; - - if (fromPosition.getLedgerId() == toPosition.getLedgerId()) { - // If the 2 positions are in the same ledger - count = toPosition.getEntryId() - fromPosition.getEntryId() - 1; - count += fromIncluded ? 1 : 0; - count += toIncluded ? 1 : 0; - } else { - fail("Undefined behavior on mock"); - } - return count; - })).when(ledgerMock).getNumberOfEntries(any()); - cursorMock = mock(ManagedCursorImpl.class); doReturn(null).when(cursorMock).getLastIndividualDeletedRange(); doReturn(subscriptionName).when(cursorMock).getName(); - doReturn(ledgerMock).when(cursorMock).getManagedLedger(); consumerMock = mock(Consumer.class); channelMock = mock(ChannelPromise.class); @@ -511,317 +465,6 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest { allEntries.forEach(entry -> entry.release()); } - - - @DataProvider(name = "initializeLastSentPosition") - private Object[][] initialLastSentPositionProvider() { - return new Object[][] { { false }, { true } }; - } - - @Test(dataProvider = "initializeLastSentPosition") - public void testLastSentPositionAndIndividuallySentPositions(final boolean initializeLastSentPosition) throws Exception { - final Position initialLastSentPosition = PositionFactory.create(1, 10); - final LongPairRangeSet<Position> expectedIndividuallySentPositions - = new ConcurrentOpenLongPairRangeSet<>(4096, PositionFactory::create); - - final Field lastSentPositionField = PersistentStickyKeyDispatcherMultipleConsumers.class - .getDeclaredField("lastSentPosition"); - lastSentPositionField.setAccessible(true); - final LongPairRangeSet<Position> individuallySentPositions = persistentDispatcher.getIndividuallySentPositionsField(); - final Supplier<Throwable> clearPosition = () -> { - try { - lastSentPositionField.set(persistentDispatcher, initializeLastSentPosition ? initialLastSentPosition : null); - individuallySentPositions.clear(); - expectedIndividuallySentPositions.clear(); - } catch (Throwable e) { - return e; - } - return null; - }; - if (!initializeLastSentPosition) { - doReturn(initialLastSentPosition).when(cursorMock).getMarkDeletedPosition(); - doAnswer(invocationOnMock -> { - // skip copy operation - return initialLastSentPosition; - }).when(cursorMock).processIndividuallyDeletedMessagesAndGetMarkDeletedPosition(any()); - } - - // Assume the range sequence is [1:0, 1:19], [2:0, 2:19], ..., [10:0, 10:19] - doAnswer((invocationOnMock -> { - final Position position = invocationOnMock.getArgument(0); - if (position.getEntryId() > 0) { - return PositionFactory.create(position.getLedgerId(), position.getEntryId() - 1); - } else if (position.getLedgerId() > 0) { - return PositionFactory.create(position.getLedgerId() - 1, 19); - } else { - throw new NullPointerException(); - } - })).when(ledgerMock).getPreviousPosition(any(Position.class)); - doAnswer((invocationOnMock -> { - final Position position = invocationOnMock.getArgument(0); - if (position.getEntryId() < 19) { - return PositionFactory.create(position.getLedgerId(), position.getEntryId() + 1); - } else { - return PositionFactory.create(position.getLedgerId() + 1, 0); - } - })).when(ledgerMock).getNextValidPosition(any(Position.class)); - doReturn(PositionFactory.create(10, 19)).when(ledgerMock).getLastConfirmedEntry(); - doAnswer((invocationOnMock -> { - final Range<Position> range = invocationOnMock.getArgument(0); - Position fromPosition = range.lowerEndpoint(); - boolean fromIncluded = range.lowerBoundType() == BoundType.CLOSED; - Position toPosition = range.upperEndpoint(); - boolean toIncluded = range.upperBoundType() == BoundType.CLOSED; - - if (fromPosition.getLedgerId() == toPosition.getLedgerId()) { - // If the 2 positions are in the same ledger - long count = toPosition.getEntryId() - fromPosition.getEntryId() - 1; - count += fromIncluded ? 1 : 0; - count += toIncluded ? 1 : 0; - return count; - } else { - long count = 0; - // If the from & to are pointing to different ledgers, then we need to : - // 1. Add the entries in the ledger pointed by toPosition - count += toPosition.getEntryId(); - count += toIncluded ? 1 : 0; - - // 2. Add the entries in the ledger pointed by fromPosition - count += 20 - (fromPosition.getEntryId() + 1); - count += fromIncluded ? 1 : 0; - - // 3. Add the whole ledgers entries in between - for (long i = fromPosition.getLedgerId() + 1; i < toPosition.getLedgerId(); i++) { - count += 20; - } - - return count; - } - })).when(ledgerMock).getNumberOfEntries(any()); - assertEquals(ledgerMock.getNextValidPosition(PositionFactory.create(1, 0)), PositionFactory.create(1, 1)); - assertEquals(ledgerMock.getNextValidPosition(PositionFactory.create(1, 19)), PositionFactory.create(2, 0)); - assertEquals(ledgerMock.getPreviousPosition(PositionFactory.create(2, 0)), PositionFactory.create(1, 19)); - assertThrows(NullPointerException.class, () -> ledgerMock.getPreviousPosition(PositionFactory.create(0, 0))); - assertEquals(ledgerMock.getNumberOfEntries(Range.openClosed( - PositionFactory.create(1, 0), PositionFactory.create(1, 0))), 0); - assertEquals(ledgerMock.getNumberOfEntries(Range.openClosed( - PositionFactory.create(1, -1), PositionFactory.create(1, 9))), 10); - assertEquals(ledgerMock.getNumberOfEntries(Range.openClosed( - PositionFactory.create(1, 19), PositionFactory.create(2, -1))), 0); - assertEquals(ledgerMock.getNumberOfEntries(Range.openClosed( - PositionFactory.create(1, 19), PositionFactory.create(2, 9))), 10); - assertEquals(ledgerMock.getNumberOfEntries(Range.openClosed( - PositionFactory.create(1, -1), PositionFactory.create(3, 19))), 60); - - // Add a consumer - final Consumer consumer1 = mock(Consumer.class); - doReturn("consumer1").when(consumer1).consumerName(); - when(consumer1.getAvailablePermits()).thenReturn(1000); - doReturn(true).when(consumer1).isWritable(); - doReturn(channelMock).when(consumer1).sendMessages(anyList(), any(EntryBatchSizes.class), - any(EntryBatchIndexesAcks.class), anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class)); - persistentDispatcher.addConsumer(consumer1); - - /* - On single ledger - */ - - // Expected individuallySentPositions (isp): [(1:-1, 1:8]] (init) -> [(1:-1, 1:9]] (update) -> [] (remove) - // Expected lastSentPosition (lsp): 1:10 (init) -> 1:10 (remove) - // upper bound and the new entry are less than initial last sent position - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, -1, 1, 8); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 9, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - // isp: [(1:-1, 1:9]] -> [(1:-1, 1:10]] -> [] - // lsp: 1:10 -> 1:10 - // upper bound is less than initial last sent position - // upper bound and the new entry are less than or equal to initial last sent position - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, -1, 1, 9); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 10, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - // isp: [(1:-1, 1:2], (1:3, 1:4], (1:5, 1:6]] -> [(1:-1, 1:2], (1:3, 1:4], (1:5, 1:6], (1:9, 1:10]] -> [] - // lsp: 1:10 -> 1:10 - // upper bound and the new entry are less than or equal to initial last sent position - // individually sent positions has multiple ranges - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, -1, 1, 2); - individuallySentPositions.addOpenClosed(1, 3, 1, 4); - individuallySentPositions.addOpenClosed(1, 5, 1, 6); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 10, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - // isp: [(1:-1, 1:10]] -> [(1:-1, 1:11]] -> [] - // lsp: 1:10 -> 1:11 - // upper bound is less than or equal to initial last sent position - // the new entry is next position of initial last sent position - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, -1, 1, 10); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 11, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), PositionFactory.create(1, 11).toString()); - - // isp: [(1:-1, 1:9]] -> [(1:-1, 1:9], (1:10, 1:11]] -> [] - // lsp: 1:10 -> 1:11 - // upper bound is less than initial last sent position - // the new entry is next position of initial last sent position - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, -1, 1, 9); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 11, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), PositionFactory.create(1, 11).toString()); - - // isp: [(1:11, 1:15]] -> [(1:10, 1:15]] -> [] - // lsp: 1:10 -> 1:15 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry is next position of initial last sent position - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 15); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 11, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), PositionFactory.create(1, 15).toString()); - - // isp: [(1:11, 1:15]] -> [(1:10, 1:16]] -> [] - // lsp: 1:10 -> 1:16 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entries contain next position of initial last sent position - // first of the new entries is less than initial last sent position - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 15); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 9, createMessage("test", 1)), - EntryImpl.create(1, 11, createMessage("test", 2)), - EntryImpl.create(1, 16, createMessage("test", 3))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), PositionFactory.create(1, 16).toString()); - - // isp: [(1:11, 1:15]] -> [(1:11, 1:15]] -> [(1:11, 1:15]] - // lsp: 1:10 -> 1:10 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry isn't next position of initial last sent position - // the range contains the new entry - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 15); - expectedIndividuallySentPositions.addOpenClosed(1, 11, 1, 15); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 15, createMessage("test", 1))), true); - assertEquals(individuallySentPositions.toString(), expectedIndividuallySentPositions.toString()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - // isp: [(1:11, 1:15]] -> [(1:11, 1:16]] -> [(1:11, 1:16]] - // lsp: 1:10 -> 1:10 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry isn't next position of initial last sent position - // the range doesn't contain the new entry - // the new entry is next position of upper bound - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 15); - expectedIndividuallySentPositions.addOpenClosed(1, 11, 1, 16); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 16, createMessage("test", 1))), true); - assertEquals(individuallySentPositions.toString(), expectedIndividuallySentPositions.toString()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - // isp: [(1:11, 1:15]] -> [(1:11, 1:15], (1:16, 1:17]] -> [(1:11, 1:15], (1:16, 1:17]] - // lsp: 1:10 -> 1:10 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry isn't next position of initial last sent position - // the range doesn't contain the new entry - // the new entry isn't next position of upper bound - // the new entry is same ledger - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 15); - expectedIndividuallySentPositions.addOpenClosed(1, 11, 1, 15); - expectedIndividuallySentPositions.addOpenClosed(1, 16, 1, 17); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 17, createMessage("test", 1))), true); - assertEquals(individuallySentPositions.toString(), expectedIndividuallySentPositions.toString()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - /* - On multiple contiguous ledgers - */ - - // isp: [(1:11, 1:18]] -> [(1:11, 1:18], (2:-1, 2:0]] -> [(1:11, 1:18], (2:-1, 2:0]] - // lsp: 1:10 -> 1:10 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry isn't next position of initial last sent position - // the range doesn't contain the new entry - // the new entry isn't next position of upper bound - // the new entry isn't same ledger - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 18); - expectedIndividuallySentPositions.addOpenClosed(1, 11, 1, 18); - expectedIndividuallySentPositions.addOpenClosed(2, -1, 2, 0); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(2, 0, createMessage("test", 1))), true); - assertEquals(individuallySentPositions.toString(), expectedIndividuallySentPositions.toString()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - - // isp: [(1:11, 1:19], (2:-1, 2:0]] -> [(1:10, 1:19], (2:-1, 2:0]] -> [] - // lsp: 1:10 -> 2:0 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry is next position of initial last sent position - // the new entry isn't same ledger - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 19); - individuallySentPositions.addOpenClosed(2, -1, 2, 0); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 11, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), PositionFactory.create(2, 0).toString()); - - // isp: [(1:11, 1:19], (2:-1, 2:19], (3:-1, 3:0]] -> [(1:10, 1:19], (2:-1, 2:19], (3:-1, 3:0]] -> [] - // lsp: 1:10 -> 3:0 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry is next position of initial last sent position - // the new entry isn't same ledger - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 19); - individuallySentPositions.addOpenClosed(2, -1, 2, 19); - individuallySentPositions.addOpenClosed(3, -1, 3, 0); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(1, 11, createMessage("test", 1))), true); - assertTrue(individuallySentPositions.isEmpty()); - assertEquals(persistentDispatcher.getLastSentPosition(), PositionFactory.create(3, 0).toString()); - - // isp: [(1:11, 1:19], (2:-1, 2:0]] -> [(1:11, 1:19], (2:-1, 2:1]] -> [(1:11, 1:19], (2:-1, 2:1]] - // lsp: 1:10 -> 1:10 - // upper bound is greater than initial last sent position - // the range doesn't contain next position of initial last sent position - // the new entry isn't next position of initial last sent position - // the new entry isn't same ledger - assertNull(clearPosition.get()); - individuallySentPositions.addOpenClosed(1, 11, 1, 19); - individuallySentPositions.addOpenClosed(2, -1, 2, 0); - expectedIndividuallySentPositions.addOpenClosed(1, 11, 1, 19); - expectedIndividuallySentPositions.addOpenClosed(2, -1, 2, 1); - persistentDispatcher.sendMessagesToConsumers(PersistentDispatcherMultipleConsumers.ReadType.Normal, - Arrays.asList(EntryImpl.create(2, 1, createMessage("test", 1))), true); - assertEquals(individuallySentPositions.toString(), expectedIndividuallySentPositions.toString()); - assertEquals(persistentDispatcher.getLastSentPosition(), initialLastSentPosition.toString()); - } - private ByteBuf createMessage(String message, int sequenceId) { return createMessage(message, sequenceId, "testKey"); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java index 14403765105..5b2998216e8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java @@ -233,7 +233,7 @@ public class ConsumerStatsTest extends ProducerConsumerBase { "unackedMessages", "avgMessagesPerEntry", "blockedConsumerOnUnackedMsgs", - "lastSentPositionWhenJoining", + "readPositionWhenJoining", "lastAckedTime", "lastAckedTimestamp", "lastConsumedTime", diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java index e8fd5378316..92c51da64d3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java @@ -18,9 +18,6 @@ */ package org.apache.pulsar.client.api; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; @@ -29,7 +26,6 @@ import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; import com.google.common.collect.Lists; import java.io.IOException; -import java.lang.reflect.Field; import java.nio.file.Files; import java.nio.file.Paths; import java.util.ArrayList; @@ -37,7 +33,6 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -54,7 +49,6 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Supplier; import java.util.stream.Collectors; import lombok.Cleanup; import org.apache.bookkeeper.mledger.Position; @@ -62,24 +56,17 @@ import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.BrokerTestUtil; -import org.apache.pulsar.broker.service.StickyKeyConsumerSelector; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.nonpersistent.NonPersistentStickyKeyDispatcherMultipleConsumers; -import org.apache.pulsar.broker.service.persistent.MessageRedeliveryController; -import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.persistent.PersistentSubscription; import org.apache.pulsar.broker.service.persistent.PersistentTopic; import org.apache.pulsar.client.impl.ConsumerImpl; -import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.common.api.proto.KeySharedMode; import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.policies.data.RetentionPolicies; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.util.Murmur3_32Hash; -import org.apache.pulsar.common.util.collections.ConcurrentOpenLongPairRangeSet; -import org.apache.pulsar.common.util.collections.LongPairRangeSet; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.slf4j.Logger; @@ -1109,21 +1096,13 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase { final String topicName = "persistent://public/default/change-allow-ooo-delivery-" + UUID.randomUUID(); final String subName = "my-sub"; - final Consumer<byte[]> consumer1 = pulsarClient.newConsumer() + Consumer<byte[]> consumer = pulsarClient.newConsumer() .topic(topicName) .subscriptionName(subName) .subscriptionType(SubscriptionType.Key_Shared) .keySharedPolicy(KeySharedPolicy.autoSplitHashRange().setAllowOutOfOrderDelivery(true)) .subscribe(); - @Cleanup - final Producer<byte[]> producer = pulsarClient.newProducer() - .topic(topicName) - .enableBatching(false) - .create(); - producer.send("message".getBytes()); - Awaitility.await().untilAsserted(() -> assertNotNull(consumer1.receive(100, TimeUnit.MILLISECONDS))); - CompletableFuture<Optional<Topic>> future = pulsar.getBrokerService().getTopicIfExists(topicName); assertTrue(future.isDone()); assertTrue(future.get().isPresent()); @@ -1131,18 +1110,14 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase { PersistentStickyKeyDispatcherMultipleConsumers dispatcher = (PersistentStickyKeyDispatcherMultipleConsumers) topic.getSubscription(subName).getDispatcher(); assertTrue(dispatcher.isAllowOutOfOrderDelivery()); - assertNull(dispatcher.getLastSentPositionField()); - assertNull(dispatcher.getIndividuallySentPositionsField()); - consumer1.close(); + consumer.close(); - final Consumer<byte[]> consumer2 = pulsarClient.newConsumer() + consumer = pulsarClient.newConsumer() .topic(topicName) .subscriptionName(subName) .subscriptionType(SubscriptionType.Key_Shared) .keySharedPolicy(KeySharedPolicy.autoSplitHashRange().setAllowOutOfOrderDelivery(false)) .subscribe(); - producer.send("message".getBytes()); - Awaitility.await().untilAsserted(() -> assertNotNull(consumer2.receive(100, TimeUnit.MILLISECONDS))); future = pulsar.getBrokerService().getTopicIfExists(topicName); assertTrue(future.isDone()); @@ -1150,9 +1125,7 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase { topic = future.get().get(); dispatcher = (PersistentStickyKeyDispatcherMultipleConsumers) topic.getSubscription(subName).getDispatcher(); assertFalse(dispatcher.isAllowOutOfOrderDelivery()); - assertNotNull(dispatcher.getLastSentPositionField()); - assertNotNull(dispatcher.getIndividuallySentPositionsField()); - consumer2.close(); + consumer.close(); } @Test(timeOut = 30_000) @@ -1226,370 +1199,6 @@ public class KeySharedSubscriptionTest extends ProducerConsumerBase { l.await(); } - @DataProvider(name = "preSend") - private Object[][] preSendProvider() { - return new Object[][] { { false }, { true } }; - } - - @Test(timeOut = 30_000, dataProvider = "preSend") - public void testCheckBetweenSkippingAndRecentlyJoinedConsumers(boolean preSend) throws Exception { - conf.setSubscriptionKeySharedUseConsistentHashing(true); - - final String topicName = "persistent://public/default/recently-joined-consumers-" + UUID.randomUUID(); - final String subName = "my-sub"; - - @Cleanup - final Producer<String> p = pulsarClient.newProducer(Schema.STRING) - .topic(topicName) - .create(); - if (preSend) { - // verify that the test succeeds even if the topic has a message - p.send("msg"); - } - - final Supplier<ConsumerBuilder<String>> cb = () -> pulsarClient.newConsumer(Schema.STRING) - .topic(topicName) - .subscriptionInitialPosition(SubscriptionInitialPosition.Latest) - .subscriptionName(subName) - .subscriptionType(SubscriptionType.Key_Shared) - .keySharedPolicy(KeySharedPolicy.autoSplitHashRange() - .setAllowOutOfOrderDelivery(false)); - - // create 2 consumers - final String c1ConsumerName = "c1"; - @Cleanup - final Consumer<String> c1 = cb.get().consumerName(c1ConsumerName).receiverQueueSize(1).subscribe(); - @Cleanup - final Consumer<String> c2 = cb.get().consumerName("c2").receiverQueueSize(1000).subscribe(); - - final PersistentStickyKeyDispatcherMultipleConsumers dispatcher = - (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName).getDispatcher(); - final Field recentlyJoinedConsumersField = PersistentStickyKeyDispatcherMultipleConsumers.class.getDeclaredField("recentlyJoinedConsumers"); - recentlyJoinedConsumersField.setAccessible(true); - final LinkedHashMap<org.apache.pulsar.broker.service.Consumer, Position> recentlyJoinedConsumers = (LinkedHashMap<org.apache.pulsar.broker.service.Consumer, Position>) recentlyJoinedConsumersField.get(dispatcher); - final String keyA = "key-a"; - final int hashA = Murmur3_32Hash.getInstance().makeHash(keyA.getBytes()); - final Map<Integer, String> hashConsumerMap = new HashMap<>(); - hashConsumerMap.put(hashA, c1.getConsumerName()); - - // enforce the selector will return c1 if keyA - final Field selectorField = PersistentStickyKeyDispatcherMultipleConsumers.class.getDeclaredField("selector"); - selectorField.setAccessible(true); - final StickyKeyConsumerSelector selector = spy((StickyKeyConsumerSelector) selectorField.get(dispatcher)); - selectorField.set(dispatcher, selector); - doAnswer((invocationOnMock -> { - final int hash = invocationOnMock.getArgument(0); - final String consumerName = hashConsumerMap.getOrDefault(hash, c2.getConsumerName()); - return dispatcher.getConsumers().stream().filter(consumer -> consumer.consumerName().equals(consumerName)).findFirst().get(); - })).when(selector).select(anyInt()); - - // send and receive - Awaitility.await().untilAsserted(() -> assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getConsumers().stream().filter(c -> c.getConsumerName().equals(c1ConsumerName)).findFirst().get().getAvailablePermits(), 1)); - final MessageIdImpl msg0Id = (MessageIdImpl) p.newMessage().key(keyA).value("msg-0").send(); - Awaitility.await().untilAsserted(() -> assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getConsumers().stream().filter(c -> c.getConsumerName().equals(c1ConsumerName)).findFirst().get().getAvailablePermits(), 0)); - - final MessageIdImpl msg1Id = (MessageIdImpl) p.newMessage().key(keyA).value("msg-1").send(); - Awaitility.await().untilAsserted(() -> assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getMsgBacklog(), 2)); - - final Field redeliveryMessagesField = PersistentDispatcherMultipleConsumers.class - .getDeclaredField("redeliveryMessages"); - redeliveryMessagesField.setAccessible(true); - final MessageRedeliveryController redeliveryMessages = (MessageRedeliveryController) redeliveryMessagesField.get(dispatcher); - - final Set<Position> replayMsgSet = redeliveryMessages.getMessagesToReplayNow(3); - assertEquals(replayMsgSet.size(), 1); - final Position replayMsg = replayMsgSet.stream().findAny().get(); - assertEquals(replayMsg, PositionFactory.create(msg1Id.getLedgerId(), msg1Id.getEntryId())); - - // add c3 - final String c3ConsumerName = "c3"; - hashConsumerMap.put(hashA, c3ConsumerName); - @Cleanup - final Consumer<String> c3 = cb.get().consumerName(c3ConsumerName).subscribe(); - final List<Message<String>> c3Msgs = new ArrayList<>(); - final org.apache.pulsar.broker.service.Consumer c3Broker = dispatcher.getConsumers().stream().filter(consumer -> consumer.consumerName().equals(c3ConsumerName)).findFirst().get(); - assertEquals(recentlyJoinedConsumers.get(c3Broker), PositionFactory.create(msg0Id.getLedgerId(), msg0Id.getEntryId())); - - // None of messages are sent to c3. - Message<String> c3Msg = c3.receive(100, TimeUnit.MILLISECONDS); - assertNull(c3Msg); - - // Disconnect c1 - c1.close(); - - c3Msg = c3.receive(100, TimeUnit.MILLISECONDS); - assertNotNull(c3Msg); - c3Msgs.add(c3Msg); - // The mark delete position will move forward. Then remove c3 from recentlyJoinedConsumers. - c3.acknowledge(c3Msg); - Awaitility.await().untilAsserted(() -> assertNull(recentlyJoinedConsumers.get(c3Broker))); - c3Msg = c3.receive(100, TimeUnit.MILLISECONDS); - assertNotNull(c3Msg); - c3Msgs.add(c3Msg); - c3.acknowledge(c3Msg); - - // check ordering - assertTrue(c3Msgs.get(0).getMessageId().compareTo(c3Msgs.get(1).getMessageId()) < 0); - } - - @Test(timeOut = 30_000) - public void testLastSentPositionWhenRecreatingDispatcher() throws Exception { - // The lastSentPosition and individuallySentPositions should be initialized - // by the markDeletedPosition and individuallyDeletedMessages. - final String topicName = "persistent://public/default/rewind-" + UUID.randomUUID(); - final String subName = "my-sub"; - - final int numMessages = 9; - final List<String> keys = Arrays.asList("key-a", "key-b", "key-c"); - final AtomicInteger receiveCounter = new AtomicInteger(); - final AtomicInteger ackCounter = new AtomicInteger(); - - @Cleanup - final Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32) - .topic(topicName) - .enableBatching(false) - .create(); - - final Supplier<ConsumerBuilder<Integer>> cb = () -> pulsarClient.newConsumer(Schema.INT32) - .topic(topicName) - .subscriptionName(subName) - .subscriptionType(SubscriptionType.Key_Shared) - .keySharedPolicy(KeySharedPolicy.autoSplitHashRange() - .setAllowOutOfOrderDelivery(false)); - - @Cleanup - final Consumer<Integer> c1 = cb.get().messageListener((c, msg) -> { - if (keys.get(0).equals(msg.getKey())) { - try { - c.acknowledge(msg); - ackCounter.getAndIncrement(); - } catch (PulsarClientException e) { - fail(e.getMessage()); - } - } - receiveCounter.getAndIncrement(); - }).subscribe(); - - PersistentStickyKeyDispatcherMultipleConsumers dispatcher = - (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName).getDispatcher(); - LongPairRangeSet<Position> individuallySentPositionsField = dispatcher.getIndividuallySentPositionsField(); - final ManagedCursorImpl cursor = (ManagedCursorImpl) ((PersistentSubscription) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName)).getCursor(); - final ManagedLedgerImpl ledger = (ManagedLedgerImpl) cursor.getManagedLedger(); - - MessageIdImpl msgId = null; - for (int i = 0; i < numMessages; i++) { - msgId = (MessageIdImpl) producer.newMessage().key(keys.get(i % keys.size())).value(i).send(); - } - - // wait for consumption - Awaitility.await().untilAsserted(() -> assertEquals(receiveCounter.get(), numMessages)); - assertEquals(ackCounter.get(), numMessages / keys.size()); - assertEquals(dispatcher.getLastSentPositionField(), PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId())); - assertTrue(individuallySentPositionsField.isEmpty()); - receiveCounter.set(0); - ackCounter.set(0); - - // create expected values - final Position expectedLastSentPosition = ledger.getNextValidPosition(cursor.getMarkDeletedPosition()); - final ConcurrentOpenLongPairRangeSet<Position> - expectedIndividuallySentPositions = new ConcurrentOpenLongPairRangeSet<>(4096, PositionFactory::create); - cursor.getIndividuallyDeletedMessagesSet().forEach(range -> { - final Position lower = range.lowerEndpoint(); - final Position upper = range.upperEndpoint(); - expectedIndividuallySentPositions.addOpenClosed(lower.getLedgerId(), lower.getEntryId(), upper.getLedgerId(), upper.getEntryId()); - return true; - }); - - // modify subscription type to close current dispatcher - admin.topics().createSubscription(topicName, "sub-alt", MessageId.earliest); - c1.close(); - @Cleanup - final Consumer<Integer> c2 = pulsarClient.newConsumer(Schema.INT32) - .topic(topicName) - .subscriptionName(subName) - .subscriptionType(SubscriptionType.Exclusive) - .subscribe(); - c2.close(); - assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getType(), SubscriptionType.Exclusive.toString()); - - @Cleanup - final Consumer<Integer> c3 = cb.get().receiverQueueSize(0).subscribe(); - dispatcher = (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName).getDispatcher(); - individuallySentPositionsField = dispatcher.getIndividuallySentPositionsField(); - - assertNull(dispatcher.getLastSentPositionField()); - assertTrue(individuallySentPositionsField.isEmpty()); - - assertNotNull(c3.receive()); - - // validate the individuallySentPosition is initialized by the individuallyDeletedMessages - // if it is not initialized expectedly, it has sent-hole of key-c messages because key-c messages are not scheduled to be dispatched to some consumer(already acked). - assertEquals(dispatcher.getLastSentPositionField(), expectedLastSentPosition); - assertEquals(individuallySentPositionsField.toString(), expectedIndividuallySentPositions.toString()); - } - - @Test(timeOut = 30_000) - public void testLastSentPositionWhenResettingCursor() throws Exception { - // The lastSentPosition and individuallySentPositions should be cleared if reset-cursor operation is executed. - final String nsName = "public/default"; - final String topicName = "persistent://" + nsName + "/reset-cursor-" + UUID.randomUUID(); - final String subName = "my-sub"; - - final int numMessages = 10; - final List<String> keys = Arrays.asList("key-a", "key-b"); - final AtomicInteger ackCounter = new AtomicInteger(); - - @Cleanup - final Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32) - .topic(topicName) - .enableBatching(false) - .create(); - - final Supplier<ConsumerBuilder<Integer>> cb = () -> pulsarClient.newConsumer(Schema.INT32) - .topic(topicName) - .subscriptionName(subName) - .subscriptionType(SubscriptionType.Key_Shared) - .receiverQueueSize(0) - .keySharedPolicy(KeySharedPolicy.autoSplitHashRange() - .setAllowOutOfOrderDelivery(false)); - - @Cleanup - final Consumer<Integer> c1 = cb.get().consumerName("c1").subscribe(); - @Cleanup - final Consumer<Integer> c2 = cb.get().consumerName("c2").subscribe(); - - // set retention policy - admin.namespaces().setRetention(nsName, new RetentionPolicies(1, 1024 * 1024)); - - // enforce the selector will return c1 if keys.get(0) - final PersistentStickyKeyDispatcherMultipleConsumers dispatcher = - (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName).getDispatcher(); - final int hashA = Murmur3_32Hash.getInstance().makeHash(keys.get(0).getBytes()); - final Map<Integer, String> hashConsumerMap = new HashMap<>(); - hashConsumerMap.put(hashA, c1.getConsumerName()); - final Field selectorField = PersistentStickyKeyDispatcherMultipleConsumers.class.getDeclaredField("selector"); - selectorField.setAccessible(true); - final StickyKeyConsumerSelector selector = spy((StickyKeyConsumerSelector) selectorField.get(dispatcher)); - selectorField.set(dispatcher, selector); - doAnswer((invocationOnMock -> { - final int hash = invocationOnMock.getArgument(0); - final String consumerName = hashConsumerMap.getOrDefault(hash, c2.getConsumerName()); - return dispatcher.getConsumers().stream().filter(consumer -> consumer.consumerName().equals(consumerName)).findFirst().get(); - })).when(selector).select(anyInt()); - - for (int i = 0; i < numMessages; i++) { - producer.newMessage().key(keys.get(i % keys.size())).value(i).send(); - } - - // consume some messages - for (int i = 0; i < numMessages / keys.size(); i++) { - final Message<Integer> msg = c2.receive(); - if (msg != null) { - c2.acknowledge(msg); - ackCounter.getAndIncrement(); - } - } - assertEquals(ackCounter.get(), numMessages / keys.size()); - - // store current lastSentPosition for comparison - final LongPairRangeSet<Position> individuallySentPositionsField = dispatcher.getIndividuallySentPositionsField(); - assertNotNull(dispatcher.getLastSentPositionField()); - assertFalse(individuallySentPositionsField.isEmpty()); - - // reset cursor and receive a message - admin.topics().resetCursor(topicName, subName, MessageId.earliest, true); - - // validate the lastSentPosition and individuallySentPositions are cleared after resetting cursor - assertNull(dispatcher.getLastSentPositionField()); - assertTrue(individuallySentPositionsField.isEmpty()); - } - - @Test(timeOut = 30_000) - public void testLastSentPositionWhenSkipping() throws Exception { - // The lastSentPosition and individuallySentPositions should be updated if skip operation is executed. - // There are updated to follow the new markDeletedPosition. - final String topicName = "persistent://public/default/skip-" + UUID.randomUUID(); - final String subName = "my-sub"; - - final int numMessages = 10; - final List<String> keys = Arrays.asList("key-a", "key-b"); - final int numSkip = 2; - final AtomicInteger ackCounter = new AtomicInteger(); - - @Cleanup - final Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32) - .topic(topicName) - .enableBatching(false) - .create(); - - final Supplier<ConsumerBuilder<Integer>> cb = () -> pulsarClient.newConsumer(Schema.INT32) - .topic(topicName) - .subscriptionName(subName) - .subscriptionType(SubscriptionType.Key_Shared) - .keySharedPolicy(KeySharedPolicy.autoSplitHashRange() - .setAllowOutOfOrderDelivery(false)) - .receiverQueueSize(0); - - @Cleanup - final Consumer<Integer> c1 = cb.get().consumerName("c1").subscribe(); - @Cleanup - final Consumer<Integer> c2 = cb.get().consumerName("c2").subscribe(); - - // enforce the selector will return c1 if keys.get(0) - final PersistentStickyKeyDispatcherMultipleConsumers dispatcher = - (PersistentStickyKeyDispatcherMultipleConsumers) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName).getDispatcher(); - final int hashA = Murmur3_32Hash.getInstance().makeHash(keys.get(0).getBytes()); - final Map<Integer, String> hashConsumerMap = new HashMap<>(); - hashConsumerMap.put(hashA, c1.getConsumerName()); - final Field selectorField = PersistentStickyKeyDispatcherMultipleConsumers.class.getDeclaredField("selector"); - selectorField.setAccessible(true); - final StickyKeyConsumerSelector selector = spy((StickyKeyConsumerSelector) selectorField.get(dispatcher)); - selectorField.set(dispatcher, selector); - doAnswer((invocationOnMock -> { - final int hash = invocationOnMock.getArgument(0); - final String consumerName = hashConsumerMap.getOrDefault(hash, c2.getConsumerName()); - return dispatcher.getConsumers().stream().filter(consumer -> consumer.consumerName().equals(consumerName)).findFirst().get(); - })).when(selector).select(anyInt()); - - final List<Position> positionList = new ArrayList<>(); - for (int i = 0; i < numMessages; i++) { - final MessageIdImpl msgId = (MessageIdImpl) producer.newMessage().key(keys.get(i % keys.size())).value(i).send(); - positionList.add(PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId())); - } - - // consume some messages - for (int i = 0; i < numSkip; i++) { - final Message<Integer> msg = c2.receive(); - if (msg != null) { - c2.acknowledge(msg); - ackCounter.getAndIncrement(); - } - } - assertEquals(ackCounter.get(), numSkip); - final ManagedCursorImpl managedCursor = ((ManagedCursorImpl) ((PersistentSubscription) pulsar.getBrokerService().getTopicIfExists(topicName).get().get().getSubscription(subName)).getCursor()); - Awaitility.await().untilAsserted(() -> assertEquals(managedCursor.getIndividuallyDeletedMessagesSet().size(), 2)); - - // store current lastSentPosition for comparison - final Position lastSentPositionBeforeSkip = dispatcher.getLastSentPositionField(); - final LongPairRangeSet<Position> individuallySentPositionsField = dispatcher.getIndividuallySentPositionsField(); - assertNotNull(lastSentPositionBeforeSkip); - assertFalse(individuallySentPositionsField.isEmpty()); - - // skip messages and receive a message - admin.topics().skipMessages(topicName, subName, numSkip); - final MessageIdImpl msgIdAfterSkip = (MessageIdImpl) c1.receive().getMessageId(); - final Position positionAfterSkip = PositionFactory.create(msgIdAfterSkip.getLedgerId(), - msgIdAfterSkip.getEntryId()); - assertEquals(positionAfterSkip, positionList.get(4)); - - // validate the lastSentPosition is updated to the new markDeletedPosition - // validate the individuallySentPositions is updated expectedly (removeAtMost the new markDeletedPosition) - final Position lastSentPosition = dispatcher.getLastSentPositionField(); - assertNotNull(lastSentPosition); - assertTrue(lastSentPosition.compareTo(lastSentPositionBeforeSkip) > 0); - assertEquals(lastSentPosition, positionList.get(4)); - assertTrue(individuallySentPositionsField.isEmpty()); - } private KeySharedMode getKeySharedModeOfSubscription(Topic topic, String subscription) { if (TopicName.get(topic.getName()).getDomain().equals(TopicDomain.persistent)) { diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java index 5f2cf7b209e..d2d3600df96 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ConsumerStats.java @@ -72,8 +72,8 @@ public interface ConsumerStats { /** Flag to verify if consumer is blocked due to reaching threshold of unacked messages. */ boolean isBlockedConsumerOnUnackedMsgs(); - /** The last sent position of the cursor when the consumer joining. */ - String getLastSentPositionWhenJoining(); + /** The read position of the cursor when the consumer joining. */ + String getReadPositionWhenJoining(); /** Address of this consumer. */ String getAddress(); diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java index cabef1ca960..d4850adaa6f 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java @@ -118,12 +118,6 @@ public interface SubscriptionStats { /** This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription. */ Map<String, String> getConsumersAfterMarkDeletePosition(); - /** The last sent position of the cursor. This is for Key_Shared subscription. */ - String getLastSentPosition(); - - /** Set of individually sent ranges. This is for Key_Shared subscription. */ - String getIndividuallySentPositions(); - /** SubscriptionProperties (key/value strings) associated with this subscribe. */ Map<String, String> getSubscriptionProperties(); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java index b4c5d21e692..de36b330b7f 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/ConsumerStatsImpl.java @@ -77,8 +77,8 @@ public class ConsumerStatsImpl implements ConsumerStats { /** Flag to verify if consumer is blocked due to reaching threshold of unacked messages. */ public boolean blockedConsumerOnUnackedMsgs; - /** The last sent position of the cursor when the consumer joining. */ - public String lastSentPositionWhenJoining; + /** The read position of the cursor when the consumer joining. */ + public String readPositionWhenJoining; /** Address of this consumer. */ private String address; @@ -113,7 +113,7 @@ public class ConsumerStatsImpl implements ConsumerStats { this.availablePermits += stats.availablePermits; this.unackedMessages += stats.unackedMessages; this.blockedConsumerOnUnackedMsgs = stats.blockedConsumerOnUnackedMsgs; - this.lastSentPositionWhenJoining = stats.lastSentPositionWhenJoining; + this.readPositionWhenJoining = stats.readPositionWhenJoining; return this; } @@ -141,8 +141,8 @@ public class ConsumerStatsImpl implements ConsumerStats { this.clientVersion = clientVersion; } - public String getLastSentPositionWhenJoining() { - return lastSentPositionWhenJoining; + public String getReadPositionWhenJoining() { + return readPositionWhenJoining; } public String getLastAckedTime() { diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java index ab4d07c7ae4..a8ea0060629 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java @@ -126,12 +126,6 @@ public class SubscriptionStatsImpl implements SubscriptionStats { /** This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription. */ public Map<String, String> consumersAfterMarkDeletePosition; - /** The last sent position of the cursor. This is for Key_Shared subscription. */ - public String lastSentPosition; - - /** Set of individually sent ranges. This is for Key_Shared subscription. */ - public String individuallySentPositions; - /** The number of non-contiguous deleted messages ranges. */ public int nonContiguousDeletedMessagesRanges;
