renaming sinks and sources api to be consistent with the rest of Pul…

…sar (#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

@@ -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>

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinkBase.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SinksBase.java

@@ -28,7 +28,7 @@
 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.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

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourceBase.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SourcesBase.java

@@ -28,7 +28,7 @@
 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.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 SourceStatus getSourceStatus(final @PathParam("tenant") String tenant,
             @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 void startSource(final @PathParam("tenant") String tenant,
             @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();

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 @@
 @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 {
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sinks.java

@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.admin.v3;
+
+import io.swagger.annotations.Api;
+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("/sinks")
+@Api(value = "/sinks", description = "Sinks admin apis", tags = "sinks")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class Sinks extends SinksBase {
+}

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 @@
 @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 {
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v3/Sources.java

@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.admin.v3;
+
+import io.swagger.annotations.Api;
+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("/sources")
+@Api(value = "/sources", description = "Sources admin apis", tags = "sources")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class Sources extends SourcesBase {
+}

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

@@ -40,8 +40,8 @@
 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 PulsarAdmin(String serviceUrl,
         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 Functions functions() {
     }
 
     /**
-     *
-     * @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() {