shibd commented on code in PR #24730:
URL: https://github.com/apache/pulsar/pull/24730#discussion_r2343311221
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java:
##########
@@ -143,34 +153,38 @@ private synchronized CompletableFuture<Void>
validateKeySharedMeta(Consumer cons
}
}
- private synchronized Consumer findConflictingConsumer(List<IntRange>
ranges) {
- for (IntRange intRange : ranges) {
- Map.Entry<Integer, Consumer> ceilingEntry =
rangeMap.ceilingEntry(intRange.getStart());
- Map.Entry<Integer, Consumer> floorEntry =
rangeMap.floorEntry(intRange.getEnd());
-
- if (floorEntry != null && floorEntry.getKey() >=
intRange.getStart()) {
- return floorEntry.getValue();
- }
-
- if (ceilingEntry != null && ceilingEntry.getKey() <=
intRange.getEnd()) {
- return ceilingEntry.getValue();
+ private synchronized Consumer findConflictingConsumer(List<IntRange>
newConsumerRanges) {
+ for (IntRange newRange : newConsumerRanges) {
+ // 1. Check for potential conflicts with existing ranges that
start before newRange's start.
+ Map.Entry<Integer, Pair<Range, Consumer>> conflictBeforeStart =
rangeMap.floorEntry(newRange.getStart());
+ if (conflictBeforeStart != null) {
+ Range existingRange = conflictBeforeStart.getValue().getLeft();
+ if (checkRangesOverlap(newRange, existingRange)) {
+ return conflictBeforeStart.getValue().getRight();
+ }
}
-
- if (ceilingEntry != null && floorEntry != null &&
ceilingEntry.getValue().equals(floorEntry.getValue())) {
- KeySharedMeta keySharedMeta =
ceilingEntry.getValue().getKeySharedMeta();
- for (IntRange range : keySharedMeta.getHashRangesList()) {
- int start = Math.max(intRange.getStart(),
range.getStart());
- int end = Math.min(intRange.getEnd(), range.getEnd());
- if (end >= start) {
- return ceilingEntry.getValue();
- }
+ // 2. Check for potential conflicts with existing ranges that
start after newRange's start.
+ Map.Entry<Integer, Pair<Range, Consumer>> conflictAfterStart =
rangeMap.ceilingEntry(newRange.getStart());
+ if (conflictAfterStart != null) {
+ Range existingRange = conflictAfterStart.getValue().getLeft();
+ if (checkRangesOverlap(newRange, existingRange)) {
+ return conflictAfterStart.getValue().getRight();
}
}
}
return null;
}
- Map<Integer, Consumer> getRangeConsumer() {
+
+ private boolean checkRangesOverlap(IntRange range1, Range range2) {
+ return Math.max(range1.getStart(), range2.getStart()) <=
Math.min(range1.getEnd(), range2.getEnd());
Review Comment:
I'm wondering if allowing this might create some ambiguity. For instance,
which range would a value like 100 fall into?
I tested it, and this wasn't allowed before this PR, and the client-side
also prevents this. So, the behavior is consistent with how it was before.
```
java.lang.IllegalArgumentException: Ranges for KeyShared policy with overlap
between [0, 9999] and [9999, 29999]
at
org.apache.pulsar.client.api.KeySharedPolicy$KeySharedPolicySticky.validate(KeySharedPolicy.java:118)
at
org.apache.pulsar.client.impl.ConsumerBuilderImpl.keySharedPolicy(ConsumerBuilderImpl.java:525)
at
org.apache.pulsar.client.api.KeySharedSubscriptionTest.testCustomStickyRange(KeySharedSubscriptionTest.java:2579)
```
https://github.com/apache/pulsar/blob/14129e352e4bb4e06cbf2c6c581edcdbaf9f6775/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/KeySharedPolicy.java#L105-L123
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
