{"record":{"id":"1bfd08a4f039e34b","repo":"apache/pulsar","slug":"range-conflict-with-consumer-conflictingconsumer","errorCode":null,"errorMessage":"Range conflict with consumer ${conflictingConsumer}","messagePattern":"Range conflict with consumer (.+?)","errorType":"exception","errorClass":"BrokerServiceException.ConsumerAssignException","httpStatus":null,"severity":"error","filePath":"pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java","lineNumber":73,"sourceCode":"        this.rangeMap = new ConcurrentSkipListMap<>();\n    }\n\n    @Override\n    public synchronized CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer consumer) {\n        return validateKeySharedMeta(consumer).thenApply(__ -> {\n            try {\n                return internalAddConsumer(consumer);\n            } catch (BrokerServiceException.ConsumerAssignException e) {\n                throw FutureUtil.wrapToCompletionException(e);\n            }\n        });\n    }\n\n    private synchronized Optional<ImpactedConsumersResult> internalAddConsumer(Consumer consumer)\n            throws BrokerServiceException.ConsumerAssignException {\n        Consumer conflictingConsumer = findConflictingConsumer(consumer.getKeySharedMeta().getHashRangesList());\n        if (conflictingConsumer != null) {\n            throw new BrokerServiceException.ConsumerAssignException(\"Range conflict with consumer \"\n                    + conflictingConsumer);\n        }\n        for (IntRange intRange : consumer.getKeySharedMeta().getHashRangesList()) {\n            rangeMap.put(intRange.getStart(), Pair.of(Range.of(intRange.getStart(), intRange.getEnd()), consumer));\n        }\n        return Optional.empty();\n    }\n\n    @Override\n    public synchronized Optional<ImpactedConsumersResult> removeConsumer(Consumer consumer) {\n        rangeMap.entrySet().removeIf(entry -> entry.getValue().getRight().equals(consumer));\n        return Optional.empty();\n    }\n\n    @Override\n    public synchronized ConsumerHashAssignmentsSnapshot getConsumerHashAssignmentsSnapshot() {\n        List<HashRangeAssignment> result = new ArrayList<>();\n        for (Map.Entry<Integer, Pair<Range, Consumer>> entry : rangeMap.entrySet()) {","sourceCodeStart":55,"sourceCodeEnd":91,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeExclusiveStickyKeyConsumerSelector.java#L55-L91","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Adjust the consumer's key_shared hash ranges so they don't overlap existing assignments (query current ranges or coordinate out-of-band)","Disconnect/remove the conflicting consumer before subscribing with those ranges","Catch ConsumerAssignException and retry with a corrected range configuration","Auto-assign ranges (omit explicit hashRanges) instead of manual range specification"],"exampleFix":"// before\nKeySharedMeta ksm = KeySharedMeta.newHashRangeRequest(\n    List.of(IntRange.of(0, 65535))); // overlaps other consumer\n// after\nKeySharedMeta ksm = KeySharedMeta.newHashRangeRequest(\n    List.of(IntRange.of(32768, 65535))); // disjoint range","handlingStrategy":"validation","validationCode":"// ensure requested ranges don't overlap ranges already assigned\nSet<IntRange> requested = ...; Set<IntRange> existing = ...;\nfor (IntRange r : requested) {\n    if (existing.stream().anyMatch(x -> x.getStart() <= r.getEnd() && r.getStart() <= x.getEnd()))\n        throw new IllegalStateException(\"overlap with existing range\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    consumer = pulsarClient.newConsumer().keySharedPolicy(\n        KeySharedPolicy.stickyHashRange().ranges(IntRange.of(0, 32767))).subscribe();\n} catch (PulsarClientException e) {\n    if (e.getCause() instanceof ConsumerAssignException) { /* recompute ranges */ }\n}","preventionTips":["Coordinate explicit hash ranges centrally so each consumer gets disjoint ranges","Prefer auto-split key sharing unless manual ranges are required","Document range ownership per consumer instance to avoid copy-paste overlaps"],"tags":["java","broker","key-shared","hash-range","consumer-assignment"],"backgroundTag":"hash-range-conflict","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}