[feat][broker] PIP-468: Filter scalable topics by property via secondary index (#25632)

Author: merlimat

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicResources.java

@@ -18,7 +18,10 @@
  */
 package org.apache.pulsar.broker.resources;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
@@ -28,6 +31,7 @@
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.Codec;
 import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.metadata.api.MetadataCache;
 import org.apache.pulsar.metadata.api.MetadataStore;
 import org.apache.pulsar.metadata.api.MetadataStoreException;
@@ -54,6 +58,16 @@ public class ScalableTopicResources extends BaseResources<ScalableTopicMetadata>
     private static final String SUBSCRIPTIONS_SEGMENT = "subscriptions";
     private static final String CONSUMERS_SEGMENT = "consumers";
 
+    /**
+     * Use the topic's {@code properties} map verbatim as the secondary-index entries.
+     * Each property {@code k -> v} is registered as the index named {@code k} with
+     * secondary key {@code v}; querying by that key/value pair via
+     * {@link MetadataStore#findByIndex} returns the record. Index names live in a
+     * per-record-type namespace, so there's no need to disambiguate them with a prefix.
+     */
+    private static final Function<ScalableTopicMetadata, Map<String, String>> PROPERTY_INDEX_EXTRACTOR =
+            metadata -> metadata.getProperties() != null ? metadata.getProperties() : Map.of();
+
     private final MetadataCache<SubscriptionMetadata> subscriptionCache;
     private final MetadataCache<ConsumerRegistration> consumerRegistrationCache;
 
@@ -64,7 +78,7 @@ public ScalableTopicResources(MetadataStore store, int operationTimeoutSec) {
     }
 
     public CompletableFuture<Void> createScalableTopicAsync(TopicName tn, ScalableTopicMetadata metadata) {
-        return createAsync(topicPath(tn), metadata);
+        return getCache().create(topicPath(tn), metadata, PROPERTY_INDEX_EXTRACTOR);
     }
 
     public CompletableFuture<Optional<ScalableTopicMetadata>> getScalableTopicMetadataAsync(TopicName tn) {
@@ -82,7 +96,10 @@ public CompletableFuture<Optional<ScalableTopicMetadata>> getScalableTopicMetada
     public CompletableFuture<Void> updateScalableTopicAsync(TopicName tn,
                                                              Function<ScalableTopicMetadata,
                                                                      ScalableTopicMetadata> updateFunction) {
-        return setAsync(topicPath(tn), updateFunction);
+        // Refresh property indexes on every update — the modify function may add or remove
+        // properties and the underlying store needs to see the post-modification view.
+        return getCache().readModifyUpdate(topicPath(tn), updateFunction, PROPERTY_INDEX_EXTRACTOR)
+                .thenApply(__ -> null);
     }
 
     public CompletableFuture<Void> deleteScalableTopicAsync(TopicName tn) {
@@ -100,6 +117,41 @@ public CompletableFuture<List<String>> listScalableTopicsAsync(NamespaceName ns)
                         .collect(Collectors.toList()));
     }
 
+    /**
+     * List scalable topics in a namespace whose {@code properties} map contains the given
+     * key/value pair. On stores with native secondary index support (Oxia) this is served
+     * by the index registered at create/update time; otherwise it falls back to a children
+     * scan + per-record property check.
+     *
+     * @param ns            the namespace to scope the query to
+     * @param propertyKey   property name to filter on
+     * @param propertyValue exact property value to match
+     * @return fully qualified scalable topic names matching the property
+     */
+    public CompletableFuture<List<String>> findScalableTopicsByPropertyAsync(
+            NamespaceName ns, String propertyKey, String propertyValue) {
+        String scanPathPrefix = joinPath(SCALABLE_TOPIC_PATH, ns.toString());
+        ObjectMapper mapper = ObjectMapperFactory.getMapper().getObjectMapper();
+        return getStore().findByIndex(scanPathPrefix, propertyKey, propertyValue, result -> {
+                    // Fallback path (no native index): re-check the property on the loaded record.
+                    try {
+                        ScalableTopicMetadata md =
+                                mapper.readValue(result.getValue(), ScalableTopicMetadata.class);
+                        return md.getProperties() != null
+                                && propertyValue.equals(md.getProperties().get(propertyKey));
+                    } catch (IOException e) {
+                        return false;
+                    }
+                })
+                .thenApply(results -> results.stream()
+                        .map(r -> {
+                            String path = r.getStat().getPath();
+                            String encoded = path.substring(path.lastIndexOf('/') + 1);
+                            return TopicName.get("topic", ns, Codec.decode(encoded)).toString();
+                        })
+                        .collect(Collectors.toList()));
+    }
+
     // --- Subscriptions ---
 
     /**

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java

@@ -93,10 +93,19 @@ public void getList(
             @ApiParam(value = "Specify the tenant", required = true)
             @PathParam("tenant") String tenant,
             @ApiParam(value = "Specify the namespace", required = true)
-            @PathParam("namespace") String namespace) {
+            @PathParam("namespace") String namespace,
+            @ApiParam(value = "Filter by topic property name (must be paired with propertyValue)")
+            @QueryParam("propertyKey") String propertyKey,
+            @ApiParam(value = "Filter by topic property value (must be paired with propertyKey)")
+            @QueryParam("propertyValue") String propertyValue) {
         validateNamespaceName(tenant, namespace);
+        boolean filterByProperty = propertyKey != null && !propertyKey.isEmpty()
+                && propertyValue != null;
         validateNamespaceOperationAsync(namespaceName, NamespaceOperation.GET_TOPICS)
-                .thenCompose(__ -> resources().listScalableTopicsAsync(namespaceName))
+                .thenCompose(__ -> filterByProperty
+                        ? resources().findScalableTopicsByPropertyAsync(
+                                namespaceName, propertyKey, propertyValue)
+                        : resources().listScalableTopicsAsync(namespaceName))
                 .thenAccept(asyncResponse::resume)
                 .exceptionally(ex -> {
                     log.error().attr("clientAppId", clientAppId()).attr("namespace", namespaceName)

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/ScalableTopicsListByPropertyTest.java

@@ -0,0 +1,83 @@
+/*
+ * 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.broker.admin;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.testng.annotations.Test;
+
+/**
+ * End-to-end coverage for the property-filtered list endpoint
+ * ({@code GET /admin/v2/scalable/{tenant}/{namespace}?propertyKey&propertyValue}).
+ * Drives the full HTTP path through the {@link org.apache.pulsar.client.admin.PulsarAdmin}
+ * client against a real shared broker, verifying that topics created with
+ * properties through the admin API are queryable via the secondary index.
+ */
+public class ScalableTopicsListByPropertyTest extends SharedPulsarBaseTest {
+
+    private String namespace() {
+        return getNamespace();
+    }
+
+    private String topicName(String suffix) {
+        return "topic://" + namespace() + "/" + suffix + "-" + UUID.randomUUID().toString().substring(0, 8);
+    }
+
+    @Test
+    public void listScalableTopicsFilteredByProperty() throws Exception {
+        String aliceTopic = topicName("alice");
+        String bobTopic = topicName("bob");
+        String carolTopic = topicName("carol");
+
+        // Each topic gets a different owner; alice and bob share a team. The filter
+        // should be able to surface either subset on demand.
+        admin.scalableTopics().createScalableTopic(aliceTopic, 1,
+                Map.of("owner", "alice", "team", "platform"));
+        admin.scalableTopics().createScalableTopic(bobTopic, 1,
+                Map.of("owner", "bob", "team", "platform"));
+        admin.scalableTopics().createScalableTopic(carolTopic, 1,
+                Map.of("owner", "carol", "team", "data"));
+
+        // Filter by owner=alice — single match.
+        List<String> alice = admin.scalableTopics()
+                .listScalableTopicsByProperty(namespace(), "owner", "alice");
+        assertEquals(alice, List.of(aliceTopic));
+
+        // Filter by team=platform — alice + bob.
+        Set<String> platform = new HashSet<>(admin.scalableTopics()
+                .listScalableTopicsByProperty(namespace(), "team", "platform"));
+        assertEquals(platform, Set.of(aliceTopic, bobTopic));
+
+        // Unmatched value — empty result.
+        assertTrue(admin.scalableTopics()
+                .listScalableTopicsByProperty(namespace(), "owner", "nonexistent")
+                .isEmpty());
+
+        // Sanity-check: the un-filtered listing still returns every topic in the namespace.
+        Set<String> all = new HashSet<>(admin.scalableTopics().listScalableTopics(namespace()));
+        assertTrue(all.containsAll(Set.of(aliceTopic, bobTopic, carolTopic)),
+                "expected all three created topics to appear in the unfiltered list, got " + all);
+    }
+}

pulsar-broker/src/test/java/org/apache/pulsar/broker/resources/ScalableTopicPropertyIndexTest.java

@@ -0,0 +1,174 @@
+/*
+ * 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.broker.resources;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.metadata.impl.LocalMemoryMetadataStore;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Coverage for the {@code findScalableTopicsByPropertyAsync} entry point on
+ * {@link ScalableTopicResources}: verifies that topic properties registered at
+ * create/update time are queryable via the secondary-index API, that updates
+ * refresh the index, and that the filter is correctly scoped to a namespace.
+ *
+ * <p>Uses {@link LocalMemoryMetadataStore} which does not implement native
+ * secondary indexes, so this exercises the fallback scan + per-record property
+ * predicate path. The Oxia-native path (where the index is consulted directly)
+ * is covered by the metadata-store-level secondary index tests.
+ */
+public class ScalableTopicPropertyIndexTest {
+
+    private MetadataStoreExtended store;
+    private ScalableTopicResources resources;
+
+    @BeforeMethod
+    public void setUp() throws Exception {
+        store = new LocalMemoryMetadataStore("memory:local",
+                MetadataStoreConfig.builder().build());
+        resources = new ScalableTopicResources(store, 30);
+    }
+
+    @AfterMethod(alwaysRun = true)
+    public void tearDown() throws Exception {
+        if (store != null) {
+            store.close();
+        }
+    }
+
+    private TopicName topicIn(NamespaceName ns, String localName) {
+        return TopicName.get("topic://" + ns + "/" + localName);
+    }
+
+    private ScalableTopicMetadata metaWithProps(Map<String, String> props) {
+        return ScalableTopicMetadata.builder()
+                .epoch(0)
+                .nextSegmentId(1)
+                .properties(new HashMap<>(props))
+                .build();
+    }
+
+    @Test
+    public void findsTopicByExactPropertyValue() throws Exception {
+        NamespaceName ns = NamespaceName.get("tenant/ns-a");
+
+        resources.createScalableTopicAsync(topicIn(ns, "t-alice"),
+                metaWithProps(Map.of("owner", "alice", "team", "platform"))).get();
+        resources.createScalableTopicAsync(topicIn(ns, "t-bob"),
+                metaWithProps(Map.of("owner", "bob", "team", "platform"))).get();
+        resources.createScalableTopicAsync(topicIn(ns, "t-carol"),
+                metaWithProps(Map.of("owner", "carol", "team", "data"))).get();
+
+        // Filter by owner=alice — only t-alice matches.
+        List<String> aliceOwned = resources
+                .findScalableTopicsByPropertyAsync(ns, "owner", "alice").get();
+        assertEquals(aliceOwned, List.of("topic://tenant/ns-a/t-alice"));
+
+        // Filter by team=platform — both alice and bob.
+        Set<String> platform = new HashSet<>(resources
+                .findScalableTopicsByPropertyAsync(ns, "team", "platform").get());
+        assertEquals(platform, Set.of(
+                "topic://tenant/ns-a/t-alice",
+                "topic://tenant/ns-a/t-bob"));
+    }
+
+    @Test
+    public void findIsScopedToNamespace() throws Exception {
+        NamespaceName nsA = NamespaceName.get("tenant/ns-a");
+        NamespaceName nsB = NamespaceName.get("tenant/ns-b");
+
+        // Same property in two namespaces — find must return only the one we asked for.
+        resources.createScalableTopicAsync(topicIn(nsA, "t1"),
+                metaWithProps(Map.of("owner", "alice"))).get();
+        resources.createScalableTopicAsync(topicIn(nsB, "t2"),
+                metaWithProps(Map.of("owner", "alice"))).get();
+
+        List<String> inNsA = resources
+                .findScalableTopicsByPropertyAsync(nsA, "owner", "alice").get();
+        assertEquals(inNsA, List.of("topic://tenant/ns-a/t1"));
+
+        List<String> inNsB = resources
+                .findScalableTopicsByPropertyAsync(nsB, "owner", "alice").get();
+        assertEquals(inNsB, List.of("topic://tenant/ns-b/t2"));
+    }
+
+    @Test
+    public void noMatchReturnsEmptyList() throws Exception {
+        NamespaceName ns = NamespaceName.get("tenant/ns-empty");
+        resources.createScalableTopicAsync(topicIn(ns, "t1"),
+                metaWithProps(Map.of("owner", "alice"))).get();
+
+        // Wrong value
+        assertTrue(resources.findScalableTopicsByPropertyAsync(ns, "owner", "bob")
+                .get().isEmpty());
+
+        // Wrong key
+        assertTrue(resources.findScalableTopicsByPropertyAsync(ns, "team", "alice")
+                .get().isEmpty());
+    }
+
+    @Test
+    public void updateRefreshesIndex() throws Exception {
+        NamespaceName ns = NamespaceName.get("tenant/ns-update");
+        TopicName tn = topicIn(ns, "t-mutating");
+
+        resources.createScalableTopicAsync(tn,
+                metaWithProps(Map.of("owner", "alice"))).get();
+        assertEquals(resources.findScalableTopicsByPropertyAsync(ns, "owner", "alice").get(),
+                List.of(tn.toString()));
+
+        // Reassign owner via update — the new owner must be queryable, and the
+        // old owner's entry must no longer match this topic.
+        resources.updateScalableTopicAsync(tn, current -> {
+            current.getProperties().put("owner", "bob");
+            return current;
+        }).get();
+
+        assertEquals(resources.findScalableTopicsByPropertyAsync(ns, "owner", "bob").get(),
+                List.of(tn.toString()));
+        assertTrue(resources.findScalableTopicsByPropertyAsync(ns, "owner", "alice")
+                .get().isEmpty());
+    }
+
+    @Test
+    public void topicWithoutPropertiesIsNotMatched() throws Exception {
+        NamespaceName ns = NamespaceName.get("tenant/ns-noprops");
+        resources.createScalableTopicAsync(topicIn(ns, "t-anon"),
+                metaWithProps(Map.of())).get();
+        resources.createScalableTopicAsync(topicIn(ns, "t-tagged"),
+                metaWithProps(Map.of("owner", "alice"))).get();
+
+        // Filtering by any property must skip the un-tagged record.
+        List<String> matches = resources
+                .findScalableTopicsByPropertyAsync(ns, "owner", "alice").get();
+        assertEquals(matches, List.of("topic://tenant/ns-noprops/t-tagged"));
+    }
+}