[feat] Pulsar Functions status-summary support across admin API, CLI, and worker runtime.

conf/functions_worker.yml

@@ -129,6 +129,8 @@ clusterCoordinationTopicName: coordinate
 useCompactedMetadataTopic: false
 # Number of threads to use for HTTP requests processing. Default is set to 8
 numHttpServerThreads: 8
+# Maximum parallelism used by function status-summary batch queries. Default is 4.
+functionsStatusSummaryMaxParallelism: 4
 
 # function assignment and scheduler
 

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Functions.java

@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.client.admin;
 
+import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -32,6 +34,8 @@
 import org.apache.pulsar.common.policies.data.FunctionInstanceStatsData;
 import org.apache.pulsar.common.policies.data.FunctionStats;
 import org.apache.pulsar.common.policies.data.FunctionStatus;
+import org.apache.pulsar.common.policies.data.FunctionStatusPage;
+import org.apache.pulsar.common.policies.data.FunctionStatusSummary;
 
 /**
  * Admin interface for function management.
@@ -69,6 +73,148 @@ public interface Functions {
      */
     CompletableFuture<List<String>> getFunctionsAsync(String tenant, String namespace);
 
+    /**
+     * Get a batch status summary for all functions in a namespace.
+     * <p/>
+     * Returns a lightweight summary (name, state, instance counts) for every function
+     * under the given tenant/namespace in a single API call. This is not equivalent to
+     * calling {@link #getFunctionStatus} for each function - it returns only aggregated
+     * counts, not per-instance details.
+     * <p/>
+     * Individual function failures are isolated: a function whose status cannot be
+     * retrieved will appear with {@code state = UNKNOWN} and a non-null {@code error}.
+     *
+     * @param tenant
+     *            Tenant name
+     * @param namespace
+     *            Namespace name
+     *
+     * @return list of status summaries, one per function
+     *
+     * @throws NotAuthorizedException
+     *             Don't have admin permission
+     * @throws PulsarAdminException
+     *             Unexpected error
+     */
+    default FunctionStatusPage getFunctionsWithStatus(String tenant, String namespace)
+            throws PulsarAdminException {
+        return getFunctionsWithStatus(tenant, namespace, null, null);
+    }
+
+    /**
+     * Get a batch status summary for all functions in a namespace asynchronously.
+     * <p/>
+     * Async version of {@link #getFunctionsWithStatus(String, String)}.
+     *
+     * @param tenant
+     *            Tenant name
+     * @param namespace
+     *            Namespace name
+     *
+     * @return a future that completes with the list of status summaries
+     */
+    default CompletableFuture<FunctionStatusPage> getFunctionsWithStatusAsync(String tenant,
+                                                                               String namespace) {
+        return getFunctionsWithStatusAsync(tenant, namespace, null, null);
+    }
+
+    /**
+     * Get a paginated batch status summary for functions in a namespace.
+     * <p/>
+     * The {@code startAfter} is an exclusive cursor based on function name
+     * in lexicographical order.
+     *
+     * @param tenant
+     *            Tenant name
+     * @param namespace
+     *            Namespace name
+     * @param limit
+     *            Maximum number of functions to return; must be greater than 0 when provided
+     * @param startAfter
+     *            Exclusive continuation token from previous page; null means from beginning
+     * @return list of status summaries for the requested page
+     * @throws PulsarAdminException
+     *             Unexpected error
+     */
+    default FunctionStatusPage getFunctionsWithStatus(
+            String tenant, String namespace, Integer limit, String startAfter)
+            throws PulsarAdminException {
+        if (limit != null && limit <= 0) {
+            throw new IllegalArgumentException("limit must be greater than 0");
+        }
+
+        List<String> functionNames = getFunctions(tenant, namespace);
+        List<String> pagedNames = new ArrayList<>(functionNames);
+        pagedNames.sort(String::compareTo);
+
+        int startIndex = 0;
+        if (startAfter != null && !startAfter.isEmpty()) {
+            while (startIndex < pagedNames.size() && pagedNames.get(startIndex).compareTo(startAfter) <= 0) {
+                startIndex++;
+            }
+        }
+
+        int endIndex = limit == null ? pagedNames.size() : Math.min(pagedNames.size(), startIndex + limit);
+        List<FunctionStatusSummary> summaries = new ArrayList<>(Math.max(0, endIndex - startIndex));
+        for (int index = startIndex; index < endIndex; index++) {
+            String functionName = pagedNames.get(index);
+            try {
+                FunctionStatus status = getFunctionStatus(tenant, namespace, functionName);
+                FunctionStatusSummary.SummaryState state = status.getNumInstances() <= 0
+                        ? FunctionStatusSummary.SummaryState.UNKNOWN
+                        : status.getNumRunning() == status.getNumInstances()
+                                ? FunctionStatusSummary.SummaryState.RUNNING
+                                : status.getNumRunning() == 0
+                                        ? FunctionStatusSummary.SummaryState.STOPPED
+                                        : FunctionStatusSummary.SummaryState.PARTIAL;
+                summaries.add(FunctionStatusSummary.builder()
+                        .name(functionName)
+                        .state(state)
+                        .numInstances(status.getNumInstances())
+                        .numRunning(status.getNumRunning())
+                        .build());
+            } catch (PulsarAdminException e) {
+                summaries.add(FunctionStatusSummary.builder()
+                        .name(functionName)
+                        .state(FunctionStatusSummary.SummaryState.UNKNOWN)
+                        .error(e.getMessage())
+                        .build());
+            }
+        }
+        summaries.sort(Comparator.comparing(FunctionStatusSummary::getName));
+
+        String nextStartAfter = endIndex < pagedNames.size() ? pagedNames.get(endIndex - 1) : null;
+        return FunctionStatusPage.builder()
+                .summaries(summaries)
+                .nextStartAfter(nextStartAfter)
+                .build();
+    }
+
+    /**
+     * Async paginated version of {@link #getFunctionsWithStatus(String, String, Integer, String)}.
+     *
+     * @param tenant
+     *            Tenant name
+     * @param namespace
+     *            Namespace name
+     * @param limit
+     *            Maximum number of functions to return; must be greater than 0 when provided
+     * @param startAfter
+     *            Exclusive cursor (function name) from previous page; null means from beginning
+     * @return a future that completes with the paginated response
+     */
+    default CompletableFuture<FunctionStatusPage> getFunctionsWithStatusAsync(
+            String tenant, String namespace, Integer limit, String startAfter) {
+        try {
+            return CompletableFuture.completedFuture(
+                    getFunctionsWithStatus(tenant, namespace, limit, startAfter));
+        } catch (Exception e) {
+            CompletableFuture<FunctionStatusPage> future = new CompletableFuture<>();
+            future.completeExceptionally(e);
+            return future;
+        }
+    }
+
     /**
      * Get the configuration for the specified function.
      * <p/>

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/FunctionStatusPage.java

@@ -0,0 +1,34 @@
+/*
+ * 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.policies.data;
+
+import java.util.List;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class FunctionStatusPage {
+    private List<FunctionStatusSummary> summaries;
+    private String nextStartAfter;
+}

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/FunctionStatusSummary.java

@@ -0,0 +1,67 @@
+/*
+ * 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.policies.data;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Summary status of a single Pulsar Function, used by the batch
+ * status-summary endpoint to avoid N+1 API calls.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class FunctionStatusSummary {
+
+    /**
+     * Runtime state derived from instance counts.
+     */
+    public enum SummaryState {
+        RUNNING,
+        STOPPED,
+        PARTIAL,
+        UNKNOWN
+    }
+
+    /**
+     * Classification of the status retrieval failure.
+     */
+    public enum ErrorType {
+        AUTHENTICATION_FAILED,
+        FUNCTION_NOT_FOUND,
+        NETWORK_ERROR,
+        INTERNAL_ERROR
+    }
+
+    private String name;
+    private SummaryState state;
+    private int numInstances;
+    private int numRunning;
+
+    /**
+     * Non-null when the status query for this function failed;
+     * the function will have {@code state = UNKNOWN} in that case.
+     */
+    private String error;
+    private ErrorType errorType;
+}