apache/pulsar · error · BrokerServiceException.ConsumerAssignException

Range conflict with consumer ${conflictingConsumer}

Error message

Range conflict with consumer ${conflictingConsumer}

What it means

For key_shared subscriptions with explicitly specified hash ranges (KeySharedMeta hashRanges), internalAddConsumer checks for overlaps with existing consumers' ranges. If a requested range overlaps one already assigned, it throws ConsumerAssignException naming the conflicting consumer.

Source

Thrown at pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java:73

        this.rangeMap = new ConcurrentSkipListMap<>();
    }

    @Override
    public synchronized CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer consumer) {
        return validateKeySharedMeta(consumer).thenApply(__ -> {
            try {
                return internalAddConsumer(consumer);
            } catch (BrokerServiceException.ConsumerAssignException e) {
                throw FutureUtil.wrapToCompletionException(e);
            }
        });
    }

    private synchronized Optional<ImpactedConsumersResult> internalAddConsumer(Consumer consumer)
            throws BrokerServiceException.ConsumerAssignException {
        Consumer conflictingConsumer = findConflictingConsumer(consumer.getKeySharedMeta().getHashRangesList());
        if (conflictingConsumer != null) {
            throw new BrokerServiceException.ConsumerAssignException("Range conflict with consumer "
                    + conflictingConsumer);
        }
        for (IntRange intRange : consumer.getKeySharedMeta().getHashRangesList()) {
            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().getRight().equals(consumer));
        return Optional.empty();
    }

    @Override
    public synchronized ConsumerHashAssignmentsSnapshot getConsumerHashAssignmentsSnapshot() {
        List<HashRangeAssignment> result = new ArrayList<>();
        for (Map.Entry<Integer, Pair<Range, Consumer>> entry : rangeMap.entrySet()) {

View on GitHub (pinned to 820761864e)

Solutions

1. Adjust the consumer's key_shared hash ranges so they don't overlap existing assignments (query current ranges or coordinate out-of-band)
2. Disconnect/remove the conflicting consumer before subscribing with those ranges
3. Catch ConsumerAssignException and retry with a corrected range configuration
4. Auto-assign ranges (omit explicit hashRanges) instead of manual range specification

Example fix

// before
KeySharedMeta ksm = KeySharedMeta.newHashRangeRequest(
    List.of(IntRange.of(0, 65535))); // overlaps other consumer
// after
KeySharedMeta ksm = KeySharedMeta.newHashRangeRequest(
    List.of(IntRange.of(32768, 65535))); // disjoint range

// ensure requested ranges don't overlap ranges already assigned
Set<IntRange> requested = ...; Set<IntRange> existing = ...;
for (IntRange r : requested) {
    if (existing.stream().anyMatch(x -> x.getStart() <= r.getEnd() && r.getStart() <= x.getEnd()))
        throw new IllegalStateException("overlap with existing range");
}

try {
    consumer = pulsarClient.newConsumer().keySharedPolicy(
        KeySharedPolicy.stickyHashRange().ranges(IntRange.of(0, 32767))).subscribe();
} catch (PulsarClientException e) {
    if (e.getCause() instanceof ConsumerAssignException) { /* recompute ranges */ }
}

When it happens

Trigger: Calling consumerBuilder.subscribe()/subscribeAsync() on a key_shared subscription with keySharedMeta hash ranges that intersect ranges already claimed by another consumer of the same subscription.

Common situations: Two clients configured with overlapping/duplicate explicit hash ranges; copy-pasted range configuration across consumers; ranges recalculated after a consumer joined without re-checking current assignments.

Related errors

• No more range can assigned to new consumer, assigned consume
• ${key} already exists in the dynamicConfigurationMap
• Topic factory failed to create topic
• Error creating client for HealthChecker
• configuredService should not be an instance of SystemTopicBa

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/1bfd08a4f039e34b. Report an issue: GitHub.