You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/04/07 15:27:30 UTC
(pulsar) branch master updated: [feat] PIP-460: Add topic:// and segment:// domain support to pulsar-common (#25456)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 14a811b0d7a [feat] PIP-460: Add topic:// and segment:// domain support to pulsar-common (#25456)
14a811b0d7a is described below
commit 14a811b0d7a5bc3758292d85e1f7f4b66b16412a
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Apr 7 08:27:24 2026 -0700
[feat] PIP-460: Add topic:// and segment:// domain support to pulsar-common (#25456)
---
.../apache/pulsar/common/naming/TopicDomain.java | 3 +-
.../pulsar/client/impl/PulsarClientImpl.java | 23 +++
.../org/apache/pulsar/common/naming/TopicName.java | 109 ++++++++++-
.../apache/pulsar/common/scalable/HashRange.java | 124 +++++++++++++
.../apache/pulsar/common/scalable/SegmentInfo.java | 98 ++++++++++
.../pulsar/common/scalable/SegmentState.java | 39 +---
.../pulsar/common/scalable/SegmentTopicName.java | 106 +++++++++++
.../pulsar/common/scalable/package-info.java | 33 +---
.../apache/pulsar/common/naming/TopicNameTest.java | 205 +++++++++++++++++++++
.../pulsar/common/scalable/HashRangeTest.java | 131 +++++++++++++
.../common/scalable/SegmentTopicNameTest.java | 135 ++++++++++++++
11 files changed, 940 insertions(+), 66 deletions(-)
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
index 0f7ccfe9125..82acd4a06df 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
@@ -22,7 +22,8 @@ package org.apache.pulsar.common.naming;
* Enumeration showing if a topic is persistent.
*/
public enum TopicDomain {
- persistent("persistent"), non_persistent("non-persistent");
+ persistent("persistent"), non_persistent("non-persistent"),
+ topic("topic"), segment("segment");
private String value;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index c7a43a912a5..c1034925ce1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -415,6 +415,12 @@ public class PulsarClientImpl implements PulsarClient {
return FutureUtil.failedFuture(
new PulsarClientException.InvalidTopicNameException("Invalid topic name: '" + topic + "'"));
}
+ if (isScalableDomain(topic)) {
+ return FutureUtil.failedFuture(
+ new PulsarClientException.InvalidTopicNameException(
+ "Scalable topic domains (topic://, segment://) require the V5 client SDK."
+ + " Topic: '" + topic + "'"));
+ }
if (schema instanceof AutoProduceBytesSchema) {
AutoProduceBytesSchema autoProduceBytesSchema = (AutoProduceBytesSchema) schema;
@@ -577,6 +583,12 @@ public class PulsarClientImpl implements PulsarClient {
return FutureUtil.failedFuture(
new PulsarClientException.InvalidTopicNameException("Invalid topic name: '" + topic + "'"));
}
+ if (isScalableDomain(topic)) {
+ return FutureUtil.failedFuture(
+ new PulsarClientException.InvalidTopicNameException(
+ "Scalable topic domains (topic://, segment://) require the V5 client SDK."
+ + " Topic: '" + topic + "'"));
+ }
}
if (isBlank(conf.getSubscriptionName())) {
@@ -742,6 +754,12 @@ public class PulsarClientImpl implements PulsarClient {
return FutureUtil.failedFuture(new PulsarClientException
.InvalidTopicNameException("Invalid topic name: '" + topic + "'"));
}
+ if (isScalableDomain(topic)) {
+ return FutureUtil.failedFuture(
+ new PulsarClientException.InvalidTopicNameException(
+ "Scalable topic domains (topic://, segment://) require the V5 client SDK."
+ + " Topic: '" + topic + "'"));
+ }
}
if (conf.getStartMessageId() == null) {
@@ -1431,4 +1449,9 @@ public class PulsarClientImpl implements PulsarClient {
NameResolver<InetAddress> getNameResolver() {
return DnsResolverUtil.adaptToNameResolver(addressResolver);
}
+
+ private static boolean isScalableDomain(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ return topicName.isScalable() || topicName.isSegment();
+ }
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index a1c7055a897..55042f23634 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -24,8 +24,10 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.util.Codec;
/**
@@ -49,6 +51,10 @@ public class TopicName implements ServiceUnitId {
private final int partitionIndex;
+ // segment:// fields (null for non-segment domains)
+ private final HashRange segmentRange;
+ private final long segmentId;
+
private static final ConcurrentHashMap<String, TopicName> cache = new ConcurrentHashMap<>();
public static void clearIfReachedMaxCapacity(int maxCapacity) {
@@ -132,8 +138,12 @@ public class TopicName implements ServiceUnitId {
String rest = parts.get(1);
- // Expected format: tenant/namespace/<localName>
- parts = Splitter.on("/").limit(4).splitToList(rest);
+ // Scalable topic domains (topic://, segment://) only support the new format
+ // and local names may contain '/', so use limit(3) to keep the rest as localName.
+ boolean isScalableDomain = this.domain == TopicDomain.topic
+ || this.domain == TopicDomain.segment;
+ int splitLimit = isScalableDomain ? 3 : 4;
+ parts = Splitter.on("/").limit(splitLimit).splitToList(rest);
if (parts.size() == 4) {
throw new IllegalArgumentException(
"V1 topic names (with cluster component) are no longer supported. "
@@ -142,7 +152,35 @@ public class TopicName implements ServiceUnitId {
} else if (parts.size() == 3) {
this.tenant = parts.get(0);
this.namespacePortion = parts.get(1);
- this.localName = parts.get(2);
+ String rawLocalName = parts.get(2);
+
+ // For segment:// domains, split local name into parent topic name + descriptor
+ if (this.domain == TopicDomain.segment) {
+ int lastSlash = rawLocalName.lastIndexOf('/');
+ if (lastSlash <= 0) {
+ throw new IllegalArgumentException(
+ "Invalid segment topic name: local name must contain"
+ + " '<parent-topic>/<hashStart>-<hashEnd>-<segmentId>'. Got: "
+ + completeTopicName);
+ }
+ this.localName = rawLocalName.substring(0, lastSlash);
+ String descriptor = rawLocalName.substring(lastSlash + 1);
+ String[] descParts = descriptor.split("-");
+ if (descParts.length != 3) {
+ throw new IllegalArgumentException(
+ "Invalid segment descriptor: expected '<hexStart>-<hexEnd>-<segmentId>',"
+ + " got: '" + descriptor + "'");
+ }
+ this.segmentRange = HashRange.of(
+ Integer.parseInt(descParts[0], 16),
+ Integer.parseInt(descParts[1], 16));
+ this.segmentId = Long.parseLong(descParts[2]);
+ } else {
+ this.localName = rawLocalName;
+ this.segmentRange = null;
+ this.segmentId = -1;
+ }
+
this.partitionIndex = getPartitionIndex(completeTopicName);
this.namespaceName = NamespaceName.get(tenant, namespacePortion);
} else {
@@ -157,12 +195,56 @@ public class TopicName implements ServiceUnitId {
} catch (NullPointerException e) {
throw new IllegalArgumentException("Invalid topic name: " + completeTopicName, e);
}
- this.completeTopicName = String.format("%s://%s/%s/%s",
- domain, tenant, namespacePortion, localName);
+ if (this.domain == TopicDomain.segment) {
+ this.completeTopicName = String.format("%s://%s/%s/%s/%s",
+ domain, tenant, namespacePortion, localName,
+ String.format("%04x-%04x-%d", segmentRange.start(), segmentRange.end(), segmentId));
+ } else {
+ this.completeTopicName = String.format("%s://%s/%s/%s",
+ domain, tenant, namespacePortion, localName);
+ }
}
public boolean isPersistent() {
- return TopicDomain.persistent == domain;
+ return TopicDomain.persistent == domain || TopicDomain.topic == domain || TopicDomain.segment == domain;
+ }
+
+ public boolean isScalable() {
+ return TopicDomain.topic == domain;
+ }
+
+ public boolean isSegment() {
+ return TopicDomain.segment == domain;
+ }
+
+ /**
+ * Get the segment hash range for segment:// topics.
+ *
+ * @return the hash range, or empty for non-segment domains
+ */
+ public Optional<HashRange> getSegmentRange() {
+ return Optional.ofNullable(segmentRange);
+ }
+
+ /**
+ * Get the segment ID for segment:// topics.
+ *
+ * @return the segment ID, or -1 for non-segment domains
+ */
+ public long getSegmentId() {
+ return segmentId;
+ }
+
+ /**
+ * Get the segment descriptor string (e.g. "0000-7fff-1") for segment:// topics.
+ *
+ * @return the descriptor, or null for non-segment domains
+ */
+ public String getSegmentDescriptor() {
+ if (segmentRange == null) {
+ return null;
+ }
+ return String.format("%04x-%04x-%d", segmentRange.start(), segmentRange.end(), segmentId);
}
/**
@@ -299,6 +381,14 @@ public class TopicName implements ServiceUnitId {
public String getPersistenceNamingEncoding() {
// The convention is: domain://tenant/namespace/topic
// We want to persist in the order: tenant/namespace/domain/topic
+ // For segment topics, include the segment descriptor: tenant/namespace/segment/topic/descriptor
+ if (domain == TopicDomain.segment && segmentRange != null) {
+ return String.format("%s/%s/%s/%s/%04x-%04x-%d", tenant, namespacePortion, domain,
+ getEncodedLocalName(), segmentRange.start(), segmentRange.end(), segmentId);
+ }
+ if (domain == TopicDomain.topic) {
+ return String.format("%s/%s/%s", tenant, namespacePortion, getEncodedLocalName());
+ }
return String.format("%s/%s/%s/%s", tenant, namespacePortion, domain, getEncodedLocalName());
}
@@ -325,6 +415,13 @@ public class TopicName implements ServiceUnitId {
localName = Codec.decode(parts.get(3));
return String.format("%s://%s/%s/%s", domain, tenant, namespacePortion, localName);
} else if (parts.size() == 5) {
+ if ("segment".equals(parts.get(2))) {
+ // Segment topic ML name: tenant/namespace/segment/topic/descriptor
+ tenant = parts.get(0);
+ namespacePortion = parts.get(1);
+ localName = Codec.decode(parts.get(3));
+ return String.format("segment://%s/%s/%s/%s", tenant, namespacePortion, localName, parts.get(4));
+ }
// Legacy V1 managed ledger name: tenant/cluster/namespace/domain/topic
// Convert to V2 format, dropping the cluster component
tenant = parts.get(0);
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/HashRange.java b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/HashRange.java
new file mode 100644
index 00000000000..8faf682a396
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/HashRange.java
@@ -0,0 +1,124 @@
+/*
+ * 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.common.scalable;
+
+/**
+ * Represents an inclusive hash range [start, end] within a 16-bit hash space (0x0000-0xFFFF).
+ */
+public record HashRange(int start, int end) implements Comparable<HashRange> {
+
+ public static final int MIN_HASH = 0x0000;
+ public static final int MAX_HASH = 0xFFFF;
+
+ public HashRange {
+ if (start < MIN_HASH || start > MAX_HASH) {
+ throw new IllegalArgumentException("start must be in [0x0000, 0xFFFF], got: " + start);
+ }
+ if (end < MIN_HASH || end > MAX_HASH) {
+ throw new IllegalArgumentException("end must be in [0x0000, 0xFFFF], got: " + end);
+ }
+ if (end < start) {
+ throw new IllegalArgumentException("end must be >= start, got: [" + start + ", " + end + "]");
+ }
+ }
+
+ public static HashRange of(int start, int end) {
+ return new HashRange(start, end);
+ }
+
+ public static HashRange full() {
+ return new HashRange(MIN_HASH, MAX_HASH);
+ }
+
+ public boolean contains(int hash) {
+ return hash >= start && hash <= end;
+ }
+
+ public boolean contains(HashRange other) {
+ return start <= other.start && end >= other.end;
+ }
+
+ public boolean isAdjacentTo(HashRange other) {
+ return this.end + 1 == other.start || other.end + 1 == this.start;
+ }
+
+ public int size() {
+ return end - start + 1;
+ }
+
+ /**
+ * Split this range at the midpoint into two sub-ranges.
+ *
+ * @return array of two HashRange objects: [start, mid] and [mid+1, end]
+ * @throws IllegalStateException if the range cannot be split (size < 2)
+ */
+ public HashRange[] split() {
+ if (size() < 2) {
+ throw new IllegalStateException("Cannot split range of size " + size());
+ }
+ int mid = start + (end - start) / 2;
+ return new HashRange[]{
+ new HashRange(start, mid),
+ new HashRange(mid + 1, end)
+ };
+ }
+
+ /**
+ * Merge this range with an adjacent range.
+ *
+ * @param other the adjacent range to merge with
+ * @return the merged range
+ * @throws IllegalArgumentException if ranges are not adjacent
+ */
+ public HashRange merge(HashRange other) {
+ if (this.end + 1 == other.start) {
+ return new HashRange(this.start, other.end);
+ } else if (other.end + 1 == this.start) {
+ return new HashRange(other.start, this.end);
+ }
+ throw new IllegalArgumentException(
+ "Ranges are not adjacent: " + this + " and " + other);
+ }
+
+ @Override
+ public int compareTo(HashRange o) {
+ int result = Integer.compare(start, o.start);
+ if (result == 0) {
+ result = Integer.compare(end, o.end);
+ }
+ return result;
+ }
+
+ public String toHexString() {
+ return String.format("%04x-%04x", start, end);
+ }
+
+ public static HashRange fromHexString(String hex) {
+ String[] parts = hex.split("-", 2);
+ if (parts.length != 2) {
+ throw new IllegalArgumentException("Invalid hash range format: " + hex);
+ }
+ return new HashRange(Integer.parseInt(parts[0], 16), Integer.parseInt(parts[1], 16));
+ }
+
+ @Override
+ public String toString() {
+ return "[" + String.format("%04x", start) + ", " + String.format("%04x", end) + "]";
+ }
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentInfo.java b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentInfo.java
new file mode 100644
index 00000000000..de6f6aae18e
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentInfo.java
@@ -0,0 +1,98 @@
+/*
+ * 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.common.scalable;
+
+import java.util.List;
+
+/**
+ * Describes a single segment in a scalable topic's DAG.
+ *
+ * <p>Each segment covers an inclusive hash range and has a unique monotonically increasing ID.
+ * Segments are linked by parent/child edges that form a DAG representing the split/merge history.
+ * Active segments are the leaves (no children); sealed segments are internal nodes.
+ *
+ * @param segmentId monotonically increasing, unique within the topic
+ * @param hashRange inclusive hash range [start, end]
+ * @param state ACTIVE or SEALED
+ * @param parentIds parent segment IDs in the DAG (empty for initial/root segments)
+ * @param childIds child segment IDs in the DAG (empty for active leaf segments)
+ * @param createdAtEpoch epoch when this segment was created
+ * @param sealedAtEpoch epoch when sealed (-1 if still active)
+ */
+public record SegmentInfo(
+ long segmentId,
+ HashRange hashRange,
+ SegmentState state,
+ List<Long> parentIds,
+ List<Long> childIds,
+ long createdAtEpoch,
+ long sealedAtEpoch
+) {
+ public SegmentInfo {
+ parentIds = parentIds != null ? List.copyOf(parentIds) : List.of();
+ childIds = childIds != null ? List.copyOf(childIds) : List.of();
+ }
+
+ /** Create a new active segment with no parents. */
+ public static SegmentInfo active(long segmentId, HashRange hashRange, long createdAtEpoch) {
+ return new SegmentInfo(segmentId, hashRange, SegmentState.ACTIVE,
+ List.of(), List.of(), createdAtEpoch, -1);
+ }
+
+ /** Create a new active segment with the given parent IDs. */
+ public static SegmentInfo active(long segmentId, HashRange hashRange,
+ List<Long> parentIds, long createdAtEpoch) {
+ return new SegmentInfo(segmentId, hashRange, SegmentState.ACTIVE,
+ parentIds, List.of(), createdAtEpoch, -1);
+ }
+
+ /** Return a sealed copy of this segment with the given child IDs. */
+ public SegmentInfo sealed(long sealedAtEpoch, List<Long> childIds) {
+ return new SegmentInfo(segmentId, hashRange, SegmentState.SEALED,
+ parentIds, childIds, createdAtEpoch, sealedAtEpoch);
+ }
+
+ /** Return a copy with different parent IDs. */
+ public SegmentInfo withParentIds(List<Long> parentIds) {
+ return new SegmentInfo(segmentId, hashRange, state,
+ parentIds, childIds, createdAtEpoch, sealedAtEpoch);
+ }
+
+ /** Return a copy with different child IDs. */
+ public SegmentInfo withChildIds(List<Long> childIds) {
+ return new SegmentInfo(segmentId, hashRange, state,
+ parentIds, childIds, createdAtEpoch, sealedAtEpoch);
+ }
+
+ public boolean isActive() {
+ return state == SegmentState.ACTIVE;
+ }
+
+ public boolean isSealed() {
+ return state == SegmentState.SEALED;
+ }
+
+ public boolean isRoot() {
+ return parentIds.isEmpty();
+ }
+
+ public boolean isLeaf() {
+ return childIds.isEmpty();
+ }
+}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentState.java
similarity index 53%
copy from pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
copy to pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentState.java
index 0f7ccfe9125..cfad8f84090 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentState.java
@@ -16,35 +16,16 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.common.naming;
+package org.apache.pulsar.common.scalable;
/**
- * Enumeration showing if a topic is persistent.
+ * State of a segment in a scalable topic.
+ *
+ * <p>Split and merge operations are atomic in the metadata store, so there are no
+ * intermediate transition states. A segment is either actively receiving writes
+ * or sealed (no longer receiving writes, but data still accessible until expiry).
*/
-public enum TopicDomain {
- persistent("persistent"), non_persistent("non-persistent");
-
- private String value;
-
- private TopicDomain(String value) {
- this.value = value;
- }
-
- public String value() {
- return this.value;
- }
-
- public static TopicDomain getEnum(String value) {
- for (TopicDomain e : values()) {
- if (e.value.equalsIgnoreCase(value)) {
- return e;
- }
- }
- throw new IllegalArgumentException("Invalid topic domain: '" + value + "'");
- }
-
- @Override
- public String toString() {
- return this.value;
- }
-}
\ No newline at end of file
+public enum SegmentState {
+ ACTIVE,
+ SEALED
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentTopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentTopicName.java
new file mode 100644
index 00000000000..971623da9b2
--- /dev/null
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/SegmentTopicName.java
@@ -0,0 +1,106 @@
+/*
+ * 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.common.scalable;
+
+import org.apache.pulsar.common.naming.TopicDomain;
+import org.apache.pulsar.common.naming.TopicName;
+
+/**
+ * Utility for constructing and parsing segment topic names.
+ *
+ * <p>Segment topic names follow the format:
+ * {@code segment://tenant/namespace/parent-topic-local-name/{hexStart}-{hexEnd}-{segmentId}}
+ *
+ * <p>For example:
+ * <ul>
+ * <li>{@code segment://tenant/ns/my-topic/0000-ffff-0} - single segment covering full range</li>
+ * <li>{@code segment://tenant/ns/my-topic/0000-7fff-1} - segment covering lower half</li>
+ * </ul>
+ *
+ * <p>TopicName natively parses segment:// URIs: {@code getLocalName()} returns the parent topic name,
+ * {@code getSegmentRange()} returns the hash range, and {@code getSegmentId()} returns the segment ID.
+ */
+public final class SegmentTopicName {
+
+ private SegmentTopicName() {
+ }
+
+ /**
+ * Construct a segment topic name from a parent topic and segment info.
+ *
+ * @param parentTopic the parent scalable topic (domain must be "topic")
+ * @param hashRange the hash range of the segment
+ * @param segmentId the segment ID
+ * @return the segment TopicName
+ */
+ public static TopicName fromParent(TopicName parentTopic, HashRange hashRange, long segmentId) {
+ if (parentTopic.getDomain() != TopicDomain.topic) {
+ throw new IllegalArgumentException(
+ "Parent topic must have domain 'topic', got: " + parentTopic.getDomain());
+ }
+ String segmentTopicName = TopicDomain.segment.value() + "://"
+ + parentTopic.getTenant() + "/"
+ + parentTopic.getNamespacePortion() + "/"
+ + parentTopic.getLocalName() + "/"
+ + formatDescriptor(hashRange, segmentId);
+ return TopicName.get(segmentTopicName);
+ }
+
+ /**
+ * Get the parent scalable topic name from a segment topic name.
+ *
+ * @param segmentTopic the segment topic name
+ * @return the parent TopicName with domain "topic"
+ */
+ public static TopicName getParentTopicName(TopicName segmentTopic) {
+ validateSegmentDomain(segmentTopic);
+ return TopicName.get(TopicDomain.topic.value(), segmentTopic.getTenant(),
+ segmentTopic.getNamespacePortion(), segmentTopic.getLocalName());
+ }
+
+ /**
+ * Extract the hash range from a segment topic name.
+ */
+ public static HashRange getHashRange(TopicName segmentTopic) {
+ validateSegmentDomain(segmentTopic);
+ return segmentTopic.getSegmentRange().orElseThrow();
+ }
+
+ /**
+ * Extract the segment ID from a segment topic name.
+ */
+ public static long getSegmentId(TopicName segmentTopic) {
+ validateSegmentDomain(segmentTopic);
+ return segmentTopic.getSegmentId();
+ }
+
+ /**
+ * Format a segment descriptor string: "{hexStart}-{hexEnd}-{segmentId}".
+ */
+ public static String formatDescriptor(HashRange hashRange, long segmentId) {
+ return String.format("%04x-%04x-%d", hashRange.start(), hashRange.end(), segmentId);
+ }
+
+ private static void validateSegmentDomain(TopicName topicName) {
+ if (topicName.getDomain() != TopicDomain.segment) {
+ throw new IllegalArgumentException(
+ "Expected segment domain, got: " + topicName.getDomain());
+ }
+ }
+}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/package-info.java
similarity index 53%
copy from pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
copy to pulsar-common/src/main/java/org/apache/pulsar/common/scalable/package-info.java
index 0f7ccfe9125..ecd6f13e99b 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/naming/TopicDomain.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/scalable/package-info.java
@@ -16,35 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.common.naming;
-
/**
- * Enumeration showing if a topic is persistent.
+ * Scalable topic data model: hash ranges, segment topic names,
+ * and supporting types shared between broker and client.
*/
-public enum TopicDomain {
- persistent("persistent"), non_persistent("non-persistent");
-
- private String value;
-
- private TopicDomain(String value) {
- this.value = value;
- }
-
- public String value() {
- return this.value;
- }
-
- public static TopicDomain getEnum(String value) {
- for (TopicDomain e : values()) {
- if (e.value.equalsIgnoreCase(value)) {
- return e;
- }
- }
- throw new IllegalArgumentException("Invalid topic domain: '" + value + "'");
- }
-
- @Override
- public String toString() {
- return this.value;
- }
-}
\ No newline at end of file
+package org.apache.pulsar.common.scalable;
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
index f19053da29a..412d3696b1e 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/naming/TopicNameTest.java
@@ -21,9 +21,11 @@ package org.apache.pulsar.common.naming;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.util.Codec;
import org.testng.annotations.Test;
@@ -298,6 +300,209 @@ public class TopicNameTest {
assertEquals(tp2.toString(), "persistent://tenant1/namespace1/tp1-partition-0-DLQ-partition-0");
}
+ @Test
+ public void testScalableTopicDomain() {
+ // topic:// domain - basic parsing
+ TopicName tn = TopicName.get("topic://tenant/namespace/my-topic");
+ assertEquals(tn.getDomain(), TopicDomain.topic);
+ assertEquals(tn.getTenant(), "tenant");
+ assertEquals(tn.getNamespacePortion(), "namespace");
+ assertEquals(tn.getLocalName(), "my-topic");
+ assertEquals(tn.toString(), "topic://tenant/namespace/my-topic");
+ assertTrue(tn.isScalable());
+ assertFalse(tn.isSegment());
+ assertFalse(tn.getSegmentRange().isPresent());
+
+ // topic:// equality and factory methods
+ assertEquals(TopicName.get("topic://tenant/namespace/my-topic"),
+ TopicName.get("topic", "tenant", "namespace", "my-topic"));
+
+ // topic:// namespace
+ assertEquals(TopicName.get("topic://t/ns/x").getNamespace(), "t/ns");
+
+ // topic:// isValid
+ assertTrue(TopicName.isValid("topic://tenant/namespace/my-topic"));
+ assertTrue(TopicName.isValid("segment://tenant/namespace/my-topic/0000-ffff-0"));
+
+ // topic:// blank local name rejected
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.get("topic://tenant/namespace/"));
+
+ // topic:// missing namespace rejected
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.get("topic://tenant"));
+
+ // Partitioned topic with topic:// domain
+ tn = TopicName.get("topic://tenant/namespace/my-topic-partition-3");
+ assertTrue(tn.isPartitioned());
+ assertEquals(tn.getPartitionIndex(), 3);
+ assertEquals(tn.getPartitionedTopicName(), "topic://tenant/namespace/my-topic");
+
+ // Non-partitioned with topic:// domain
+ tn = TopicName.get("topic://tenant/namespace/my-topic");
+ assertFalse(tn.isPartitioned());
+ assertEquals(tn.getPartitionIndex(), -1);
+ }
+
+ @Test
+ public void testSegmentDomain() {
+ // segment:// domain - full range
+ TopicName tn = TopicName.get("segment://tenant/namespace/my-topic/0000-ffff-0");
+ assertEquals(tn.getDomain(), TopicDomain.segment);
+ assertTrue(tn.isSegment());
+ assertEquals(tn.getTenant(), "tenant");
+ assertEquals(tn.getNamespacePortion(), "namespace");
+ assertEquals(tn.getLocalName(), "my-topic");
+ assertEquals(tn.getSegmentRange(), java.util.Optional.of(HashRange.of(0x0000, 0xFFFF)));
+ assertEquals(tn.getSegmentId(), 0);
+ assertEquals(tn.toString(), "segment://tenant/namespace/my-topic/0000-ffff-0");
+
+ // segment:// with lower half-range
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-7fff-1");
+ assertEquals(tn.getLocalName(), "my-topic");
+ assertEquals(tn.getSegmentRange(), java.util.Optional.of(HashRange.of(0x0000, 0x7FFF)));
+ assertEquals(tn.getSegmentId(), 1);
+
+ // segment:// with upper half-range
+ tn = TopicName.get("segment://tenant/namespace/my-topic/8000-ffff-2");
+ assertEquals(tn.getLocalName(), "my-topic");
+ assertEquals(tn.getSegmentRange(), java.util.Optional.of(HashRange.of(0x8000, 0xFFFF)));
+ assertEquals(tn.getSegmentId(), 2);
+
+ // segment:// missing descriptor rejected
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.get("segment://tenant/namespace/my-topic"));
+
+ // segment:// blank local name rejected
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.get("segment://tenant/namespace/"));
+
+ // segment:// invalid descriptor rejected
+ assertThrows(IllegalArgumentException.class,
+ () -> TopicName.get("segment://tenant/namespace/my-topic/bad-descriptor"));
+ }
+
+ @Test
+ public void testScalableTopicPersistence() {
+ // topic:// and segment:// are persistent (backed by managed ledgers)
+ assertTrue(TopicName.get("topic://tenant/namespace/t").isPersistent());
+ assertTrue(TopicName.get("segment://tenant/namespace/t/0000-ffff-0").isPersistent());
+
+ // persistent:// and non-persistent:// are not scalable
+ assertFalse(TopicName.get("persistent://tenant/namespace/t").isScalable());
+ assertFalse(TopicName.get("non-persistent://tenant/namespace/t").isScalable());
+
+ // segment:// is not scalable (it's a segment of a scalable topic)
+ assertFalse(TopicName.get("segment://tenant/namespace/t/0000-ffff-0").isScalable());
+ }
+
+ @Test
+ public void testScalableTopicPersistenceNamingEncoding() {
+ // topic:// persistence encoding (no domain component, no managed ledger)
+ TopicName tn = TopicName.get("topic://tenant/namespace/my-topic");
+ assertEquals(tn.getPersistenceNamingEncoding(), "tenant/namespace/my-topic");
+
+ // segment:// persistence encoding includes descriptor
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-ffff-0");
+ assertEquals(tn.getPersistenceNamingEncoding(), "tenant/namespace/segment/my-topic/0000-ffff-0");
+
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-7fff-1");
+ assertEquals(tn.getPersistenceNamingEncoding(), "tenant/namespace/segment/my-topic/0000-7fff-1");
+
+ tn = TopicName.get("segment://tenant/namespace/my-topic/8000-ffff-2");
+ assertEquals(tn.getPersistenceNamingEncoding(), "tenant/namespace/segment/my-topic/8000-ffff-2");
+
+ // segment:// with special characters in local name
+ tn = TopicName.get("segment://tenant/namespace/a:b/0000-ffff-0");
+ assertEquals(tn.getPersistenceNamingEncoding(), "tenant/namespace/segment/a%3Ab/0000-ffff-0");
+ }
+
+ @Test
+ public void testScalableTopicFromPersistenceNamingEncoding() {
+ // segment:// round-trip through persistence encoding
+ String segmentTopic = "segment://tenant/namespace/my-topic/0000-ffff-0";
+ TopicName tn = TopicName.get(segmentTopic);
+ String mlName = tn.getPersistenceNamingEncoding();
+ assertEquals(mlName, "tenant/namespace/segment/my-topic/0000-ffff-0");
+ assertEquals(TopicName.fromPersistenceNamingEncoding(mlName), segmentTopic);
+
+ // segment:// with split range
+ segmentTopic = "segment://tenant/namespace/my-topic/0000-7fff-1";
+ tn = TopicName.get(segmentTopic);
+ mlName = tn.getPersistenceNamingEncoding();
+ assertEquals(TopicName.fromPersistenceNamingEncoding(mlName), segmentTopic);
+
+ // segment:// with special characters in local name
+ segmentTopic = "segment://tenant/namespace/a:b/0000-ffff-0";
+ tn = TopicName.get(segmentTopic);
+ mlName = tn.getPersistenceNamingEncoding();
+ assertEquals(TopicName.fromPersistenceNamingEncoding(mlName), segmentTopic);
+
+ // topic:// persistence encoding is 3-part (no domain, no managed ledger)
+ // so fromPersistenceNamingEncoding is not applicable for topic://
+ }
+
+ @Test
+ public void testScalableTopicSchemaName() {
+ // topic:// schema name (uses tenant/namespacePortion/encodedLocalName)
+ TopicName tn = TopicName.get("topic://tenant/namespace/my-topic");
+ assertEquals(tn.getSchemaName(), "tenant/namespace/my-topic");
+
+ // segment:// schema name — should use the parent topic name, not the descriptor
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-ffff-0");
+ assertEquals(tn.getSchemaName(), "tenant/namespace/my-topic");
+
+ // Different segment ranges of the same topic share the same schema name
+ TopicName seg1 = TopicName.get("segment://tenant/namespace/my-topic/0000-7fff-1");
+ TopicName seg2 = TopicName.get("segment://tenant/namespace/my-topic/8000-ffff-2");
+ assertEquals(seg1.getSchemaName(), seg2.getSchemaName());
+
+ // topic:// with special characters
+ tn = TopicName.get("topic://tenant/namespace/a:b");
+ assertEquals(tn.getSchemaName(), "tenant/namespace/a%3Ab");
+ }
+
+ @Test
+ public void testScalableTopicLookupName() {
+ // topic:// lookup name
+ TopicName tn = TopicName.get("topic://tenant/namespace/my-topic");
+ assertEquals(tn.getLookupName(), "topic/tenant/namespace/my-topic");
+
+ // segment:// lookup name — includes the parent topic name (not descriptor)
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-ffff-0");
+ assertEquals(tn.getLookupName(), "segment/tenant/namespace/my-topic");
+ }
+
+ @Test
+ public void testScalableTopicRestPath() {
+ // topic:// rest path with domain
+ TopicName tn = TopicName.get("topic://tenant/namespace/my-topic");
+ assertEquals(tn.getRestPath(), "topic/tenant/namespace/my-topic");
+ assertEquals(tn.getRestPath(false), "tenant/namespace/my-topic");
+
+ // segment:// rest path with domain
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-ffff-0");
+ assertEquals(tn.getRestPath(), "segment/tenant/namespace/my-topic");
+ assertEquals(tn.getRestPath(false), "tenant/namespace/my-topic");
+ }
+
+ @Test
+ public void testSegmentDescriptor() {
+ // segment:// getSegmentDescriptor
+ TopicName tn = TopicName.get("segment://tenant/namespace/my-topic/0000-ffff-0");
+ assertEquals(tn.getSegmentDescriptor(), "0000-ffff-0");
+
+ tn = TopicName.get("segment://tenant/namespace/my-topic/0000-7fff-1");
+ assertEquals(tn.getSegmentDescriptor(), "0000-7fff-1");
+
+ tn = TopicName.get("segment://tenant/namespace/my-topic/8000-ffff-2");
+ assertEquals(tn.getSegmentDescriptor(), "8000-ffff-2");
+
+ // Non-segment domains return null
+ assertNull(TopicName.get("persistent://tenant/namespace/my-topic").getSegmentDescriptor());
+ assertNull(TopicName.get("topic://tenant/namespace/my-topic").getSegmentDescriptor());
+ }
+
@Test
public void testToFullTopicName() {
// There is no constraint for local topic name
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/scalable/HashRangeTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/scalable/HashRangeTest.java
new file mode 100644
index 00000000000..d1bfc3c066b
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/scalable/HashRangeTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.common.scalable;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import org.testng.annotations.Test;
+
+public class HashRangeTest {
+
+ @Test
+ public void testFullRange() {
+ HashRange full = HashRange.full();
+ assertEquals(full.start(), 0x0000);
+ assertEquals(full.end(), 0xFFFF);
+ assertEquals(full.size(), 65536);
+ }
+
+ @Test
+ public void testContainsHash() {
+ HashRange range = HashRange.of(0x1000, 0x2000);
+ assertTrue(range.contains(0x1000));
+ assertTrue(range.contains(0x1500));
+ assertTrue(range.contains(0x2000));
+ assertFalse(range.contains(0x0FFF));
+ assertFalse(range.contains(0x2001));
+ }
+
+ @Test
+ public void testContainsRange() {
+ HashRange outer = HashRange.of(0x1000, 0x3000);
+ HashRange inner = HashRange.of(0x1500, 0x2500);
+ assertTrue(outer.contains(inner));
+ assertFalse(inner.contains(outer));
+ }
+
+ @Test
+ public void testSplit() {
+ HashRange full = HashRange.full();
+ HashRange[] halves = full.split();
+ assertEquals(halves[0], HashRange.of(0x0000, 0x7FFF));
+ assertEquals(halves[1], HashRange.of(0x8000, 0xFFFF));
+ assertEquals(halves[0].size() + halves[1].size(), full.size());
+ }
+
+ @Test
+ public void testSplitSmallRange() {
+ HashRange range = HashRange.of(100, 101);
+ HashRange[] parts = range.split();
+ assertEquals(parts[0], HashRange.of(100, 100));
+ assertEquals(parts[1], HashRange.of(101, 101));
+ }
+
+ @Test
+ public void testCannotSplitSingleValue() {
+ HashRange range = HashRange.of(42, 42);
+ assertThrows(IllegalStateException.class, range::split);
+ }
+
+ @Test
+ public void testMergeAdjacent() {
+ HashRange left = HashRange.of(0x0000, 0x7FFF);
+ HashRange right = HashRange.of(0x8000, 0xFFFF);
+ assertEquals(left.merge(right), HashRange.full());
+ assertEquals(right.merge(left), HashRange.full());
+ }
+
+ @Test
+ public void testMergeNonAdjacent() {
+ HashRange a = HashRange.of(0x0000, 0x3FFF);
+ HashRange b = HashRange.of(0x8000, 0xFFFF);
+ assertThrows(IllegalArgumentException.class, () -> a.merge(b));
+ }
+
+ @Test
+ public void testIsAdjacentTo() {
+ HashRange a = HashRange.of(0x0000, 0x7FFF);
+ HashRange b = HashRange.of(0x8000, 0xFFFF);
+ assertTrue(a.isAdjacentTo(b));
+ assertTrue(b.isAdjacentTo(a));
+ assertFalse(a.isAdjacentTo(HashRange.of(0x9000, 0xFFFF)));
+ }
+
+ @Test
+ public void testHexRoundTrip() {
+ HashRange range = HashRange.of(0x0A0B, 0xF0F0);
+ String hex = range.toHexString();
+ assertEquals(hex, "0a0b-f0f0");
+ assertEquals(HashRange.fromHexString(hex), range);
+ }
+
+ @Test
+ public void testComparable() {
+ HashRange a = HashRange.of(0x1000, 0x2000);
+ HashRange b = HashRange.of(0x3000, 0x4000);
+ assertTrue(a.compareTo(b) < 0);
+ assertTrue(b.compareTo(a) > 0);
+ assertEquals(a.compareTo(a), 0);
+ }
+
+ @Test
+ public void testInvalidRange() {
+ assertThrows(IllegalArgumentException.class, () -> HashRange.of(-1, 100));
+ assertThrows(IllegalArgumentException.class, () -> HashRange.of(0, 0x10000));
+ assertThrows(IllegalArgumentException.class, () -> HashRange.of(100, 50));
+ }
+
+ @Test
+ public void testToString() {
+ HashRange range = HashRange.of(0x0000, 0xFFFF);
+ assertEquals(range.toString(), "[0000, ffff]");
+ }
+}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/scalable/SegmentTopicNameTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/scalable/SegmentTopicNameTest.java
new file mode 100644
index 00000000000..89d378f540a
--- /dev/null
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/scalable/SegmentTopicNameTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.common.scalable;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import org.apache.pulsar.common.naming.TopicDomain;
+import org.apache.pulsar.common.naming.TopicName;
+import org.testng.annotations.Test;
+
+public class SegmentTopicNameTest {
+
+ @Test
+ public void testFromParent() {
+ TopicName parent = TopicName.get("topic://tenant/ns/my-topic");
+ HashRange range = HashRange.of(0x0000, 0x7FFF);
+ TopicName segment = SegmentTopicName.fromParent(parent, range, 1);
+
+ assertEquals(segment.toString(), "segment://tenant/ns/my-topic/0000-7fff-1");
+ assertEquals(segment.getDomain(), TopicDomain.segment);
+ assertEquals(segment.getTenant(), "tenant");
+ assertEquals(segment.getNamespacePortion(), "ns");
+ assertEquals(segment.getLocalName(), "my-topic");
+ assertEquals(segment.getSegmentRange(), java.util.Optional.of(range));
+ assertEquals(segment.getSegmentId(), 1);
+ }
+
+ @Test
+ public void testFromParentFullRange() {
+ TopicName parent = TopicName.get("topic://public/default/test");
+ HashRange range = HashRange.full();
+ TopicName segment = SegmentTopicName.fromParent(parent, range, 0);
+
+ assertEquals(segment.toString(), "segment://public/default/test/0000-ffff-0");
+ }
+
+ @Test
+ public void testGetParentTopicName() {
+ TopicName segment = TopicName.get("segment://tenant/ns/my-topic/0000-7fff-1");
+ TopicName parent = SegmentTopicName.getParentTopicName(segment);
+
+ assertEquals(parent.toString(), "topic://tenant/ns/my-topic");
+ assertEquals(parent.getDomain(), TopicDomain.topic);
+ }
+
+ @Test
+ public void testGetHashRange() {
+ TopicName segment = TopicName.get("segment://tenant/ns/my-topic/0000-7fff-1");
+ HashRange range = SegmentTopicName.getHashRange(segment);
+
+ assertEquals(range, HashRange.of(0x0000, 0x7FFF));
+ }
+
+ @Test
+ public void testGetSegmentId() {
+ TopicName segment = TopicName.get("segment://tenant/ns/my-topic/0000-7fff-42");
+ long segmentId = SegmentTopicName.getSegmentId(segment);
+
+ assertEquals(segmentId, 42);
+ }
+
+ @Test
+ public void testRoundTrip() {
+ TopicName parent = TopicName.get("topic://my-tenant/my-ns/orders");
+ HashRange range = HashRange.of(0x4000, 0x7FFF);
+ long segmentId = 5;
+
+ TopicName segment = SegmentTopicName.fromParent(parent, range, segmentId);
+ TopicName recoveredParent = SegmentTopicName.getParentTopicName(segment);
+ HashRange recoveredRange = SegmentTopicName.getHashRange(segment);
+ long recoveredId = SegmentTopicName.getSegmentId(segment);
+
+ assertEquals(recoveredParent.toString(), parent.toString());
+ assertEquals(recoveredRange, range);
+ assertEquals(recoveredId, segmentId);
+ }
+
+ @Test
+ public void testFromParentRejectsPersistentDomain() {
+ TopicName persistent = TopicName.get("persistent://tenant/ns/my-topic");
+ assertThrows(IllegalArgumentException.class,
+ () -> SegmentTopicName.fromParent(persistent, HashRange.full(), 0));
+ }
+
+ @Test
+ public void testGetParentRejectsPersistentDomain() {
+ TopicName persistent = TopicName.get("persistent://tenant/ns/my-topic");
+ assertThrows(IllegalArgumentException.class,
+ () -> SegmentTopicName.getParentTopicName(persistent));
+ }
+
+ @Test
+ public void testTopicDomainParsing() {
+ TopicName topic = TopicName.get("topic://t/ns/foo");
+ assertEquals(topic.getDomain(), TopicDomain.topic);
+ assertEquals(topic.getTenant(), "t");
+ assertEquals(topic.getNamespacePortion(), "ns");
+ assertEquals(topic.getLocalName(), "foo");
+ }
+
+ @Test
+ public void testSegmentDomainParsing() {
+ TopicName segment = TopicName.get("segment://t/ns/foo/0000-ffff-0");
+ assertEquals(segment.getDomain(), TopicDomain.segment);
+ assertEquals(segment.getTenant(), "t");
+ assertEquals(segment.getNamespacePortion(), "ns");
+ assertEquals(segment.getLocalName(), "foo");
+ assertEquals(segment.getSegmentRange(), java.util.Optional.of(HashRange.full()));
+ assertEquals(segment.getSegmentId(), 0);
+ assertTrue(segment.isSegment());
+ }
+
+ @Test
+ public void testFormatDescriptor() {
+ String desc = SegmentTopicName.formatDescriptor(HashRange.of(0x0000, 0x7FFF), 3);
+ assertEquals(desc, "0000-7fff-3");
+ }
+}