poorbarcode commented on code in PR #24730:
URL: https://github.com/apache/pulsar/pull/24730#discussion_r2341267490
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java:
##########
@@ -129,10 +124,25 @@ private synchronized CompletableFuture<Void>
validateKeySharedMeta(Consumer cons
return FutureUtil.failedFuture(new
BrokerServiceException.ConsumerAssignException(
"Ranges for KeyShared policy must not be empty."));
}
- for (IntRange intRange : ranges) {
- if (intRange.getStart() > intRange.getEnd()) {
+ List<IntRange> sortedRanges = new ArrayList<>(ranges);
+ sortedRanges.sort(Comparator.comparingInt(IntRange::getStart));
+ for (int i = 0; i < sortedRanges.size(); i++) {
+ IntRange currentRange = sortedRanges.get(i);
+ // 1. Validate: check if start > end for the current range
+ if (currentRange.getStart() > currentRange.getEnd()) {
return FutureUtil.failedFuture(
- new
BrokerServiceException.ConsumerAssignException("Fixed hash range start > end"));
+ new
BrokerServiceException.ConsumerAssignException("Fixed hash range start > end
for range: "
+ + "[" + currentRange.getStart() + "," +
currentRange.getEnd() + "]"));
+ }
+ // 2. Validate: check for overlaps with the next range in the
sorted list
+ if (i < sortedRanges.size() - 1) {
+ IntRange nextRange = sortedRanges.get(i + 1);
Review Comment:
When `i == 0 and sortedRanges.size == 1`, the line may face an
`IndexOutOfBoundsException`
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java:
##########
@@ -73,47 +74,41 @@ private synchronized Optional<ImpactedConsumersResult>
internalAddConsumer(Consu
+ conflictingConsumer);
}
for (IntRange intRange :
consumer.getKeySharedMeta().getHashRangesList()) {
- rangeMap.put(intRange.getStart(), consumer);
- rangeMap.put(intRange.getEnd(), consumer);
+ rangeMap.put(intRange.getStart(),
Pair.of(Range.of(intRange.getStart(), intRange.getEnd()), consumer));
}
return Optional.empty();
}
@Override
public synchronized Optional<ImpactedConsumersResult>
removeConsumer(Consumer consumer) {
- rangeMap.entrySet().removeIf(entry ->
entry.getValue().equals(consumer));
+ rangeMap.entrySet().removeIf(entry ->
entry.getValue().getRight().equals(consumer));
return Optional.empty();
}
@Override
public synchronized ConsumerHashAssignmentsSnapshot
getConsumerHashAssignmentsSnapshot() {
List<HashRangeAssignment> result = new ArrayList<>();
- Map.Entry<Integer, Consumer> prev = null;
- for (Map.Entry<Integer, Consumer> entry: rangeMap.entrySet()) {
- if (prev == null) {
- prev = entry;
- } else {
- if (prev.getValue().equals(entry.getValue())) {
- result.add(new HashRangeAssignment(Range.of(prev.getKey(),
entry.getKey()), entry.getValue()));
- }
- prev = null;
- }
+ for (Map.Entry<Integer, Pair<Range, Consumer>> entry :
rangeMap.entrySet()) {
+ Range assignedRange = entry.getValue().getLeft();
+ Consumer assignedConsumer = entry.getValue().getRight();
+ result.add(new HashRangeAssignment(assignedRange,
assignedConsumer));
Review Comment:
Since the method `getConsumerKeyHashRanges()` is the unique caller of the
method `getConsumerHashAssignmentsSnapshot()`, we can overwrite the method
`getConsumerKeyHashRanges()` in this class, which is simpler: loop consumers
and put all `consumer.getKeySharedMeta().getHashRangesList()`
##########
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:
Can we allow the intersection of one point between two ranges like follows?
- `range 1`: `[0-100]`
- `range 2`: `[100-200]`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
