[pulsar] branch master updated: renaming sinks and sources api to be consistent with the rest of Pulsar (#4363)

[jerrypeng]

This is an automated email from the ASF dual-hosted git repository.

jerrypeng 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 aa40ad6   renaming sinks and sources api to be consistent with the 
rest of Pulsar (#4363)
aa40ad6 is described below

commit aa40ad6ddeb4208d439df2c9080fc73b1e6ffa3b
Author: Boyang Jerry Peng <jerry.boyang.p...@gmail.com>
AuthorDate: Sun May 26 21:48:16 2019 -0700

     renaming sinks and sources api to be consistent with the rest of Pulsar 
(#4363)
    
    * Rename sources and sinks CLI to be consistent with rest of Pulsar
    
    * renaming sinks and sources api to be consistent with the rest of Pulsar
    
    * use new interfaces in cmd
---
 pulsar-broker/pom.xml                              |   4 +-
 .../admin/impl/{SinkBase.java => SinksBase.java}   |  10 +-
 .../impl/{SourceBase.java => SourcesBase.java}     |  12 +-
 .../org/apache/pulsar/broker/admin/v3/Sink.java    |   8 +-
 .../broker/admin/v3/{Sink.java => Sinks.java}      |   8 +-
 .../org/apache/pulsar/broker/admin/v3/Source.java  |   8 +-
 .../broker/admin/v3/{Source.java => Sources.java}  |   8 +-
 .../apache/pulsar/client/admin/PulsarAdmin.java    |  38 ++-
 .../java/org/apache/pulsar/client/admin/Sink.java  | 329 +-------------------
 .../pulsar/client/admin/{Sink.java => Sinks.java}  |   2 +-
 .../org/apache/pulsar/client/admin/Source.java     | 330 +--------------------
 .../client/admin/{Source.java => Sources.java}     |   2 +-
 .../internal/{SinkImpl.java => SinksImpl.java}     |   5 +-
 .../internal/{SourceImpl.java => SourcesImpl.java} |   5 +-
 .../java/org/apache/pulsar/admin/cli/CmdSinks.java |  36 +--
 .../org/apache/pulsar/admin/cli/CmdSources.java    |  36 +--
 .../apache/pulsar/admin/cli/PulsarAdminTool.java   |  29 +-
 .../org/apache/pulsar/admin/cli/TestCmdSinks.java  |   8 +-
 .../apache/pulsar/admin/cli/TestCmdSources.java    |   8 +-
 .../pulsar/functions/worker/rest/Resources.java    |  12 +-
 .../rest/api/{SinkImpl.java => SinksImpl.java}     |   4 +-
 .../rest/api/{SourceImpl.java => SourcesImpl.java} |   4 +-
 ...V2Resource.java => FunctionsApiV2Resource.java} |   4 +-
 ...V3Resource.java => FunctionsApiV3Resource.java} |   4 +-
 .../worker/rest/api/v3/SinkApiV3Resource.java      | 224 +-------------
 ...kApiV3Resource.java => SinksApiV3Resource.java} |  16 +-
 .../worker/rest/api/v3/SourceApiV3Resource.java    | 228 +-------------
 ...piV3Resource.java => SourcesApiV3Resource.java} |  28 +-
 .../rest/api/v2/FunctionApiV2ResourceTest.java     |   2 +-
 .../rest/api/v3/FunctionApiV3ResourceTest.java     |   5 +-
 .../worker/rest/api/v3/SinkApiV3ResourceTest.java  |   8 +-
 .../rest/api/v3/SourceApiV3ResourceTest.java       |   8 +-
 32 files changed, 224 insertions(+), 1209 deletions(-)

diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index ac63f00..c9f7e36 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -404,7 +404,7 @@
                 </apiSource>
                 <apiSource>
                   <springmvc>false</springmvc>
-                  
<locations>org.apache.pulsar.broker.admin.v3.Source</locations>
+                  
<locations>org.apache.pulsar.broker.admin.v3.Sources</locations>
                   <schemes>http,https</schemes>
                   <basePath>/admin/v3</basePath>
                   <info>
@@ -421,7 +421,7 @@
                 </apiSource>
                 <apiSource>
                   <springmvc>false</springmvc>
-                  <locations>org.apache.pulsar.broker.admin.v3.Sink</locations>
+                  
<locations>org.apache.pulsar.broker.admin.v3.Sinks</locations>
                   <schemes>http,https</schemes>
                   <basePath>/admin/v3</basePath>
                   <info>
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinksBase.java
similarity index 98%
rename from 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java
rename to 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinksBase.java
index f15003c..50839d8 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinksBase.java
@@ -28,7 +28,7 @@ import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.common.io.SinkConfig;
 import org.apache.pulsar.common.policies.data.SinkStatus;
 import org.apache.pulsar.functions.worker.WorkerService;
-import org.apache.pulsar.functions.worker.rest.api.SinkImpl;
+import org.apache.pulsar.functions.worker.rest.api.SinksImpl;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.glassfish.jersey.media.multipart.FormDataParam;
 
@@ -47,12 +47,12 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Supplier;
 
