srkukarni closed pull request #2947: Check for presence of namespace/tenant
before creating function
URL: https://github.com/apache/pulsar/pull/2947
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
index c0d81e5c80..fb081e5bf7 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionTlsTest.java
@@ -29,16 +29,14 @@
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URL;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
import org.apache.bookkeeper.test.PortManager;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.ServiceConfigurationUtils;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.broker.authentication.AuthenticationService;
+import org.apache.pulsar.client.admin.Namespaces;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.client.api.Authentication;
@@ -78,6 +76,7 @@
String workerId;
WorkerServer workerServer;
PulsarAdmin functionAdmin;
+ private List<String> namespaceList = new LinkedList<>();
private final int ZOOKEEPER_PORT = PortManager.nextFreePort();
private final int workerServicePort = PortManager.nextFreePort();
private final int workerServicePortTls = PortManager.nextFreePort();
@@ -123,6 +122,9 @@ void setup(Method method) throws Exception {
Set<String> admins = Sets.newHashSet("superUser");
TenantInfo tenantInfo = new TenantInfo(admins, null);
when(tenants.getTenantInfo(any())).thenReturn(tenantInfo);
+ Namespaces namespaces = mock(Namespaces.class);
+ when(admin.namespaces()).thenReturn(namespaces);
+ when(namespaces.getNamespaces(any())).thenReturn(namespaceList);
// mock: once authentication passes, function should return response:
function already exist
FunctionMetaDataManager dataManager =
mock(FunctionMetaDataManager.class);
@@ -205,6 +207,7 @@ public void testAuthorization() throws Exception {
final String sinkTopic = "persistent://" + replNamespace + "/output";
final String functionName = "PulsarSink-test";
final String subscriptionName = "test-sub";
+ namespaceList.add(replNamespace);
String jarFilePathUrl = String.format("%s:%s",
org.apache.pulsar.common.functions.Utils.FILE,
PulsarSink.class.getProtectionDomain().getCodeSource().getLocation().getPath());
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index d14d4f696d..af8c050b2a 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -153,6 +153,46 @@ public Response registerFunction(final String tenant,
final String namespace, fi
return getUnavailableResponse();
}
+ if (tenant == null) {
+ return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData("Tenant is not provided")).build();
+ }
+ if (namespace == null) {
+ return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData("Namespace is not
provided")).build();
+ }
+ if (componentName == null) {
+ return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData(componentType + " Name is not
provided")).build();
+ }
+
+ try {
+ TenantInfo tenantInfo =
worker().getBrokerAdmin().tenants().getTenantInfo(tenant);
+ String qualifedNamespace = tenant + "/" + namespace;
+ if
(!worker().getBrokerAdmin().namespaces().getNamespaces(tenant).contains(qualifedNamespace))
{
+ log.error("{}/{}/{} Namespace {} does not exist", tenant,
namespace,
+ componentName, namespace);
+ return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData("Namespace does not
exist")).build();
+ }
+ } catch (PulsarAdminException.NotAuthorizedException e) {
+ log.error("{}/{}/{} Client [{}] is not admin and authorized to
operate {} on tenant", tenant, namespace,
+ componentName, clientRole, componentType);
+ return
Response.status(Status.UNAUTHORIZED).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData("client is not authorize to perform
operation")).build();
+
+ } catch (PulsarAdminException.NotFoundException e) {
+ log.error("{}/{}/{} Tenant {} does not exist", tenant, namespace,
+ componentName, tenant);
+ return
Response.status(Status.BAD_REQUEST).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData("Tenant does not exist")).build();
+ } catch (PulsarAdminException e) {
+ log.error("{}/{}/{} Issues getting tenant data", tenant, namespace,
+ componentName, e);
+ return
Response.status(Status.INTERNAL_SERVER_ERROR).type(MediaType.APPLICATION_JSON)
+ .entity(new ErrorData(e.getMessage())).build();
+ }
+
try {
if (!isAuthorizedRole(tenant, clientRole)) {
log.error("{}/{}/{} Client [{}] is not admin and authorized to
register {}", tenant, namespace,
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 c3995a83d9..e03d90d3f1 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
@@ -47,7 +47,12 @@
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.core.config.Configurator;
+import org.apache.pulsar.client.admin.Namespaces;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.common.policies.data.ErrorData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
@@ -109,6 +114,11 @@ public String process(String input, Context context) {
private static final int parallelism = 1;
private WorkerService mockedWorkerService;
+ private PulsarAdmin mockedPulsarAdmin;
+ private Tenants mockedTenants;
+ private Namespaces mockedNamespaces;
+ private TenantInfo mockedTenantInfo;
+ private List<String> namespaceList = new LinkedList<>();
private FunctionMetaDataManager mockedManager;
private FunctionRuntimeManager mockedFunctionRunTimeManager;
private RuntimeFactory mockedRuntimeFactory;
@@ -118,14 +128,19 @@ public String process(String input, Context context) {
private FormDataContentDisposition mockedFormData;
@BeforeMethod
- public void setup() {
+ public void setup() throws Exception {
this.mockedManager = mock(FunctionMetaDataManager.class);
this.mockedFunctionRunTimeManager = mock(FunctionRuntimeManager.class);
+ this.mockedTenantInfo = mock(TenantInfo.class);
this.mockedRuntimeFactory = mock(RuntimeFactory.class);
this.mockedInputStream = mock(InputStream.class);
this.mockedNamespace = mock(Namespace.class);
this.mockedFormData = mock(FormDataContentDisposition.class);
when(mockedFormData.getFileName()).thenReturn("test");
+ this.mockedPulsarAdmin = mock(PulsarAdmin.class);
+ this.mockedTenants = mock(Tenants.class);
+ this.mockedNamespaces = mock(Namespaces.class);
+ namespaceList.add(tenant + "/" + namespace);
this.mockedWorkerService = mock(WorkerService.class);
when(mockedWorkerService.getFunctionMetaDataManager()).thenReturn(mockedManager);
@@ -133,6 +148,11 @@ public void setup() {
when(mockedFunctionRunTimeManager.getRuntimeFactory()).thenReturn(mockedRuntimeFactory);
when(mockedWorkerService.getDlogNamespace()).thenReturn(mockedNamespace);
when(mockedWorkerService.isInitialized()).thenReturn(true);
+
when(mockedWorkerService.getBrokerAdmin()).thenReturn(mockedPulsarAdmin);
+ when(mockedPulsarAdmin.tenants()).thenReturn(mockedTenants);
+ when(mockedPulsarAdmin.namespaces()).thenReturn(mockedNamespaces);
+ when(mockedTenants.getTenantInfo(any())).thenReturn(mockedTenantInfo);
+ when(mockedNamespaces.getNamespaces(any())).thenReturn(namespaceList);
// worker config
WorkerConfig workerConfig = new WorkerConfig()
@@ -485,6 +505,22 @@ public void testRegisterFunctionSuccess() throws Exception
{
assertEquals(Status.OK.getStatusCode(), response.getStatus());
}
+ @Test
+ public void testRegisterFunctionNonexistantNamespace() throws Exception {
+ this.namespaceList.clear();
+ Response response = registerDefaultFunction();
+ assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+ assertEquals(new ErrorData("Namespace does not exist").reason,
((ErrorData) response.getEntity()).reason);
+ }
+
+ @Test
+ public void testRegisterFunctionNonexistantTenant() throws Exception {
+
when(mockedTenants.getTenantInfo(any())).thenThrow(PulsarAdminException.NotFoundException.class);
+ Response response = registerDefaultFunction();
+ assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+ assertEquals(new ErrorData("Tenant does not exist").reason,
((ErrorData) response.getEntity()).reason);
+ }
+
@Test
public void testRegisterFunctionFailure() throws Exception {
mockStatic(Utils.class);
@@ -1207,6 +1243,7 @@ public void testRegisterFunctionWithConflictingFields()
throws IOException {
String actualTenant = "DIFFERENT_TENANT";
String actualNamespace = "DIFFERENT_NAMESPACE";
String actualName = "DIFFERENT_NAME";
+ this.namespaceList.add(actualTenant + "/" + actualNamespace);
String fileLocation =
FutureUtil.class.getProtectionDomain().getCodeSource().getLocation().getPath();
String filePackageUrl = "file://" + fileLocation;
diff --git
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2ResourceTest.java
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2ResourceTest.java
index 93a0a747c7..5e710e03d5 100644
---
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2ResourceTest.java
+++
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SinkApiV2ResourceTest.java
@@ -24,7 +24,12 @@
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.core.config.Configurator;
+import org.apache.pulsar.client.admin.Namespaces;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.common.policies.data.ErrorData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.functions.api.utils.IdentityFunction;
import org.apache.pulsar.functions.proto.Function;
@@ -102,6 +107,11 @@ public IObjectFactory getObjectFactory() {
private String INVALID_JAR_FILE_PATH;
private WorkerService mockedWorkerService;
+ private PulsarAdmin mockedPulsarAdmin;
+ private Tenants mockedTenants;
+ private Namespaces mockedNamespaces;
+ private TenantInfo mockedTenantInfo;
+ private List<String> namespaceList = new LinkedList<>();
private FunctionMetaDataManager mockedManager;
private FunctionRuntimeManager mockedFunctionRunTimeManager;
private RuntimeFactory mockedRuntimeFactory;
@@ -119,6 +129,11 @@ public void setup() throws Exception {
this.mockedNamespace = mock(Namespace.class);
this.mockedFormData = mock(FormDataContentDisposition.class);
when(mockedFormData.getFileName()).thenReturn("test");
+ this.mockedTenantInfo = mock(TenantInfo.class);
+ this.mockedPulsarAdmin = mock(PulsarAdmin.class);
+ this.mockedTenants = mock(Tenants.class);
+ this.mockedNamespaces = mock(Namespaces.class);
+ namespaceList.add(tenant + "/" + namespace);
this.mockedWorkerService = mock(WorkerService.class);
when(mockedWorkerService.getFunctionMetaDataManager()).thenReturn(mockedManager);
@@ -126,6 +141,11 @@ public void setup() throws Exception {
when(mockedFunctionRunTimeManager.getRuntimeFactory()).thenReturn(mockedRuntimeFactory);
when(mockedWorkerService.getDlogNamespace()).thenReturn(mockedNamespace);
when(mockedWorkerService.isInitialized()).thenReturn(true);
+
when(mockedWorkerService.getBrokerAdmin()).thenReturn(mockedPulsarAdmin);
+ when(mockedPulsarAdmin.tenants()).thenReturn(mockedTenants);
+ when(mockedPulsarAdmin.namespaces()).thenReturn(mockedNamespaces);
+ when(mockedTenants.getTenantInfo(any())).thenReturn(mockedTenantInfo);
+ when(mockedNamespaces.getNamespaces(any())).thenReturn(namespaceList);
URL file =
Thread.currentThread().getContextClassLoader().getResource(JAR_FILE_NAME);
if (file == null) {
@@ -431,6 +451,7 @@ public void testRegisterSinkConflictingFields() throws
Exception {
String actualTenant = "DIFFERENT_TENANT";
String actualNamespace = "DIFFERENT_NAMESPACE";
String actualName = "DIFFERENT_NAME";
+ this.namespaceList.add(actualTenant + "/" + actualNamespace);
when(mockedManager.containsFunction(eq(tenant), eq(namespace),
eq(sink))).thenReturn(true);
when(mockedManager.containsFunction(eq(actualTenant),
eq(actualNamespace), eq(actualName))).thenReturn(false);
@@ -1064,4 +1085,20 @@ public void testOnlyGetSinks() throws Exception {
assertEquals(Status.OK.getStatusCode(), response.getStatus());
assertEquals(new Gson().toJson(functions), response.getEntity());
}
+
+ @Test
+ public void testRegisterFunctionNonexistantNamespace() throws Exception {
+ this.namespaceList.clear();
+ Response response = registerDefaultSink();
+ assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+ assertEquals(new ErrorData("Namespace does not exist").reason,
((ErrorData) response.getEntity()).reason);
+ }
+
+ @Test
+ public void testRegisterFunctionNonexistantTenant() throws Exception {
+
when(mockedTenants.getTenantInfo(any())).thenThrow(PulsarAdminException.NotFoundException.class);
+ Response response = registerDefaultSink();
+ assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+ assertEquals(new ErrorData("Tenant does not exist").reason,
((ErrorData) response.getEntity()).reason);
+ }
}
diff --git
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2ResourceTest.java
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2ResourceTest.java
index b7e5d130a0..0c420a3e2f 100644
---
a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2ResourceTest.java
+++
b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/api/v2/SourceApiV2ResourceTest.java
@@ -24,7 +24,12 @@
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.core.config.Configurator;
+import org.apache.pulsar.client.admin.Namespaces;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.common.policies.data.ErrorData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.functions.api.utils.IdentityFunction;
import org.apache.pulsar.functions.proto.Function.*;
@@ -90,6 +95,11 @@ public IObjectFactory getObjectFactory() {
private String INVALID_JAR_FILE_PATH;
private WorkerService mockedWorkerService;
+ private PulsarAdmin mockedPulsarAdmin;
+ private Tenants mockedTenants;
+ private Namespaces mockedNamespaces;
+ private TenantInfo mockedTenantInfo;
+ private List<String> namespaceList = new LinkedList<>();
private FunctionMetaDataManager mockedManager;
private FunctionRuntimeManager mockedFunctionRunTimeManager;
private RuntimeFactory mockedRuntimeFactory;
@@ -107,6 +117,11 @@ public void setup() throws Exception {
this.mockedNamespace = mock(Namespace.class);
this.mockedFormData = mock(FormDataContentDisposition.class);
when(mockedFormData.getFileName()).thenReturn("test");
+ this.mockedTenantInfo = mock(TenantInfo.class);
+ this.mockedPulsarAdmin = mock(PulsarAdmin.class);
+ this.mockedTenants = mock(Tenants.class);
+ this.mockedNamespaces = mock(Namespaces.class);
+ namespaceList.add(tenant + "/" + namespace);
this.mockedWorkerService = mock(WorkerService.class);
when(mockedWorkerService.getFunctionMetaDataManager()).thenReturn(mockedManager);
@@ -114,6 +129,11 @@ public void setup() throws Exception {
when(mockedFunctionRunTimeManager.getRuntimeFactory()).thenReturn(mockedRuntimeFactory);
when(mockedWorkerService.getDlogNamespace()).thenReturn(mockedNamespace);
when(mockedWorkerService.isInitialized()).thenReturn(true);
+
when(mockedWorkerService.getBrokerAdmin()).thenReturn(mockedPulsarAdmin);
+ when(mockedPulsarAdmin.tenants()).thenReturn(mockedTenants);
+ when(mockedPulsarAdmin.namespaces()).thenReturn(mockedNamespaces);
+ when(mockedTenants.getTenantInfo(any())).thenReturn(mockedTenantInfo);
+ when(mockedNamespaces.getNamespaces(any())).thenReturn(namespaceList);
URL file =
Thread.currentThread().getContextClassLoader().getResource(JAR_FILE_NAME);
if (file == null) {
@@ -402,6 +422,7 @@ public void testRegisterSourceConflictingFields() throws
Exception {
String actualTenant = "DIFFERENT_TENANT";
String actualNamespace = "DIFFERENT_NAMESPACE";
String actualName = "DIFFERENT_NAME";
+ this.namespaceList.add(actualTenant + "/" + actualNamespace);
when(mockedManager.containsFunction(eq(tenant), eq(namespace),
eq(source))).thenReturn(true);
when(mockedManager.containsFunction(eq(actualTenant),
eq(actualNamespace), eq(actualName))).thenReturn(false);
@@ -1086,4 +1107,20 @@ public void testOnlyGetSources() throws Exception {
assertEquals(Status.OK.getStatusCode(), response.getStatus());
assertEquals(new Gson().toJson(functions), response.getEntity());
}
+
+ @Test
+ public void testRegisterFunctionNonexistantNamespace() throws Exception {
+ this.namespaceList.clear();
+ Response response = registerDefaultSource();
+ assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+ assertEquals(new ErrorData("Namespace does not exist").reason,
((ErrorData) response.getEntity()).reason);
+ }
+
+ @Test
+ public void testRegisterFunctionNonexistantTenant() throws Exception {
+
when(mockedTenants.getTenantInfo(any())).thenThrow(PulsarAdminException.NotFoundException.class);
+ Response response = registerDefaultSource();
+ assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+ assertEquals(new ErrorData("Tenant does not exist").reason,
((ErrorData) response.getEntity()).reason);
+ }
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
