PIP-22: Dead Letter Topic #2508
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| /** | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pulsar.broker.service; | ||
|
|
||
| import org.apache.bookkeeper.mledger.Position; | ||
|
|
||
| import java.util.List; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| public class InMemoryRedeliveryTracker implements RedeliveryTracker { | ||
|
|
||
| private ConcurrentHashMap<Position, AtomicInteger> trackerCache = new ConcurrentHashMap<>(16); | ||
|
|
||
| @Override | ||
| public int incrementAndGetRedeliveryCount(Position position) { | ||
| trackerCache.putIfAbsent(position, new AtomicInteger(0)); | ||
| return trackerCache.get(position).incrementAndGet(); | ||
| } | ||
|
|
||
| @Override | ||
| public int getRedeliveryCount(Position position) { | ||
| return trackerCache.getOrDefault(position, new AtomicInteger(0)).get(); | ||
| } | ||
|
|
||
| @Override | ||
| public void remove(Position position) { | ||
| trackerCache.remove(position); | ||
| } | ||
|
|
||
| @Override | ||
| public void removeBatch(List<Position> positions) { | ||
| if (positions != null) { | ||
| positions.forEach(this::remove); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void clear() { | ||
| trackerCache.clear(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| /** | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pulsar.broker.service; | ||
|
|
||
| import org.apache.bookkeeper.mledger.Position; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public interface RedeliveryTracker { | ||
|
|
||
| int incrementAndGetRedeliveryCount(Position position); | ||
|
|
||
| int getRedeliveryCount(Position position); | ||
|
|
||
| void remove(Position position); | ||
|
|
||
| void removeBatch(List<Position> positions); | ||
|
|
||
| void clear(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /** | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pulsar.broker.service; | ||
|
|
||
| import org.apache.bookkeeper.mledger.Position; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public class RedeliveryTrackerDisabled implements RedeliveryTracker { | ||
|
|
||
| public static final RedeliveryTrackerDisabled REDELIVERY_TRACKER_DISABLED = new RedeliveryTrackerDisabled(); | ||
|
|
||
| private RedeliveryTrackerDisabled() {} | ||
|
|
||
| @Override | ||
| public int incrementAndGetRedeliveryCount(Position position) { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public int getRedeliveryCount(Position position) { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public void remove(Position position) { | ||
| // no-op | ||
| } | ||
|
|
||
| @Override | ||
| public void removeBatch(List<Position> positions) { | ||
| // no-op | ||
| } | ||
|
|
||
| @Override | ||
| public void clear() { | ||
| // no-op | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Instead of keeping a separate redelivery tracker, would it be possible to reuse the
pendingAcksmaps on theConsumerclass?pendingAcksis a map for which the key is a pair longs and the value is a pair of longs as well.(ledgerId, entryId) --> (batchSize, none)As you can see, the second
longfield in the value is currently unused. We could keep the counter there.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
After i check the pending acks in consumer, i recommend use redeliveryTracker to track the redelivery count:
First, in a subscription only have one redelivery tracker, pending acks is binding to consumer, we don't need to record redelivery count per consumer and if record it in consumer, will get a different behavior is
max redelivery count per consumer, but for now we defined the redelivery count per subscription.Second, i think we need separate pending acks and redeliveryTracker, it's better to implement another redeliveryTracker.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sure, what I was thinking was more to store it in consumer pending acks and carrying it over to the new consumer when redelivering.
Sure, these are separate concepts. Though I think the current name
pendingAcksis a bit misleading. What the map really tracks is "message that were pushed to this consumer and not acked".My main concern with the new map is to have to potentially accumulate lots of items in the concurrent hash map, irrespective of whether the DLQ is being used or not. Especially, this might be mostly not the case in exclusive/failover subscriptions, where data is typically processed in order.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm understand your concern, i think the tracker map is limited by maxUnackMessages, only put element into tracker map when trigger by redeliverUnackedMessages().
I think we can optimize it next step, i think we can add a field in proto like
enableRedeliveryTracker, broker just enable this mechanism whenenableRedeliveryTrackeris true.Yes, i'm agree, i think we can only enable the counter tracking only on Shared subscriptions, mean that just in PersistentDispatcherMultipleConsumers.