deprecate and relocate connector related endpoints in functions REST API (#4087)

* deprecate and relocate connector related endpoints in functions REST API

* cleaning up

* cleaning up

Author: jerrypeng

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java

@@ -400,6 +400,10 @@ public StreamingOutput downloadFunction(final @QueryParam("path") String path) {
             @ApiResponse(code = 408, message = "Request timeout")
     })
     @Path("/connectors")
+    @Deprecated
+    /**
+     * Deprecated in favor of moving endpoint to {@link org.apache.pulsar.broker.admin.v2.Worker}
+     */
     public List<ConnectorDefinition> getConnectorsList() throws IOException {
         return functions.getListOfConnectors();
     }

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

@@ -24,13 +24,15 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.admin.AdminResource;
 import org.apache.pulsar.common.functions.WorkerInfo;
+import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.rest.api.WorkerImpl;
 
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
+import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -96,4 +98,20 @@ public WorkerInfo getClusterLeader() {
     public Map<String, Collection<String>> getAssignments() {
         return worker.getAssignments();
     }
+
+    @GET
+    @ApiOperation(
+            value = "Fetches a list of supported Pulsar IO connectors currently running in cluster mode",
+            response = List.class
+    )
+    @ApiResponses(value = {
+            @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+            @ApiResponse(code = 400, message = "Invalid request"),
+            @ApiResponse(code = 408, message = "Request timeout")
+    })
+    @Path("/connectors")
+    @Produces(MediaType.APPLICATION_JSON)
+    public List<ConnectorDefinition> getConnectorsList() throws IOException {
+        return worker.getListOfConnectors();
+    }
 }

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

@@ -379,30 +379,39 @@ FunctionStats getFunctionStats(String tenant, String namespace, String function)
     void downloadFunction(String destinationFile, String path) throws PulsarAdminException;
 
     /**
+     * Deprecated in favor of getting sources and sinks for their own APIs
+     *
      * Fetches a list of supported Pulsar IO connectors currently running in cluster mode
      *
      * @throws PulsarAdminException
      *             Unexpected error
      *
      */
+    @Deprecated
     List<ConnectorDefinition> getConnectorsList() throws PulsarAdminException;
 
     /**
+     * Deprecated in favor of getting sources and sinks for their own APIs
+     *
      * Fetches a list of supported Pulsar IO sources currently running in cluster mode
      *
      * @throws PulsarAdminException
      *             Unexpected error
      *
      */
+    @Deprecated
     Set<String> getSources() throws PulsarAdminException;
 
     /**
+     * Deprecated in favor of getting sources and sinks for their own APIs
+     *
      * Fetches a list of supported Pulsar IO sinks currently running in cluster mode
      *
      * @throws PulsarAdminException
      *             Unexpected error
      *
      */