-public class SinkBase extends AdminResource implements Supplier<WorkerService> 
{
+public class SinksBase extends AdminResource implements 
Supplier<WorkerService> {
 
-    private final SinkImpl sink;
+    private final SinksImpl sink;
 
-    public SinkBase() {
-        this.sink = new SinkImpl(this);
+    public SinksBase() {
+        this.sink = new SinksImpl(this);
     }
 
     @Override
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourcesBase.java
similarity index 97%
rename from 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java
rename to 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourcesBase.java
index f55337e..e65eda7 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourcesBase.java
@@ -28,7 +28,7 @@ import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.common.policies.data.SourceStatus;
 import org.apache.pulsar.functions.worker.WorkerService;
-import org.apache.pulsar.functions.worker.rest.api.SourceImpl;
+import org.apache.pulsar.functions.worker.rest.api.SourcesImpl;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.glassfish.jersey.media.multipart.FormDataParam;
 
@@ -47,12 +47,12 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Supplier;
 
-public class SourceBase extends AdminResource implements 
Supplier<WorkerService> {
+public class SourcesBase extends AdminResource implements 
Supplier<WorkerService> {
 
-    private final SourceImpl source;
+    private final SourcesImpl source;
 
-    public SourceBase() {
-        this.source = new SourceImpl(this);
+    public SourcesBase() {
+        this.source = new SourcesImpl(this);
     }
 
     @Override
@@ -188,6 +188,7 @@ public class SourceBase extends AdminResource implements 
Supplier<WorkerService>
             @ApiResponse(code = 400, message = "Invalid request"),
             @ApiResponse(code = 403, message = "The requester doesn't have 
admin permissions")
     })
+    @Consumes(MediaType.APPLICATION_JSON)
     @Path("/{tenant}/{namespace}")
     public List<String> listSources(final @PathParam("tenant") String tenant,
                                     final @PathParam("namespace") String 
namespace) {
@@ -291,6 +292,7 @@ public class SourceBase extends AdminResource implements 
Supplier<WorkerService>
             @ApiResponse(code = 400, message = "Invalid request"),
             @ApiResponse(code = 408, message = "Request timeout")
     })
+    @Produces(MediaType.APPLICATION_JSON)
     @Path("/builtinsources")
     public List<ConnectorDefinition> getSourceList() {
         List<ConnectorDefinition> connectorDefinitions = 
source.getListOfConnectors();
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java
index 9191d8c..667e88e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java
@@ -19,7 +19,7 @@
 package org.apache.pulsar.broker.admin.v3;
 
 import io.swagger.annotations.Api;
-import org.apache.pulsar.broker.admin.impl.SinkBase;
+import org.apache.pulsar.broker.admin.impl.SinksBase;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.Path;
@@ -30,5 +30,9 @@ import javax.ws.rs.core.MediaType;
 @Api(value = "/sink", description = "Sink admin apis", tags = "sink")
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
-public class Sink extends SinkBase {
+@Deprecated
+/**
+ * @deprecated in favor of {@link Sinks}
+ */
+public class Sink extends SinksBase {
 }
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sinks.java
similarity index 85%
copy from 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java
copy to pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sinks.java
index 9191d8c..a693ea2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sink.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sinks.java
@@ -19,16 +19,16 @@
 package org.apache.pulsar.broker.admin.v3;
 
 import io.swagger.annotations.Api;
-import org.apache.pulsar.broker.admin.impl.SinkBase;
+import org.apache.pulsar.broker.admin.impl.SinksBase;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 
-@Path("/sink")
-@Api(value = "/sink", description = "Sink admin apis", tags = "sink")
+@Path("/sinks")
+@Api(value = "/sinks", description = "Sinks admin apis", tags = "sinks")
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
-public class Sink extends SinkBase {
+public class Sinks extends SinksBase {
 }
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java
index b8acd19..f68d228 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java
@@ -19,7 +19,7 @@
 package org.apache.pulsar.broker.admin.v3;
 
 import io.swagger.annotations.Api;
-import org.apache.pulsar.broker.admin.impl.SourceBase;
+import org.apache.pulsar.broker.admin.impl.SourcesBase;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.Path;
@@ -30,5 +30,9 @@ import javax.ws.rs.core.MediaType;
 @Api(value = "/source", description = "Source admin apis", tags = "source")
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
-public class Source extends SourceBase {
+@Deprecated
+/**
+ * @deprecated in favor of {@link Sources}
+ */
+public class Source extends SourcesBase {
 }
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sources.java
similarity index 84%
copy from 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java
copy to 
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sources.java
index b8acd19..a36ab08 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Source.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sources.java
@@ -19,16 +19,16 @@
 package org.apache.pulsar.broker.admin.v3;
 
 import io.swagger.annotations.Api;
-import org.apache.pulsar.broker.admin.impl.SourceBase;
+import org.apache.pulsar.broker.admin.impl.SourcesBase;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 
-@Path("/source")
-@Api(value = "/source", description = "Source admin apis", tags = "source")
+@Path("/sources")
+@Api(value = "/sources", description = "Sources admin apis", tags = "sources")
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
-public class Source extends SourceBase {
+public class Sources extends SourcesBase {
 }
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
index 9ab3d6d..2f182f7 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PulsarAdmin.java
@@ -40,8 +40,8 @@ import 
org.apache.pulsar.client.admin.internal.NonPersistentTopicsImpl;
 import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl;
 import org.apache.pulsar.client.admin.internal.ResourceQuotasImpl;
 import org.apache.pulsar.client.admin.internal.SchemasImpl;
-import org.apache.pulsar.client.admin.internal.SinkImpl;
-import org.apache.pulsar.client.admin.internal.SourceImpl;
+import org.apache.pulsar.client.admin.internal.SinksImpl;
+import org.apache.pulsar.client.admin.internal.SourcesImpl;
 import org.apache.pulsar.client.admin.internal.TenantsImpl;
 import org.apache.pulsar.client.admin.internal.TopicsImpl;
 import org.apache.pulsar.client.admin.internal.WorkerImpl;
@@ -87,8 +87,8 @@ public class PulsarAdmin implements Closeable {
     private final String serviceUrl;
     private final Lookup lookups;
     private final Functions functions;
-    private final Source source;
-    private final Sink sink;
+    private final Sources sources;
+    private final Sinks sinks;
     private final Worker worker;
     private final Schemas schemas;
     protected final WebTarget root;
@@ -193,8 +193,8 @@ public class PulsarAdmin implements Closeable {
         this.resourceQuotas = new ResourceQuotasImpl(root, auth);
         this.lookups = new LookupImpl(root, auth, useTls);
         this.functions = new FunctionsImpl(root, auth, httpAsyncClient);
-        this.source = new SourceImpl(root, auth, httpAsyncClient);
-        this.sink = new SinkImpl(root, auth, httpAsyncClient);
+        this.sources = new SourcesImpl(root, auth, httpAsyncClient);
+        this.sinks = new SinksImpl(root, auth, httpAsyncClient);
         this.worker = new WorkerImpl(root, auth);
         this.schemas = new SchemasImpl(root, auth);
         this.bookies = new BookiesImpl(root, auth);
@@ -340,23 +340,35 @@ public class PulsarAdmin implements Closeable {
     }
 
     /**
-     *
-     * @return the source management object
+     * @return the sources management object
+     * @deprecated in favor of {@link #sources()}
      */
+    @Deprecated
     public Source source() {
-        return source;
+        return (Source) sources;
+    }
+
+    public Sources sources() {
+        return sources;
     }
 
     /**
-     *
-     * @return the sink management object
+     * @return the sinks management object
+     * @deprecated in favor of {@link #sinks}
      */
+    @Deprecated
     public Sink sink() {
-        return sink;
+        return (Sink) sinks;
+    }
+
+    /**
+     * @return the sinks management object
+     */
+    public Sinks sinks() {
+        return sinks;
     }
 
     /**
-    *
     * @return the Worker stats
     */
    public Worker worker() {
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
index 8db53aa..c270a75 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
@@ -18,333 +18,10 @@
  */
 package org.apache.pulsar.client.admin;
 
-import 
org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
-import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
-import 
org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
-import org.apache.pulsar.common.functions.UpdateOptions;
-import org.apache.pulsar.common.io.ConnectorDefinition;
-import org.apache.pulsar.common.policies.data.SinkStatus;
-import org.apache.pulsar.common.io.SinkConfig;
-
-import java.util.List;
-
 /**
- * Admin interface for Sink management.
+ * @deprecated in favor of {@link Sinks}
  */
-public interface Sink {
-    /**
-     * Get the list of sinks.
-     * <p>
-     * Get the list of all the Pulsar Sinks.
-     * <p>
-     * Response Example:
-     *
-     * <pre>
-     * <code>["f1", "f2", "f3"]</code>
-     * </pre>
-     *
-     * @throws NotAuthorizedException
-     *             Don't have admin permission
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    List<String> listSinks(String tenant, String namespace) throws 
PulsarAdminException;
-
-    /**
-     * Get the configuration for the specified sink.
-     * <p>
-     * Response Example:
-     *
-     * <pre>
-     * <code>{ serviceUrl : "http://my-broker.example.com:8080/"; }</code>
-     * </pre>
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @return the sink configuration
-     *
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to get the configuration of 
the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    SinkConfig getSink(String tenant, String namespace, String sink) throws 
PulsarAdminException;
-
-    /**
-     * Create a new sink.
-     *
-     * @param sinkConfig
-     *            the sink configuration object
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void createSink(SinkConfig sinkConfig, String fileName) throws 
PulsarAdminException;
-
-    /**
-     * <pre>
-     * Create a new sink by providing url from which fun-pkg can be 
downloaded. supported url: http/file
-     * eg:
-     * File: file:/dir/fileName.jar
-     * Http: http://www.repo.com/fileName.jar
-     * </pre>
-     *
-     * @param sinkConfig
-     *            the sink configuration object
-     * @param pkgUrl
-     *            url from which pkg can be downloaded
-     * @throws PulsarAdminException
-     */
-    void createSinkWithUrl(SinkConfig sinkConfig, String pkgUrl) throws 
PulsarAdminException;
-
-    /**
-     * Update the configuration for a sink.
-     * <p>
-     *
-     * @param sinkConfig
-     *            the sink configuration object
-     *
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSink(SinkConfig sinkConfig, String fileName) throws 
PulsarAdminException;
-
-    /**
-     * Update the configuration for a sink.
-     * <p>
-     *
-     * @param sinkConfig
-     *            the sink configuration object
-     * @param updateOptions
-     *            options for the update operations
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSink(SinkConfig sinkConfig, String fileName, UpdateOptions 
updateOptions) throws PulsarAdminException;
-
-    /**
-     * Update the configuration for a sink.
-     * <pre>
-     * Update a sink by providing url from which fun-pkg can be downloaded. 
supported url: http/file
-     * eg:
-     * File: file:/dir/fileName.jar
-     * Http: http://www.repo.com/fileName.jar
-     * </pre>
-     *
-     * @param sinkConfig
-     *            the sink configuration object
-     * @param pkgUrl
-     *            url from which pkg can be downloaded
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSinkWithUrl(SinkConfig sinkConfig, String pkgUrl) throws 
PulsarAdminException;
-
-    /**
-     * Update the configuration for a sink.
-     * <pre>
-     * Update a sink by providing url from which fun-pkg can be downloaded. 
supported url: http/file
-     * eg:
-     * File: file:/dir/fileName.jar
-     * Http: http://www.repo.com/fileName.jar
-     * </pre>
-     *
-     * @param sinkConfig
-     *            the sink configuration object
-     * @param pkgUrl
-     *            url from which pkg can be downloaded
-     * @param updateOptions
-     *            options for the update operations
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSinkWithUrl(SinkConfig sinkConfig, String pkgUrl, UpdateOptions 
updateOptions) throws PulsarAdminException;
-
-    /**
-     * Delete an existing sink
-     * <p>
-     * Delete a sink
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @throws NotAuthorizedException
-     *             You don't have admin permission
-     * @throws NotFoundException
-     *             Cluster does not exist
-     * @throws PreconditionFailedException
-     *             Cluster is not empty
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void deleteSink(String tenant, String namespace, String sink) throws 
PulsarAdminException;
-
-    /**
-     * Gets the current status of a sink.
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    SinkStatus getSinkStatus(String tenant, String namespace, String sink) 
throws PulsarAdminException;
-
-    /**
-     * Gets the current status of a sink instance.
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     * @param id
-     *            Sink instance-id
-     * @return
-     * @throws PulsarAdminException
-     */
-    SinkStatus.SinkInstanceStatus.SinkInstanceStatusData getSinkStatus(String 
tenant, String namespace, String sink, int id)
-            throws PulsarAdminException;
-
-    /**
-     * Restart sink instance
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @param instanceId
-     *            Sink instanceId
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void restartSink(String tenant, String namespace, String sink, int 
instanceId) throws PulsarAdminException;
-
-    /**
-     * Restart all sink instances
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void restartSink(String tenant, String namespace, String sink) throws 
PulsarAdminException;
-
-
-    /**
-     * Stop sink instance
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @param instanceId
-     *            Sink instanceId
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void stopSink(String tenant, String namespace, String sink, int 
instanceId) throws PulsarAdminException;
-
-    /**
-     * Stop all sink instances
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void stopSink(String tenant, String namespace, String sink) throws 
PulsarAdminException;
-
-    /**
-     * Start sink instance
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @param instanceId
-     *            Sink instanceId
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void startSink(String tenant, String namespace, String sink, int 
instanceId) throws PulsarAdminException;
-
-    /**
-     * Start all sink instances
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param sink
-     *            Sink name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void startSink(String tenant, String namespace, String sink) throws 
PulsarAdminException;
-
+@Deprecated
+public interface Sink extends Sinks {
 
-    /**
-     * Fetches a list of supported Pulsar IO sinks currently running in 
cluster mode
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     *
-     */
-    List<ConnectorDefinition> getBuiltInSinks() throws PulsarAdminException;
 }
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sinks.java
similarity index 99%
copy from 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
copy to 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sinks.java
index 8db53aa..afa44e0 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sink.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sinks.java
@@ -31,7 +31,7 @@ import java.util.List;
 /**
  * Admin interface for Sink management.
  */
-public interface Sink {
+public interface Sinks {
     /**
      * Get the list of sinks.
      * <p>
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
index 058364e..aba3251 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
@@ -18,333 +18,9 @@
  */
 package org.apache.pulsar.client.admin;
 
-import 
org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
-import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
-import 
org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
-import org.apache.pulsar.common.functions.UpdateOptions;
-import org.apache.pulsar.common.io.ConnectorDefinition;
-import org.apache.pulsar.common.io.SourceConfig;
-import org.apache.pulsar.common.policies.data.SourceStatus;
-
-import java.util.List;
-
 /**
- * Admin interface for Source management.
+ * @deprecated in favor of {@link Sources}
  */
-public interface Source {
-    /**
-     * Get the list of sources.
-     * <p>
-     * Get the list of all the Pulsar Sources.
-     * <p>
-     * Response Example:
-     *
-     * <pre>
-     * <code>["f1", "f2", "f3"]</code>
-     * </pre>
-     *
-     * @throws NotAuthorizedException
-     *             Don't have admin permission
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    List<String> listSources(String tenant, String namespace) throws 
PulsarAdminException;
-
-    /**
-     * Get the configuration for the specified source.
-     * <p>
-     * Response Example:
-     *
-     * <pre>
-     * <code>{ serviceUrl : "http://my-broker.example.com:8080/"; }</code>
-     * </pre>
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @return the source configuration
-     *
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to get the configuration of 
the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    SourceConfig getSource(String tenant, String namespace, String source) 
throws PulsarAdminException;
-
-    /**
-     * Create a new source.
-     *
-     * @param sourceConfig
-     *            the source configuration object
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void createSource(SourceConfig sourceConfig, String fileName) throws 
PulsarAdminException;
-
-    /**
-     * <pre>
-     * Create a new source by providing url from which fun-pkg can be 
downloaded. supported url: http/file
-     * eg:
-     * File: file:/dir/fileName.jar
-     * Http: http://www.repo.com/fileName.jar
-     * </pre>
-     *
-     * @param sourceConfig
-     *            the source configuration object
-     * @param pkgUrl
-     *            url from which pkg can be downloaded
-     * @throws PulsarAdminException
-     */
-    void createSourceWithUrl(SourceConfig sourceConfig, String pkgUrl) throws 
PulsarAdminException;
-
-    /**
-     * Update the configuration for a source.
-     * <p>
-     *
-     * @param sourceConfig
-     *            the source configuration object
-     *
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSource(SourceConfig sourceConfig, String fileName) throws 
PulsarAdminException;
-
-    /**
-     * Update the configuration for a source.
-     * <p>
-     *
-     * @param sourceConfig
-     *            the source configuration object
-     * @param updateOptions
-     *            options for the update operations
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSource(SourceConfig sourceConfig, String fileName, 
UpdateOptions updateOptions) throws PulsarAdminException;
-
-    /**
-     * Update the configuration for a source.
-     * <pre>
-     * Update a source by providing url from which fun-pkg can be downloaded. 
supported url: http/file
-     * eg:
-     * File: file:/dir/fileName.jar
-     * Http: http://www.repo.com/fileName.jar
-     * </pre>
-     *
-     * @param sourceConfig
-     *            the source configuration object
-     * @param pkgUrl
-     *            url from which pkg can be downloaded
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSourceWithUrl(SourceConfig sourceConfig, String pkgUrl) throws 
PulsarAdminException;
-
-    /**
-     * Update the configuration for a source.
-     * <pre>
-     * Update a source by providing url from which fun-pkg can be downloaded. 
supported url: http/file
-     * eg:
-     * File: file:/dir/fileName.jar
-     * Http: http://www.repo.com/fileName.jar
-     * </pre>
-     *
-     * @param sourceConfig
-     *            the source configuration object
-     * @param pkgUrl
-     *            url from which pkg can be downloaded
-     * @param updateOptions
-     *            options for the update operations
-     * @throws NotAuthorizedException
-     *             You don't have admin permission to create the cluster
-     * @throws NotFoundException
-     *             Cluster doesn't exist
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void updateSourceWithUrl(SourceConfig sourceConfig, String pkgUrl, 
UpdateOptions updateOptions) throws PulsarAdminException;
-
-    /**
-     * Delete an existing source
-     * <p>
-     * Delete a source
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @throws NotAuthorizedException
-     *             You don't have admin permission
-     * @throws NotFoundException
-     *             Cluster does not exist
-     * @throws PreconditionFailedException
-     *             Cluster is not empty
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void deleteSource(String tenant, String namespace, String source) throws 
PulsarAdminException;
-
-    /**
-     * Gets the current status of a source.
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    SourceStatus getSourceStatus(String tenant, String namespace, String 
source) throws PulsarAdminException;
-
-    /**
-     * Gets the current status of a source instance.
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     * @param id
-     *            Source instance-id
-     * @return
-     * @throws PulsarAdminException
-     */
-    SourceStatus.SourceInstanceStatus.SourceInstanceStatusData 
getSourceStatus(String tenant, String namespace, String source, int id)
-            throws PulsarAdminException;
-
-    /**
-     * Restart source instance
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @param instanceId
-     *            Source instanceId
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void restartSource(String tenant, String namespace, String source, int 
instanceId) throws PulsarAdminException;
-
-    /**
-     * Restart all source instances
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void restartSource(String tenant, String namespace, String source) throws 
PulsarAdminException;
-
-
-    /**
-     * Stop source instance
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @param instanceId
-     *            Source instanceId
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void stopSource(String tenant, String namespace, String source, int 
instanceId) throws PulsarAdminException;
-
-    /**
-     * Stop all source instances
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void stopSource(String tenant, String namespace, String source) throws 
PulsarAdminException;
-
-    /**
-     * Start source instance
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @param instanceId
-     *            Source instanceId
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void startSource(String tenant, String namespace, String source, int 
instanceId) throws PulsarAdminException;
-
-    /**
-     * Start all source instances
-     *
-     * @param tenant
-     *            Tenant name
-     * @param namespace
-     *            Namespace name
-     * @param source
-     *            Source name
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     */
-    void startSource(String tenant, String namespace, String source) throws 
PulsarAdminException;
-
-
-    /**
-     * Fetches a list of supported Pulsar IO sources currently running in 
cluster mode
-     *
-     * @throws PulsarAdminException
-     *             Unexpected error
-     *
-     */
-    List<ConnectorDefinition> getBuiltInSources() throws PulsarAdminException;
+@Deprecated
+public interface Source extends Sources {
 }
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sources.java
similarity index 99%
copy from 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
copy to 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sources.java
index 058364e..ef108e2 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Source.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Sources.java
@@ -31,7 +31,7 @@ import java.util.List;
 /**
  * Admin interface for Source management.
  */
-public interface Source {
+public interface Sources {
     /**
      * Get the list of sources.
      * <p>
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinkImpl.java
 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
similarity index 98%
rename from 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinkImpl.java
rename to 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
index 3da82ce..77a25cc 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinkImpl.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SinksImpl.java
@@ -22,6 +22,7 @@ import com.google.gson.Gson;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.Sink;
+import org.apache.pulsar.client.admin.Sinks;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.common.functions.UpdateOptions;
 import org.apache.pulsar.common.io.ConnectorDefinition;
@@ -48,12 +49,12 @@ import static org.asynchttpclient.Dsl.post;
 import static org.asynchttpclient.Dsl.put;
 
 @Slf4j
-public class SinkImpl extends ComponentResource implements Sink {
+public class SinksImpl extends ComponentResource implements Sinks, Sink {
 
     private final WebTarget sink;
     private final AsyncHttpClient asyncHttpClient;
 
-    public SinkImpl(WebTarget web, Authentication auth, AsyncHttpClient 
asyncHttpClient) {
+    public SinksImpl(WebTarget web, Authentication auth, AsyncHttpClient 
asyncHttpClient) {
         super(auth);
         this.sink = web.path("/admin/v3/sink");
         this.asyncHttpClient = asyncHttpClient;
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourceImpl.java
 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
similarity index 98%
rename from 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourceImpl.java
rename to 
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
index cd7b838..d8f57a3 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourceImpl.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/SourcesImpl.java
@@ -22,6 +22,7 @@ import com.google.gson.Gson;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.Source;
+import org.apache.pulsar.client.admin.Sources;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.common.functions.UpdateOptions;
 import org.apache.pulsar.common.io.ConnectorDefinition;
@@ -48,12 +49,12 @@ import static org.asynchttpclient.Dsl.post;
 import static org.asynchttpclient.Dsl.put;
 
 @Slf4j
-public class SourceImpl extends ComponentResource implements Source {
+public class SourcesImpl extends ComponentResource implements Sources, Source {
 
     private final WebTarget source;
     private final AsyncHttpClient asyncHttpClient;
 
-    public SourceImpl(WebTarget web, Authentication auth, AsyncHttpClient 
asyncHttpClient) {
+    public SourcesImpl(WebTarget web, Authentication auth, AsyncHttpClient 
asyncHttpClient) {
         super(auth);
         this.source = web.path("/admin/v3/source");
         this.asyncHttpClient = asyncHttpClient;
diff --git 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index 0f2cecb..b6029a9 100644
--- 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++ 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -70,7 +70,7 @@ public class CmdSinks extends CmdBase {
     private final LocalSinkRunner localSinkRunner;
 
     public CmdSinks(PulsarAdmin admin) {
-        super("sink", admin);
+        super("sinks", admin);
         createSink = new CreateSink();
         updateSink = new UpdateSink();
         deleteSink = new DeleteSink();
@@ -202,9 +202,9 @@ public class CmdSinks extends CmdBase {
         @Override
         void runCmd() throws Exception {
             if (Utils.isFunctionPackageUrlSupported(archive)) {
-                admin.sink().createSinkWithUrl(sinkConfig, 
sinkConfig.getArchive());
+                admin.sinks().createSinkWithUrl(sinkConfig, 
sinkConfig.getArchive());
             } else {
-                admin.sink().createSink(sinkConfig, sinkConfig.getArchive());
+                admin.sinks().createSink(sinkConfig, sinkConfig.getArchive());
             }
             print("Created successfully");
         }
@@ -221,9 +221,9 @@ public class CmdSinks extends CmdBase {
             UpdateOptions updateOptions = new UpdateOptions();
             updateOptions.setUpdateAuthData(updateAuthData);
             if (Utils.isFunctionPackageUrlSupported(archive)) {
-                admin.sink().updateSinkWithUrl(sinkConfig, 
sinkConfig.getArchive(), updateOptions);
+                admin.sinks().updateSinkWithUrl(sinkConfig, 
sinkConfig.getArchive(), updateOptions);
             } else {
-                admin.sink().updateSink(sinkConfig, sinkConfig.getArchive(), 
updateOptions);
+                admin.sinks().updateSink(sinkConfig, sinkConfig.getArchive(), 
updateOptions);
             }
             print("Updated successfully");
         }
@@ -465,7 +465,7 @@ public class CmdSinks extends CmdBase {
         protected String validateSinkType(String sinkType) throws IOException {
             Set<String> availableSinks;
             try {
-                availableSinks = 
admin.sink().getBuiltInSinks().stream().map(ConnectorDefinition::getName).collect(Collectors.toSet());
+                availableSinks = 
admin.sinks().getBuiltInSinks().stream().map(ConnectorDefinition::getName).collect(Collectors.toSet());
             } catch (PulsarAdminException e) {
                 throw new IOException(e);
             }
@@ -515,7 +515,7 @@ public class CmdSinks extends CmdBase {
 
         @Override
         void runCmd() throws Exception {
-            admin.sink().deleteSink(tenant, namespace, sinkName);
+            admin.sinks().deleteSink(tenant, namespace, sinkName);
             print("Deleted successfully");
         }
     }
@@ -525,7 +525,7 @@ public class CmdSinks extends CmdBase {
 
         @Override
         void runCmd() throws Exception {
-            SinkConfig sinkConfig = admin.sink().getSink(tenant, namespace, 
sinkName);
+            SinkConfig sinkConfig = admin.sinks().getSink(tenant, namespace, 
sinkName);
             Gson gson = new GsonBuilder().setPrettyPrinting().create();
             System.out.println(gson.toJson(sinkConfig));
         }
@@ -554,7 +554,7 @@ public class CmdSinks extends CmdBase {
 
         @Override
         void runCmd() throws Exception {
-            List<String> sinks = admin.sink().listSinks(tenant, namespace);
+            List<String> sinks = admin.sinks().listSinks(tenant, namespace);
             Gson gson = new GsonBuilder().setPrettyPrinting().create();
             System.out.println(gson.toJson(sinks));
         }
@@ -569,9 +569,9 @@ public class CmdSinks extends CmdBase {
         @Override
         void runCmd() throws Exception {
             if (isBlank(instanceId)) {
-                print(admin.sink().getSinkStatus(tenant, namespace, sinkName));
+                print(admin.sinks().getSinkStatus(tenant, namespace, 
sinkName));
             } else {
-                print(admin.sink().getSinkStatus(tenant, namespace, sinkName, 
Integer.parseInt(instanceId)));
+                print(admin.sinks().getSinkStatus(tenant, namespace, sinkName, 
Integer.parseInt(instanceId)));
             }
         }
     }
@@ -586,12 +586,12 @@ public class CmdSinks extends CmdBase {
         void runCmd() throws Exception {
             if (isNotBlank(instanceId)) {
                 try {
-                    admin.sink().restartSink(tenant, namespace, sinkName, 
Integer.parseInt(instanceId));
+                    admin.sinks().restartSink(tenant, namespace, sinkName, 
Integer.parseInt(instanceId));
                 } catch (NumberFormatException e) {
                     System.err.println("instance-id must be a number");
                 }
             } else {
-                admin.sink().restartSink(tenant, namespace, sinkName);
+                admin.sinks().restartSink(tenant, namespace, sinkName);
             }
             System.out.println("Restarted successfully");
         }
@@ -607,12 +607,12 @@ public class CmdSinks extends CmdBase {
         void runCmd() throws Exception {
             if (isNotBlank(instanceId)) {
                 try {
-                    admin.sink().stopSink(tenant, namespace, sinkName, 
Integer.parseInt(instanceId));
+                    admin.sinks().stopSink(tenant, namespace, sinkName, 
Integer.parseInt(instanceId));
                 } catch (NumberFormatException e) {
                     System.err.println("instance-id must be a number");
                 }
             } else {
-                admin.sink().stopSink(tenant, namespace, sinkName);
+                admin.sinks().stopSink(tenant, namespace, sinkName);
             }
             System.out.println("Stopped successfully");
         }
@@ -628,12 +628,12 @@ public class CmdSinks extends CmdBase {
         void runCmd() throws Exception {
             if (isNotBlank(instanceId)) {
                 try {
-                    admin.sink().startSink(tenant, namespace, sinkName, 
Integer.parseInt(instanceId));
+                    admin.sinks().startSink(tenant, namespace, sinkName, 
Integer.parseInt(instanceId));
                 } catch (NumberFormatException e) {
                     System.err.println("instance-id must be a number");
                 }
             } else {
-                admin.sink().startSink(tenant, namespace, sinkName);
+                admin.sinks().startSink(tenant, namespace, sinkName);
             }
             System.out.println("Started successfully");
         }
@@ -643,7 +643,7 @@ public class CmdSinks extends CmdBase {
     public class ListBuiltInSinks extends BaseCommand {
         @Override
         void runCmd() throws Exception {
-            admin.sink().getBuiltInSinks().stream().filter(x -> 
isNotBlank(x.getSinkClass()))
+            admin.sinks().getBuiltInSinks().stream().filter(x -> 
isNotBlank(x.getSinkClass()))
                     .forEach(connector -> {
                         System.out.println(connector.getName());
                         
System.out.println(WordUtils.wrap(connector.getDescription(), 80));
diff --git 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
index 782b0f6..535bb33 100644
--- 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
+++ 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
@@ -73,7 +73,7 @@ public class CmdSources extends CmdBase {
     private final LocalSourceRunner localSourceRunner;
 
     public CmdSources(PulsarAdmin admin) {
-        super("source", admin);
+        super("sources", admin);
         createSource = new CreateSource();
         updateSource = new UpdateSource();
         deleteSource = new DeleteSource();
@@ -206,9 +206,9 @@ public class CmdSources extends CmdBase {
         @Override
         void runCmd() throws Exception {
             if 
(Utils.isFunctionPackageUrlSupported(this.sourceConfig.getArchive())) {
-                admin.source().createSourceWithUrl(sourceConfig, 
sourceConfig.getArchive());
+                admin.sources().createSourceWithUrl(sourceConfig, 
sourceConfig.getArchive());
             } else {
-                admin.source().createSource(sourceConfig, 
sourceConfig.getArchive());
+                admin.sources().createSource(sourceConfig, 
sourceConfig.getArchive());
             }
             print("Created successfully");
         }
@@ -225,9 +225,9 @@ public class CmdSources extends CmdBase {
             UpdateOptions updateOptions = new UpdateOptions();
             updateOptions.setUpdateAuthData(updateAuthData);
             if 
(Utils.isFunctionPackageUrlSupported(sourceConfig.getArchive())) {
-                admin.source().updateSourceWithUrl(sourceConfig, 
sourceConfig.getArchive(), updateOptions);
+                admin.sources().updateSourceWithUrl(sourceConfig, 
sourceConfig.getArchive(), updateOptions);
             } else {
-                admin.source().updateSource(sourceConfig, 
sourceConfig.getArchive(), updateOptions);
+                admin.sources().updateSource(sourceConfig, 
sourceConfig.getArchive(), updateOptions);
             }
             print("Updated successfully");
         }
@@ -419,7 +419,7 @@ public class CmdSources extends CmdBase {
         protected String validateSourceType(String sourceType) throws 
IOException {
             Set<String> availableSources;
             try {
-                availableSources = 
admin.source().getBuiltInSources().stream().map(ConnectorDefinition::getName).collect(Collectors.toSet());
+                availableSources = 
admin.sources().getBuiltInSources().stream().map(ConnectorDefinition::getName).collect(Collectors.toSet());
             } catch (PulsarAdminException e) {
                 throw new IOException(e);
             }
@@ -469,7 +469,7 @@ public class CmdSources extends CmdBase {
 
         @Override
         void runCmd() throws Exception {
-            admin.source().deleteSource(tenant, namespace, sourceName);
+            admin.sources().deleteSource(tenant, namespace, sourceName);
             print("Delete source successfully");
         }
     }
@@ -479,7 +479,7 @@ public class CmdSources extends CmdBase {
 
         @Override
         void runCmd() throws Exception {
-            SourceConfig sourceConfig = admin.source().getSource(tenant, 
namespace, sourceName);
+            SourceConfig sourceConfig = admin.sources().getSource(tenant, 
namespace, sourceName);
             Gson gson = new GsonBuilder().setPrettyPrinting().create();
             System.out.println(gson.toJson(sourceConfig));
         }
@@ -508,7 +508,7 @@ public class CmdSources extends CmdBase {
 
         @Override
         void runCmd() throws Exception {
-            List<String> sources = admin.source().listSources(tenant, 
namespace);
+            List<String> sources = admin.sources().listSources(tenant, 
namespace);
             Gson gson = new GsonBuilder().setPrettyPrinting().create();
             System.out.println(gson.toJson(sources));
         }
@@ -523,9 +523,9 @@ public class CmdSources extends CmdBase {
         @Override
         void runCmd() throws Exception {
             if (isBlank(instanceId)) {
-                print(admin.source().getSourceStatus(tenant, namespace, 
sourceName));
+                print(admin.sources().getSourceStatus(tenant, namespace, 
sourceName));
             } else {
-                print(admin.source().getSourceStatus(tenant, namespace, 
sourceName, Integer.parseInt(instanceId)));
+                print(admin.sources().getSourceStatus(tenant, namespace, 
sourceName, Integer.parseInt(instanceId)));
             };
         }
     }
@@ -540,12 +540,12 @@ public class CmdSources extends CmdBase {
         void runCmd() throws Exception {
             if (isNotBlank(instanceId)) {
                 try {
-                    admin.source().restartSource(tenant, namespace, 
sourceName, Integer.parseInt(instanceId));
+                    admin.sources().restartSource(tenant, namespace, 
sourceName, Integer.parseInt(instanceId));
                 } catch (NumberFormatException e) {
                     System.err.println("instance-id must be a number");
                 }
             } else {
-                admin.source().restartSource(tenant, namespace, sourceName);
+                admin.sources().restartSource(tenant, namespace, sourceName);
             }
             System.out.println("Restarted successfully");
         }
@@ -561,12 +561,12 @@ public class CmdSources extends CmdBase {
         void runCmd() throws Exception {
             if (isNotBlank(instanceId)) {
                 try {
-                    admin.source().stopSource(tenant, namespace, sourceName, 
Integer.parseInt(instanceId));
+                    admin.sources().stopSource(tenant, namespace, sourceName, 
Integer.parseInt(instanceId));
                 } catch (NumberFormatException e) {
                     System.err.println("instance-id must be a number");
                 }
             } else {
-                admin.source().stopSource(tenant, namespace, sourceName);
+                admin.sources().stopSource(tenant, namespace, sourceName);
             }
             System.out.println("Stopped successfully");
         }
@@ -582,12 +582,12 @@ public class CmdSources extends CmdBase {
         void runCmd() throws Exception {
             if (isNotBlank(instanceId)) {
                 try {
-                    admin.source().startSource(tenant, namespace, sourceName, 
Integer.parseInt(instanceId));
+                    admin.sources().startSource(tenant, namespace, sourceName, 
Integer.parseInt(instanceId));
                 } catch (NumberFormatException e) {
                     System.err.println("instance-id must be a number");
                 }
             } else {
-                admin.source().startSource(tenant, namespace, sourceName);
+                admin.sources().startSource(tenant, namespace, sourceName);
             }
             System.out.println("Started successfully");
         }
@@ -597,7 +597,7 @@ public class CmdSources extends CmdBase {
     public class ListBuiltInSources extends BaseCommand {
         @Override
         void runCmd() throws Exception {
-            admin.source().getBuiltInSources().stream().filter(x -> 
!StringUtils.isEmpty(x.getSourceClass()))
+            admin.sources().getBuiltInSources().stream().filter(x -> 
!StringUtils.isEmpty(x.getSourceClass()))
                     .forEach(connector -> {
                         System.out.println(connector.getName());
                         
System.out.println(WordUtils.wrap(connector.getDescription(), 80));
diff --git 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/PulsarAdminTool.java
 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/PulsarAdminTool.java
index 705a386..c3a7fd9 100644
--- 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/PulsarAdminTool.java
+++ 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/PulsarAdminTool.java
@@ -108,6 +108,11 @@ public class PulsarAdminTool {
         commandMap.put("resource-quotas", CmdResourceQuotas.class);
         commandMap.put("functions", CmdFunctions.class);
         commandMap.put("functions-worker", CmdFunctionWorker.class);
+        commandMap.put("sources", CmdSources.class);
+        commandMap.put("sinks", CmdSinks.class);
+
+        // To remain backwards compatibility for "source" and "sink" commands
+        // TODO eventually remove this
         commandMap.put("source", CmdSources.class);
         commandMap.put("sink", CmdSinks.class);
     }
@@ -119,11 +124,16 @@ public class PulsarAdminTool {
             PulsarAdmin admin = adminFactory.apply(adminBuilder);
             for (Map.Entry<String, Class<?>> c : commandMap.entrySet()) {
                 if (admin != null) {
-                    // Other mode, all components are initialized.
-                    jcommander.addCommand(c.getKey(), 
c.getValue().getConstructor(PulsarAdmin.class).newInstance(admin));
-                } else if (c.getKey().equals("functions") || 
c.getKey().equals("source") || c.getKey().equals("sink")) {
-                    // In mode localrun, only some components are initialized, 
such as source, sink and functions
-                    jcommander.addCommand(c.getKey(), 
c.getValue().getConstructor(PulsarAdmin.class).newInstance(admin));
+                    // To remain backwards compatibility for "source" and 
"sink" commands
+                    // TODO eventually remove this
+                    if (c.getKey().equals("sources") || 
c.getKey().equals("source")) {
+                        jcommander.addCommand("sources", 
c.getValue().getConstructor(PulsarAdmin.class).newInstance(admin), "source");
+                    } else if (c.getKey().equals("sinks") || 
c.getKey().equals("sink")) {
+                        jcommander.addCommand("sinks", 
c.getValue().getConstructor(PulsarAdmin.class).newInstance(admin), "sink");
+                    } else {
+                        // Other mode, all components are initialized.
+                        jcommander.addCommand(c.getKey(), 
c.getValue().getConstructor(PulsarAdmin.class).newInstance(admin));
+                    }
                 }
             }
         } catch (Exception e) {
@@ -187,6 +197,15 @@ public class PulsarAdminTool {
         } else {
             setupCommands(adminFactory);
             String cmd = args[cmdPos];
+
+            // To remain backwards compatibility for "source" and "sink" 
commands
+            // TODO eventually remove this
+            if (cmd.equals("source")) {
+                cmd = "sources";
+            } else if (cmd.equals("sink")) {
+                cmd = "sinks";
+            }
+
             JCommander obj = jcommander.getCommands().get(cmd);
             CmdBase cmdObj = (CmdBase) obj.getObjects().get(0);
 
diff --git 
a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
 
b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
index 159edaf..2a48586 100644
--- 
a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
+++ 
b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSinks.java
@@ -38,7 +38,7 @@ import java.util.*;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.pulsar.admin.cli.utils.CmdUtils;
-import org.apache.pulsar.client.admin.Sink;
+import org.apache.pulsar.client.admin.Sinks;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.Resources;
@@ -94,7 +94,7 @@ public class TestCmdSinks {
     private static final String SINK_CONFIG_STRING = "{\"created_at\":\"Mon 
Jul 02 00:33:15 0000 2018\"}";
 
     private PulsarAdmin pulsarAdmin;
-    private Sink sink;
+    private Sinks sink;
     private CmdSinks cmdSinks;
     private CmdSinks.CreateSink createSink;
     private CmdSinks.UpdateSink updateSink;
@@ -105,8 +105,8 @@ public class TestCmdSinks {
     public void setup() throws Exception {
 
         pulsarAdmin = mock(PulsarAdmin.class);
-        sink = mock(Sink.class);
-        when(pulsarAdmin.sink()).thenReturn(sink);
+        sink = mock(Sinks.class);
+        when(pulsarAdmin.sinks()).thenReturn(sink);
 
         cmdSinks = spy(new CmdSinks(pulsarAdmin));
         createSink = spy(cmdSinks.getCreateSink());
diff --git 
a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java
 
b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java
index 19c6c59..da6e99f 100644
--- 
a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java
+++ 
b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdSources.java
@@ -36,7 +36,7 @@ import java.nio.file.Files;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.pulsar.admin.cli.utils.CmdUtils;
-import org.apache.pulsar.client.admin.Source;
+import org.apache.pulsar.client.admin.Sources;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.Resources;
@@ -78,7 +78,7 @@ public class TestCmdSources {
     private static final String SINK_CONFIG_STRING = "{\"created_at\":\"Mon 
Jul 02 00:33:15 +0000 2018\"}";
 
     private PulsarAdmin pulsarAdmin;
-    private Source source;
+    private Sources source;
     private CmdSources CmdSources;
     private CmdSources.CreateSource createSource;
     private CmdSources.UpdateSource updateSource;
@@ -89,8 +89,8 @@ public class TestCmdSources {
     public void setup() throws Exception {
 
         pulsarAdmin = mock(PulsarAdmin.class);
-        source = mock(Source.class);
-        when(pulsarAdmin.source()).thenReturn(source);
+        source = mock(Sources.class);
+        when(pulsarAdmin.sources()).thenReturn(source);
 
         CmdSources = spy(new CmdSources(pulsarAdmin));
         createSource = spy(CmdSources.getCreateSource());
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
index 4a53d3e..b2f3339 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/Resources.java
@@ -23,12 +23,14 @@ import java.util.HashSet;
 import java.util.Set;
 
 import org.apache.pulsar.functions.worker.rest.api.FunctionsMetricsResource;
-import org.apache.pulsar.functions.worker.rest.api.v2.FunctionApiV2Resource;
+import org.apache.pulsar.functions.worker.rest.api.v2.FunctionsApiV2Resource;
 import org.apache.pulsar.functions.worker.rest.api.v2.WorkerApiV2Resource;
 import org.apache.pulsar.functions.worker.rest.api.v2.WorkerStatsApiV2Resource;
-import org.apache.pulsar.functions.worker.rest.api.v3.FunctionApiV3Resource;
+import org.apache.pulsar.functions.worker.rest.api.v3.FunctionsApiV3Resource;
 import org.apache.pulsar.functions.worker.rest.api.v3.SinkApiV3Resource;
+import org.apache.pulsar.functions.worker.rest.api.v3.SinksApiV3Resource;
 import org.apache.pulsar.functions.worker.rest.api.v3.SourceApiV3Resource;
+import org.apache.pulsar.functions.worker.rest.api.v3.SourcesApiV3Resource;
 import org.glassfish.jersey.media.multipart.MultiPartFeature;
 
 public final class Resources {
@@ -39,7 +41,7 @@ public final class Resources {
     public static Set<Class<?>> getApiV2Resources() {
         return new HashSet<>(
                 Arrays.asList(
-                        FunctionApiV2Resource.class,
+                        FunctionsApiV2Resource.class,
                         WorkerApiV2Resource.class,
                         WorkerStatsApiV2Resource.class,
                         MultiPartFeature.class
@@ -50,9 +52,11 @@ public final class Resources {
         return new HashSet<>(
                 Arrays.asList(
                         MultiPartFeature.class,
+                        SourcesApiV3Resource.class,
                         SourceApiV3Resource.class,
+                        SinksApiV3Resource.class,
                         SinkApiV3Resource.class,
-                        FunctionApiV3Resource.class
+                        FunctionsApiV3Resource.class
                 ));
     }
 
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinkImpl.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
similarity index 99%
rename from 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinkImpl.java
rename to 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
index e7b9ff1..fa07160 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinkImpl.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
@@ -44,7 +44,7 @@ import java.util.function.Supplier;
 import static 
org.apache.pulsar.functions.worker.rest.RestUtils.throwUnavailableException;
 
 @Slf4j
-public class SinkImpl extends ComponentImpl {
+public class SinksImpl extends ComponentImpl {
 
     private class GetSinkStatus extends GetStatus<SinkStatus, 
SinkStatus.SinkInstanceStatus.SinkInstanceStatusData> {
 
@@ -206,7 +206,7 @@ public class SinkImpl extends ComponentImpl {
         return exceptionInformation;
     }
 
-    public SinkImpl(Supplier<WorkerService> workerServiceSupplier) {
+    public SinksImpl(Supplier<WorkerService> workerServiceSupplier) {
         super(workerServiceSupplier, 
Function.FunctionDetails.ComponentType.SINK);
     }
 
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourceImpl.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
similarity index 99%
rename from 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourceImpl.java
rename to 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
index a2b59ef..d35724e 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourceImpl.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
@@ -44,7 +44,7 @@ import java.util.function.Supplier;
 import static 
org.apache.pulsar.functions.worker.rest.RestUtils.throwUnavailableException;
 
 @Slf4j
-public class SourceImpl extends ComponentImpl {
+public class SourcesImpl extends ComponentImpl {
     private class GetSourceStatus extends GetStatus<SourceStatus, 
SourceStatus.SourceInstanceStatus.SourceInstanceStatusData> {
 
         @Override
@@ -208,7 +208,7 @@ public class SourceImpl extends ComponentImpl {
         }
     }
 
-    public SourceImpl(Supplier<WorkerService> workerServiceSupplier) {
+    public SourcesImpl(Supplier<WorkerService> workerServiceSupplier) {
         super(workerServiceSupplier, 
Function.FunctionDetails.ComponentType.SOURCE);
     }
 
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java
similarity index 99%
rename from 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
rename to 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java
index c978cad..6486187 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2Resource.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java
@@ -47,11 +47,11 @@ import java.util.List;
 
 @Slf4j
 @Path("/functions")
-public class FunctionApiV2Resource extends FunctionApiResource {
+public class FunctionsApiV2Resource extends FunctionApiResource {
 
     protected final FunctionsImplV2 functions;
 
-    public FunctionApiV2Resource() {
+    public FunctionsApiV2Resource() {
         this.functions = new FunctionsImplV2(this);
     }
 
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3Resource.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
similarity index 99%
rename from 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3Resource.java
rename to 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
index 7c8136d..38f4c4b 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3Resource.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
@@ -50,11 +50,11 @@ import java.util.List;
 
 @Slf4j
 @Path("/functions")
-public class FunctionApiV3Resource extends FunctionApiResource {
+public class FunctionsApiV3Resource extends FunctionApiResource {
 
     protected final FunctionsImpl functions;
 
-    public FunctionApiV3Resource() {
+    public FunctionsApiV3Resource() {
         this.functions = new FunctionsImpl(this);
     }
 
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
index b12c381..3731b92 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
@@ -18,220 +18,20 @@
  */
 package org.apache.pulsar.functions.worker.rest.api.v3;
 
-import io.swagger.annotations.ApiOperation;
-import io.swagger.annotations.ApiResponse;
-import io.swagger.annotations.ApiResponses;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang.StringUtils;
-import org.apache.pulsar.common.functions.UpdateOptions;
-import org.apache.pulsar.common.io.ConnectorDefinition;
-import org.apache.pulsar.common.io.SinkConfig;
-import org.apache.pulsar.common.policies.data.SinkStatus;
-import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
-import org.apache.pulsar.functions.worker.rest.api.SinkImpl;
-import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
-import org.glassfish.jersey.media.multipart.FormDataParam;
+import io.swagger.annotations.Api;
 
-import javax.ws.rs.*;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.List;
 
-@Slf4j
+@Api(value = "/sink", description = "Sink admin apis", tags = "sink")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
 @Path("/sink")
-public class SinkApiV3Resource extends FunctionApiResource {
-
-    protected final SinkImpl sink;
-
-    public SinkApiV3Resource() {
-        this.sink = new SinkImpl(this);
-    }
-
-    @POST
-    @Path("/{tenant}/{namespace}/{sinkName}")
-    @Consumes(MediaType.MULTIPART_FORM_DATA)
-    public void registerSink(final @PathParam("tenant") String tenant,
-                             final @PathParam("namespace") String namespace,
-                             final @PathParam("sinkName") String sinkName,
-                             final @FormDataParam("data") InputStream 
uploadedInputStream,
-                             final @FormDataParam("data") 
FormDataContentDisposition fileDetail,
-                             final @FormDataParam("url") String functionPkgUrl,
-                             final @FormDataParam("sinkConfig") String 
sinkConfigJson) {
-
-        sink.registerFunction(tenant, namespace, sinkName, 
uploadedInputStream, fileDetail,
-                functionPkgUrl, sinkConfigJson, clientAppId(), 
clientAuthData());
-    }
-
-    @PUT
-    @Path("/{tenant}/{namespace}/{sinkName}")
-    @Consumes(MediaType.MULTIPART_FORM_DATA)
-    public void updateSink(final @PathParam("tenant") String tenant,
-                           final @PathParam("namespace") String namespace,
-                           final @PathParam("sinkName") String sinkName,
-                           final @FormDataParam("data") InputStream 
uploadedInputStream,
-                           final @FormDataParam("data") 
FormDataContentDisposition fileDetail,
-                           final @FormDataParam("url") String functionPkgUrl,
-                           final @FormDataParam("sinkConfig") String 
sinkConfigJson,
-                           final @FormDataParam("updateOptions") UpdateOptions 
updateOptions) {
-
-        sink.updateFunction(tenant, namespace, sinkName, uploadedInputStream, 
fileDetail,
-                functionPkgUrl, sinkConfigJson, clientAppId(), 
clientAuthData(), updateOptions);
-    }
-
-    @DELETE
-    @Path("/{tenant}/{namespace}/{sinkName}")
-    public void deregisterSink(final @PathParam("tenant") String tenant,
-                               final @PathParam("namespace") String namespace,
-                               final @PathParam("sinkName") String sinkName) {
-        sink.deregisterFunction(tenant, namespace, sinkName, clientAppId(), 
clientAuthData());
-    }
-
-    @GET
-    @Path("/{tenant}/{namespace}/{sinkName}")
-    public SinkConfig getSinkInfo(final @PathParam("tenant") String tenant,
-                                  final @PathParam("namespace") String 
namespace,
-                                  final @PathParam("sinkName") String sinkName)
-            throws IOException {
-        return sink.getSinkInfo(tenant, namespace, sinkName);
-    }
-
-    @GET
-    @ApiOperation(
-            value = "Displays the status of a Pulsar Sink instance",
-            response = 
SinkStatus.SinkInstanceStatus.SinkInstanceStatusData.class
-    )
-    @ApiResponses(value = {
-            @ApiResponse(code = 400, message = "Invalid request"),
-            @ApiResponse(code = 403, message = "The requester doesn't have 
admin permissions"),
-            @ApiResponse(code = 404, message = "The sink doesn't exist")
-    })
-    @Produces(MediaType.APPLICATION_JSON)
-    @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/status")
-    public SinkStatus.SinkInstanceStatus.SinkInstanceStatusData 
getSinkInstanceStatus(
-            final @PathParam("tenant") String tenant,
-            final @PathParam("namespace") String namespace,
-            final @PathParam("sinkName") String sinkName,
-            final @PathParam("instanceId") String instanceId) throws 
IOException {
-        return sink.getSinkInstanceStatus(tenant, namespace, sinkName, 
instanceId, uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @ApiOperation(
-            value = "Displays the status of a Pulsar Sink running in cluster 
mode",
-            response = SinkStatus.class
-    )
-    @ApiResponses(value = {
-            @ApiResponse(code = 400, message = "Invalid request"),
-            @ApiResponse(code = 403, message = "The requester doesn't have 
admin permissions"),
-            @ApiResponse(code = 404, message = "The sink doesn't exist")
-    })
-    @Produces(MediaType.APPLICATION_JSON)
-    @Path("/{tenant}/{namespace}/{sinkName}/status")
-    public SinkStatus getSinkStatus(final @PathParam("tenant") String tenant,
-                                    final @PathParam("namespace") String 
namespace,
-                                    final @PathParam("sinkName") String 
sinkName) throws IOException {
-        return sink.getSinkStatus(tenant, namespace, sinkName, 
uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @Path("/{tenant}/{namespace}")
-    public List<String> listSink(final @PathParam("tenant") String tenant,
-                                 final @PathParam("namespace") String 
namespace) {
-        return sink.listFunctions(tenant, namespace, clientAppId(), 
clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Restart sink instance", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-    @ApiResponse(code = 404, message = "The function does not exist"),
-    @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/restart")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void restartSink(final @PathParam("tenant") String tenant,
-                            final @PathParam("namespace") String namespace,
-                            final @PathParam("sinkName") String sinkName,
-                            final @PathParam("instanceId") String instanceId) {
-        sink.restartFunctionInstance(tenant, namespace, sinkName, instanceId, 
this.uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Restart all sink instances", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-    @ApiResponse(code = 404, message = "The function does not exist"), 
@ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sinkName}/restart")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void restartSink(final @PathParam("tenant") String tenant,
-                            final @PathParam("namespace") String namespace,
-                            final @PathParam("sinkName") String sinkName) {
-        sink.restartFunctionInstances(tenant, namespace, sinkName, 
clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Stop sink instance", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/stop")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void stopSink(final @PathParam("tenant") String tenant,
-                         final @PathParam("namespace") String namespace,
-                         final @PathParam("sinkName") String sinkName,
-                         final @PathParam("instanceId") String instanceId) {
-        sink.stopFunctionInstance(tenant, namespace, sinkName, instanceId, 
this.uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Stop all sink instances", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-    @ApiResponse(code = 404, message = "The function does not exist"),
-    @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sinkName}/stop")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void stopSink(final @PathParam("tenant") String tenant,
-                         final @PathParam("namespace") String namespace,
-                         final @PathParam("sinkName") String sinkName) {
-        sink.stopFunctionInstances(tenant, namespace, sinkName, clientAppId(), 
clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Start sink instance", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sinkName}/{instanceId}/start")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void startSink(final @PathParam("tenant") String tenant,
-                          final @PathParam("namespace") String namespace,
-                          final @PathParam("sinkName") String sinkName,
-                          final @PathParam("instanceId") String instanceId) {
-        sink.startFunctionInstance(tenant, namespace, sinkName, instanceId, 
this.uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Start all sink instances", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sinkName}/start")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void startSink(final @PathParam("tenant") String tenant,
-                          final @PathParam("namespace") String namespace,
-                          final @PathParam("sinkName") String sinkName) {
-        sink.startFunctionInstances(tenant, namespace, sinkName, 
clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @Path("/builtinsinks")
-    public List<ConnectorDefinition> getSinkList() {
-        List<ConnectorDefinition> connectorDefinitions = 
sink.getListOfConnectors();
-        List<ConnectorDefinition> retVal = new ArrayList<>();
-        for (ConnectorDefinition connectorDefinition : connectorDefinitions) {
-            if (!StringUtils.isEmpty(connectorDefinition.getSinkClass())) {
-                retVal.add(connectorDefinition);
-            }
-        }
-        return retVal;
-    }
+@Deprecated
+/**
+ * @deprecated in favor of {@link SinksApiV3Resource}
+ */
+public class SinkApiV3Resource extends SinksApiV3Resource {
 }
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinksApiV3Resource.java
similarity index 96%
copy from 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
copy to 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinksApiV3Resource.java
index b12c381..e699544 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3Resource.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SinksApiV3Resource.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.functions.worker.rest.api.v3;
 
+import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiResponse;
 import io.swagger.annotations.ApiResponses;
@@ -28,7 +29,7 @@ import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.common.io.SinkConfig;
 import org.apache.pulsar.common.policies.data.SinkStatus;
 import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
-import org.apache.pulsar.functions.worker.rest.api.SinkImpl;
+import org.apache.pulsar.functions.worker.rest.api.SinksImpl;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.glassfish.jersey.media.multipart.FormDataParam;
 
@@ -40,13 +41,16 @@ import java.util.ArrayList;
 import java.util.List;
 
 @Slf4j
-@Path("/sink")
-public class SinkApiV3Resource extends FunctionApiResource {
+@Api(value = "/sinks", description = "Sinks admin apis", tags = "sinks")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+@Path("/sinks")
+public class SinksApiV3Resource extends FunctionApiResource {
 
-    protected final SinkImpl sink;
+    protected final SinksImpl sink;
 
-    public SinkApiV3Resource() {
-        this.sink = new SinkImpl(this);
+    public SinksApiV3Resource() {
+        this.sink = new SinksImpl(this);
     }
 
     @POST
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
index 96d2082..65d68f9 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
@@ -18,224 +18,20 @@
  */
 package org.apache.pulsar.functions.worker.rest.api.v3;
 
-import io.swagger.annotations.ApiOperation;
-import io.swagger.annotations.ApiResponse;
-import io.swagger.annotations.ApiResponses;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang.StringUtils;
-import org.apache.pulsar.common.functions.UpdateOptions;
-import org.apache.pulsar.common.io.ConnectorDefinition;
-import org.apache.pulsar.common.io.SourceConfig;
-import org.apache.pulsar.common.policies.data.SourceStatus;
-import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
-import org.apache.pulsar.functions.worker.rest.api.SourceImpl;
-import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
-import org.glassfish.jersey.media.multipart.FormDataParam;
+import io.swagger.annotations.Api;
 
-import javax.ws.rs.*;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.List;
 
-@Slf4j
 @Path("/source")
-public class SourceApiV3Resource extends FunctionApiResource {
-
-    protected final SourceImpl source;
-
-    public SourceApiV3Resource() {
-        this.source = new SourceImpl(this);
-    }
-
-    @POST
-    @Path("/{tenant}/{namespace}/{sourceName}")
-    @Consumes(MediaType.MULTIPART_FORM_DATA)
-    public void registerSource(final @PathParam("tenant") String tenant,
-                                   final @PathParam("namespace") String 
namespace,
-                                   final @PathParam("sourceName") String 
sourceName,
-                                   final @FormDataParam("data") InputStream 
uploadedInputStream,
-                                   final @FormDataParam("data") 
FormDataContentDisposition fileDetail,
-                                   final @FormDataParam("url") String 
functionPkgUrl,
-                                   final @FormDataParam("sourceConfig") String 
sourceConfigJson) {
-
-        source.registerFunction(tenant, namespace, sourceName, 
uploadedInputStream, fileDetail,
-                functionPkgUrl, sourceConfigJson, clientAppId(), 
clientAuthData());
-
-    }
-
-    @PUT
-    @Path("/{tenant}/{namespace}/{sourceName}")
-    @Consumes(MediaType.MULTIPART_FORM_DATA)
-    public void updateSource(final @PathParam("tenant") String tenant,
-                             final @PathParam("namespace") String namespace,
-                             final @PathParam("sourceName") String sourceName,
-                             final @FormDataParam("data") InputStream 
uploadedInputStream,
-                             final @FormDataParam("data") 
FormDataContentDisposition fileDetail,
-                             final @FormDataParam("url") String functionPkgUrl,
-                             final @FormDataParam("sourceConfig") String 
sourceConfigJson,
-                             final @FormDataParam("updateOptions") 
UpdateOptions updateOptions) {
-
-        source.updateFunction(tenant, namespace, sourceName, 
uploadedInputStream, fileDetail,
-                functionPkgUrl, sourceConfigJson, clientAppId(), 
clientAuthData(), updateOptions);
-    }
-
-
-    @DELETE
-    @Path("/{tenant}/{namespace}/{sourceName}")
-    public void deregisterSource(final @PathParam("tenant") String tenant,
-                                 final @PathParam("namespace") String 
namespace,
-                                 final @PathParam("sourceName") String 
sourceName) {
-        source.deregisterFunction(tenant, namespace, sourceName, 
clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @Path("/{tenant}/{namespace}/{sourceName}")
-    public SourceConfig getSourceInfo(final @PathParam("tenant") String tenant,
-                                      final @PathParam("namespace") String 
namespace,
-                                      final @PathParam("sourceName") String 
sourceName)
-            throws IOException {
-        return source.getSourceInfo(tenant, namespace, sourceName);
-    }
-
-    @GET
-    @ApiOperation(
-            value = "Displays the status of a Pulsar Source instance",
-            response = 
SourceStatus.SourceInstanceStatus.SourceInstanceStatusData.class
-    )
-    @ApiResponses(value = {
-            @ApiResponse(code = 400, message = "Invalid request"),
-            @ApiResponse(code = 403, message = "The requester doesn't have 
admin permissions"),
-            @ApiResponse(code = 404, message = "The source doesn't exist")
-    })
-    @Produces(MediaType.APPLICATION_JSON)
-    @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/status")
-    public SourceStatus.SourceInstanceStatus.SourceInstanceStatusData 
getSourceInstanceStatus(
-            final @PathParam("tenant") String tenant,
-            final @PathParam("namespace") String namespace,
-            final @PathParam("sourceName") String sourceName,
-            final @PathParam("instanceId") String instanceId) throws 
IOException {
-        return source.getSourceInstanceStatus(
-            tenant, namespace, sourceName, instanceId, uri.getRequestUri(), 
clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @ApiOperation(
-            value = "Displays the status of a Pulsar Source running in cluster 
mode",
-            response = SourceStatus.class
-    )
-    @ApiResponses(value = {
-            @ApiResponse(code = 400, message = "Invalid request"),
-            @ApiResponse(code = 403, message = "The requester doesn't have 
admin permissions"),
-            @ApiResponse(code = 404, message = "The source doesn't exist")
-    })
-    @Produces(MediaType.APPLICATION_JSON)
-    @Path("/{tenant}/{namespace}/{sourceName}/status")
-    public SourceStatus getSourceStatus(final @PathParam("tenant") String 
tenant,
-                                    final @PathParam("namespace") String 
namespace,
-                                    final @PathParam("sourceName") String 
sourceName) throws IOException {
-        return source.getSourceStatus(tenant, namespace, sourceName, 
uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @Path("/{tenant}/{namespace}")
-    public List<String> listSources(final @PathParam("tenant") String tenant,
-                                    final @PathParam("namespace") String 
namespace) {
-        return source.listFunctions(tenant, namespace, clientAppId(), 
clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Restart source instance", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/restart")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void restartSource(final @PathParam("tenant") String tenant,
-                              final @PathParam("namespace") String namespace,
-                              final @PathParam("sourceName") String sourceName,
-                              final @PathParam("instanceId") String 
instanceId) {
-        source.restartFunctionInstance(tenant, namespace, sourceName, 
instanceId, this.uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Restart all source instances", response = 
Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-    @ApiResponse(code = 404, message = "The function does not exist"),
-    @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sourceName}/restart")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void restartSource(final @PathParam("tenant") String tenant,
-                              final @PathParam("namespace") String namespace,
-                              final @PathParam("sourceName") String 
sourceName) {
-        source.restartFunctionInstances(tenant, namespace, sourceName, 
clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Stop source instance", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/stop")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void stopSource(final @PathParam("tenant") String tenant,
-                           final @PathParam("namespace") String namespace,
-                           final @PathParam("sourceName") String sourceName,
-                           final @PathParam("instanceId") String instanceId) {
-        source.stopFunctionInstance(tenant, namespace, sourceName, instanceId, 
this.uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Stop all source instances", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sourceName}/stop")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void stopSource(final @PathParam("tenant") String tenant,
-                           final @PathParam("namespace") String namespace,
-                           final @PathParam("sourceName") String sourceName) {
-        source.stopFunctionInstances(tenant, namespace, sourceName, 
clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Start source instance", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sourceName}/{instanceId}/start")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void startSource(final @PathParam("tenant") String tenant,
-                            final @PathParam("namespace") String namespace,
-                            final @PathParam("sourceName") String sourceName,
-                            final @PathParam("instanceId") String instanceId) {
-        source.startFunctionInstance(tenant, namespace, sourceName, 
instanceId, this.uri.getRequestUri(), clientAppId(), clientAuthData());
-    }
-
-    @POST
-    @ApiOperation(value = "Start all source instances", response = Void.class)
-    @ApiResponses(value = { @ApiResponse(code = 400, message = "Invalid 
request"),
-            @ApiResponse(code = 404, message = "The function does not exist"),
-            @ApiResponse(code = 500, message = "Internal server error") })
-    @Path("/{tenant}/{namespace}/{sourceName}/start")
-    @Consumes(MediaType.APPLICATION_JSON)
-    public void startSource(final @PathParam("tenant") String tenant,
-                            final @PathParam("namespace") String namespace,
-                            final @PathParam("sourceName") String sourceName) {
-        source.startFunctionInstances(tenant, namespace, sourceName, 
clientAppId(), clientAuthData());
-    }
-
-    @GET
-    @Path("/builtinsources")
-    public List<ConnectorDefinition> getSourceList() {
-        List<ConnectorDefinition> connectorDefinitions = 
source.getListOfConnectors();
-        List<ConnectorDefinition> retval = new ArrayList<>();
-        for (ConnectorDefinition connectorDefinition : connectorDefinitions) {
-            if (!StringUtils.isEmpty(connectorDefinition.getSourceClass())) {
-                retval.add(connectorDefinition);
-            }
-        }
-        return retval;
-    }
+@Api(value = "/source", description = "Source admin apis", tags = "source")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+@Deprecated
+/**
+ * @deprecated in favor of {@link SourcesApiV3Resource}
+ */
+public class SourceApiV3Resource extends SourcesApiV3Resource {
 }
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourcesApiV3Resource.java
similarity index 92%
copy from 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
copy to 
pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourcesApiV3Resource.java
index 96d2082..de6df1a 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3Resource.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/SourcesApiV3Resource.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.functions.worker.rest.api.v3;
 
+import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiResponse;
 import io.swagger.annotations.ApiResponses;
@@ -28,7 +29,7 @@ import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.common.policies.data.SourceStatus;
 import org.apache.pulsar.functions.worker.rest.FunctionApiResource;
-import org.apache.pulsar.functions.worker.rest.api.SourceImpl;
+import org.apache.pulsar.functions.worker.rest.api.SourcesImpl;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.glassfish.jersey.media.multipart.FormDataParam;
 
@@ -40,13 +41,16 @@ import java.util.ArrayList;
 import java.util.List;
 
 @Slf4j
-@Path("/source")
-public class SourceApiV3Resource extends FunctionApiResource {
+@Api(value = "/sources", description = "Sources admin apis", tags = "sources")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+@Path("/sources")
+public class SourcesApiV3Resource extends FunctionApiResource {
 
-    protected final SourceImpl source;
+    protected final SourcesImpl source;
 
-    public SourceApiV3Resource() {
-        this.source = new SourceImpl(this);
+    public SourcesApiV3Resource() {
+        this.source = new SourcesImpl(this);
     }
 
     @POST
@@ -91,6 +95,7 @@ public class SourceApiV3Resource extends FunctionApiResource {
     }
 
     @GET
+    @Produces(MediaType.APPLICATION_JSON)
     @Path("/{tenant}/{namespace}/{sourceName}")
     public SourceConfig getSourceInfo(final @PathParam("tenant") String tenant,
                                       final @PathParam("namespace") String 
namespace,
@@ -139,6 +144,7 @@ public class SourceApiV3Resource extends 
FunctionApiResource {
     }
 
     @GET
+    @Produces(MediaType.APPLICATION_JSON)
     @Path("/{tenant}/{namespace}")
     public List<String> listSources(final @PathParam("tenant") String tenant,
                                     final @PathParam("namespace") String 
namespace) {
@@ -227,6 +233,16 @@ public class SourceApiV3Resource extends 
FunctionApiResource {
     }
 
     @GET
+    @ApiOperation(
+            value = "Fetches a list of supported Pulsar IO source 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")
+    })
+    @Produces(MediaType.APPLICATION_JSON)
     @Path("/builtinsources")
     public List<ConnectorDefinition> getSourceList() {
         List<ConnectorDefinition> connectorDefinitions = 
source.getListOfConnectors();
diff --git 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
index b05b0cd..4405c21 100644
--- 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
+++ 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionApiV2ResourceTest.java
@@ -92,7 +92,7 @@ import static 
org.powermock.api.mockito.PowerMockito.mockStatic;
 import static org.testng.Assert.assertEquals;
 
 /**
- * Unit test of {@link FunctionApiV2Resource}.
+ * Unit test of {@link FunctionsApiV2Resource}.
  */
 @PrepareForTest({WorkerUtils.class, InstanceUtils.class})
 @PowerMockIgnore({ "javax.management.*", "javax.ws.*", 
"org.apache.logging.log4j.*", "org.apache.pulsar.functions.api.*" })
diff --git 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java
 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java
index c364906..434cfc2 100644
--- 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java
+++ 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionApiV3ResourceTest.java
@@ -52,7 +52,7 @@ import org.apache.pulsar.functions.worker.WorkerUtils;
 import org.apache.pulsar.functions.worker.request.RequestResult;
 import org.apache.pulsar.functions.worker.rest.RestException;
 import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
-import org.apache.pulsar.functions.worker.rest.api.v2.FunctionApiV2Resource;
+import org.apache.pulsar.functions.worker.rest.api.v2.FunctionsApiV2Resource;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
@@ -76,7 +76,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
-import java.util.function.Supplier;
 
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
@@ -90,7 +89,7 @@ import static 
org.powermock.api.mockito.PowerMockito.mockStatic;
 import static org.testng.Assert.assertEquals;
 
 /**
- * Unit test of {@link FunctionApiV2Resource}.
+ * Unit test of {@link FunctionsApiV2Resource}.
  */
 @PrepareForTest({WorkerUtils.class, InstanceUtils.class})
 @PowerMockIgnore({ "javax.management.*", "javax.ws.*", 
"org.apache.logging.log4j.*", "org.apache.pulsar.functions.api.*" })
diff --git 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
index 2988e3a..3b3548a 100644
--- 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
+++ 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SinkApiV3ResourceTest.java
@@ -49,7 +49,7 @@ import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.WorkerUtils;
 import org.apache.pulsar.functions.worker.request.RequestResult;
 import org.apache.pulsar.functions.worker.rest.RestException;
-import org.apache.pulsar.functions.worker.rest.api.SinkImpl;
+import org.apache.pulsar.functions.worker.rest.api.SinksImpl;
 import org.apache.pulsar.io.cassandra.CassandraStringSink;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.powermock.api.mockito.PowerMockito;
@@ -88,7 +88,7 @@ import static 
org.powermock.api.mockito.PowerMockito.mockStatic;
 import static org.testng.Assert.assertEquals;
 
 /**
- * Unit test of {@link SinkApiV3Resource}.
+ * Unit test of {@link SinksApiV3Resource}.
  */
 @PrepareForTest({WorkerUtils.class, SinkConfigUtils.class, 
ConnectorUtils.class, FunctionCommon.class, InstanceUtils.class})
 @PowerMockIgnore({ "javax.management.*", "javax.ws.*", 
"org.apache.logging.log4j.*", "org.apache.pulsar.io.*", "java.io.*" })
@@ -125,7 +125,7 @@ public class SinkApiV3ResourceTest {
     private FunctionRuntimeManager mockedFunctionRunTimeManager;
     private RuntimeFactory mockedRuntimeFactory;
     private Namespace mockedNamespace;
-    private SinkImpl resource;
+    private SinksImpl resource;
     private InputStream mockedInputStream;
     private FormDataContentDisposition mockedFormData;
     private FunctionMetaData mockedFunctionMetaData;
@@ -175,7 +175,7 @@ public class SinkApiV3ResourceTest {
             .setPulsarServiceUrl("pulsar://localhost:6650/");
         when(mockedWorkerService.getWorkerConfig()).thenReturn(workerConfig);
 
-        this.resource = spy(new SinkImpl(() -> mockedWorkerService));
+        this.resource = spy(new SinksImpl(() -> mockedWorkerService));
         mockStatic(InstanceUtils.class);
         
PowerMockito.when(InstanceUtils.calculateSubjectType(any())).thenReturn(FunctionDetails.ComponentType.SINK);
     }
diff --git 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
index 0381af7..cd55369 100644
--- 
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
+++ 
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v3/SourceApiV3ResourceTest.java
@@ -52,7 +52,7 @@ import org.apache.pulsar.functions.worker.WorkerService;
 import org.apache.pulsar.functions.worker.WorkerUtils;
 import org.apache.pulsar.functions.worker.request.RequestResult;
 import org.apache.pulsar.functions.worker.rest.RestException;
-import org.apache.pulsar.functions.worker.rest.api.SourceImpl;
+import org.apache.pulsar.functions.worker.rest.api.SourcesImpl;
 import org.apache.pulsar.io.twitter.TwitterFireHose;
 import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
 import org.powermock.api.mockito.PowerMockito;
@@ -87,7 +87,7 @@ import static 
org.powermock.api.mockito.PowerMockito.mockStatic;
 import static org.testng.Assert.assertEquals;
 
 /**
- * Unit test of {@link SourceApiV3Resource}.
+ * Unit test of {@link SourcesApiV3Resource}.
  */
 @PrepareForTest({WorkerUtils.class, ConnectorUtils.class, 
FunctionCommon.class, InstanceUtils.class})
 @PowerMockIgnore({ "javax.management.*", "javax.ws.*", 
"org.apache.logging.log4j.*", "org.apache.pulsar.io.*" })
@@ -121,7 +121,7 @@ public class SourceApiV3ResourceTest {
     private FunctionRuntimeManager mockedFunctionRunTimeManager;
     private RuntimeFactory mockedRuntimeFactory;
     private Namespace mockedNamespace;
-    private SourceImpl resource;
+    private SourcesImpl resource;
     private InputStream mockedInputStream;
     private FormDataContentDisposition mockedFormData;
     private FunctionMetaData mockedFunctionMetaData;
@@ -170,7 +170,7 @@ public class SourceApiV3ResourceTest {
             .setPulsarServiceUrl("pulsar://localhost:6650/");
         when(mockedWorkerService.getWorkerConfig()).thenReturn(workerConfig);
 
-        this.resource = spy(new SourceImpl(() -> mockedWorkerService));
+        this.resource = spy(new SourcesImpl(() -> mockedWorkerService));
         mockStatic(InstanceUtils.class);
         
PowerMockito.when(InstanceUtils.calculateSubjectType(any())).thenReturn(FunctionDetails.ComponentType.SOURCE);
     }