You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/05/02 13:51:25 UTC
[GitHub] [pulsar] Jason918 commented on a diff in pull request #15409: Fixed key-shared delivery of messages with interleaved delays
Jason918 commented on code in PR #15409:
URL: https://github.com/apache/pulsar/pull/15409#discussion_r862863547
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java:
##########
@@ -174,29 +174,36 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
// This may happen when consumer closed. See issue #12885 for details.
if (!allowOutOfOrderDelivery) {
Set<PositionImpl> messagesToReplayNow = this.getMessagesToReplayNow(1);
- if (messagesToReplayNow != null && !messagesToReplayNow.isEmpty() && this.minReplayedPosition != null) {
- PositionImpl relayPosition = messagesToReplayNow.stream().findFirst().get();
- // If relayPosition is a new entry wither smaller position is inserted for redelivery during this async
- // read, it is possible that this relayPosition should dispatch to consumer first. So in order to
- // preserver order delivery, we need to discard this read result, and try to trigger a replay read,
- // that containing "relayPosition", by calling readMoreEntries.
- if (relayPosition.compareTo(minReplayedPosition) < 0) {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Position {} (<{}) is inserted for relay during current {} read, discard this "
- + "read and retry with readMoreEntries.",
- name, relayPosition, minReplayedPosition, readType);
- }
- if (readType == ReadType.Normal) {
- entries.forEach(entry -> {
- long stickyKeyHash = getStickyKeyHash(entry);
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
- entry.release();
- });
- } else if (readType == ReadType.Replay) {
- entries.forEach(Entry::release);
+ if (messagesToReplayNow != null && !messagesToReplayNow.isEmpty()) {
+ PositionImpl replayPosition = messagesToReplayNow.stream().findFirst().get();
+ if (this.minReplayedPosition != null) {
+ // If relayPosition is a new entry wither smaller position is inserted for redelivery during this
+ // async read, it is possible that this relayPosition should dispatch to consumer first. So in
+ // order to preserver order delivery, we need to discard this read result, and try to trigger a
+ // replay read, that containing "relayPosition", by calling readMoreEntries.
+ if (replayPosition.compareTo(minReplayedPosition) < 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Position {} (<{}) is inserted for relay during current {} read, "
+ + "discard this read and retry with readMoreEntries.",
+ name, replayPosition, minReplayedPosition, readType);
+ }
+ if (readType == ReadType.Normal) {
+ entries.forEach(entry -> {
+ long stickyKeyHash = getStickyKeyHash(entry);
+ addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ entry.release();
+ });
+ } else if (readType == ReadType.Replay) {
+ entries.forEach(Entry::release);
+ }
+ readMoreEntries();
+ return;
}
- readMoreEntries();
- return;
+ } else {
Review Comment:
Why only do this in `else` part？
It seems `replayPosition` could be popped from `delayedDeliveryTracker` if `minReplayedPosition` is null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org