+    @Deprecated
     Set<String> getSinks() throws PulsarAdminException;
 
     /**

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/RestUtils.java

@@ -24,6 +24,8 @@
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
 
+import javax.ws.rs.core.Response;
+
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public final class RestUtils {
 
@@ -36,4 +38,8 @@ public static String createMessage(String message) {
         return createBaseMessage(message).toString();
     }
 
+    public static void throwUnavailableException() {
+        throw new RestException(Response.Status.SERVICE_UNAVAILABLE,
+                "Function worker service is not done initializing. " + "Please try again in a little while.");
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java

@@ -114,10 +114,10 @@
 import static org.apache.pulsar.functions.utils.ComponentType.FUNCTION;
 import static org.apache.pulsar.functions.utils.ComponentType.SINK;
 import static org.apache.pulsar.functions.utils.ComponentType.SOURCE;
-import static org.apache.pulsar.functions.utils.FunctionCommon.extractFileFromPkgURL;
 import static org.apache.pulsar.functions.utils.FunctionCommon.getStateNamespace;
 import static org.apache.pulsar.functions.utils.FunctionCommon.getUniquePackageName;
 import static org.apache.pulsar.functions.worker.WorkerUtils.isFunctionCodeBuiltin;
+import static org.apache.pulsar.functions.worker.rest.RestUtils.throwUnavailableException;
 
 @Slf4j
 public abstract class ComponentImpl {
@@ -1675,11 +1675,6 @@ private void validateTriggerRequestParams(final String tenant,
         }
     }
 
-    protected static void throwUnavailableException() {
-        throw new RestException(Status.SERVICE_UNAVAILABLE,
-                "Function worker service is not done initializing. " + "Please try again in a little while.");
-    }
-
     private void throwStateStoreUnvailableResponse() {
         throw new RestException(Status.SERVICE_UNAVAILABLE,
                 "State storage client is not done initializing. " + "Please try again in a little while.");

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinkImpl.java

@@ -41,6 +41,8 @@
 import java.util.List;
 import java.util.function.Supplier;
 
+import static org.apache.pulsar.functions.worker.rest.RestUtils.throwUnavailableException;
+
 @Slf4j
 public class SinkImpl extends ComponentImpl {
 

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourceImpl.java

@@ -41,6 +41,8 @@
 import java.util.List;
 import java.util.function.Supplier;
 
+import static org.apache.pulsar.functions.worker.rest.RestUtils.throwUnavailableException;
+
 @Slf4j
 public class SourceImpl extends ComponentImpl {
     private class GetSourceStatus extends GetStatus<SourceStatus, SourceStatus.SourceInstanceStatus.SourceInstanceStatusData> {

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java

@@ -20,6 +20,7 @@
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.common.functions.WorkerInfo;
+import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.common.policies.data.ErrorData;
 import org.apache.pulsar.common.policies.data.FunctionStats;
 import org.apache.pulsar.common.policies.data.WorkerFunctionInstanceStats;
@@ -45,6 +46,7 @@
 import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.pulsar.functions.worker.rest.RestUtils.throwUnavailableException;
 
 @Slf4j
 public class WorkerImpl {
@@ -77,17 +79,16 @@ private boolean isWorkerServiceAvailable() {
 
     public List<WorkerInfo> getCluster() {
         if (!isWorkerServiceAvailable()) {
-            throw new RestException(Status.SERVICE_UNAVAILABLE, "Function worker service is not done initializing. Please try again in a little while.");
+            throwUnavailableException();
         }
         List<WorkerInfo> workers = worker().getMembershipManager().getCurrentMembership();
         return workers;
     }
 
     public WorkerInfo getClusterLeader() {
         if (!isWorkerServiceAvailable()) {
-            throw new RestException(Status.SERVICE_UNAVAILABLE, "Function worker service is not done initializing. Please try again in a little while.");
+            throwUnavailableException();
         }
-
         MembershipManager membershipManager = worker().getMembershipManager();
         WorkerInfo leader = membershipManager.getLeader();
 
@@ -99,9 +100,8 @@ public WorkerInfo getClusterLeader() {
     }
 
     public Map<String, Collection<String>> getAssignments() {
-
         if (!isWorkerServiceAvailable()) {
-            throw new RestException(Status.SERVICE_UNAVAILABLE, "Function worker service is not done initializing. Please try again in a little while.");
+            throwUnavailableException();
         }
 
         FunctionRuntimeManager functionRuntimeManager = worker().getFunctionRuntimeManager();
@@ -128,9 +128,7 @@ public List<org.apache.pulsar.common.stats.Metrics> getWorkerMetrics(final Strin
 
     private List<org.apache.pulsar.common.stats.Metrics> getWorkerMetrics() {
         if (!isWorkerServiceAvailable()) {
-            throw new WebApplicationException(
-                    Response.status(Status.SERVICE_UNAVAILABLE).type(MediaType.APPLICATION_JSON)
-                            .entity(new ErrorData("Function worker service is not available")).build());
+            throwUnavailableException();
         }
         return worker().getMetricsGenerator().generate();
     }
@@ -146,7 +144,7 @@ public List<WorkerFunctionInstanceStats> getFunctionsMetrics(String clientRole)
 
     private List<WorkerFunctionInstanceStats> getFunctionsMetrics() throws IOException {
         if (!isWorkerServiceAvailable()) {
-            throw new RestException(Status.SERVICE_UNAVAILABLE, "Function worker service is not done initializing. Please try again in a little while.");
+            throwUnavailableException();
         }
 
         WorkerService workerService = worker();
@@ -184,4 +182,12 @@ private List<WorkerFunctionInstanceStats> getFunctionsMetrics() throws IOExcepti
         }
         return metricsList;
     }
+
+    public List<ConnectorDefinition> getListOfConnectors() {
+        if (!isWorkerServiceAvailable()) {
+            throwUnavailableException();
+        }
+
+        return this.worker().getConnectorsManager().getConnectors();
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java

@@ -311,6 +311,10 @@ public Response downloadFunction(final @QueryParam("path") String path) {
             @ApiResponse(code = 408, message = "Request timeout")
     })
     @Path("/connectors")
+    /**
+     * Deprecated in favor of moving endpoint to {@link org.apache.pulsar.broker.admin.v2.Worker}
+     */
+    @Deprecated
     public List<ConnectorDefinition> getConnectorsList() throws IOException {
         return functions.getListOfConnectors();
     }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java

@@ -25,6 +25,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.web.AuthenticationFilter;
 import org.apache.pulsar.common.functions.WorkerInfo;
+import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.rest.api.WorkerImpl;
 
@@ -36,6 +37,7 @@
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
+import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -120,4 +122,14 @@ public WorkerInfo getClusterLeader() {
     public Map<String, Collection<String>> getAssignments() {
         return worker.getAssignments();
     }
+
+    @ApiResponses(value = {
+            @ApiResponse(code = 403, message = "The requester doesn't have admin permissions"),
+            @ApiResponse(code = 400, message = "Invalid request"),
+            @ApiResponse(code = 408, message = "Request timeout")
+    })
+    @Path("/connectors")
+    public List<ConnectorDefinition> getConnectorsList() throws IOException {
+        return worker.getListOfConnectors();
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3Resource.java

@@ -306,6 +306,10 @@ public StreamingOutput downloadFunction(final @QueryParam("path") String path) {
 
     @GET
     @Path("/connectors")
+    /**
+     * Deprecated in favor of moving endpoint to {@link org.apache.pulsar.broker.admin.v2.Worker}
+     */
+    @Deprecated
     public List<ConnectorDefinition> getConnectorsList() throws IOException {
         return functions.getListOfConnectors();
     }