Skip to content

Commit

Permalink
[package management service] check service status before run commands (
Browse files Browse the repository at this point in the history
…apache#12847)

### Motivation

`pulsar-admin` runs packages management services commands, but if the broker is not enabled the service, it will throw NPE. 

The root cause is https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PackagesBase.java#L45-L47, the `PackagesManagement` might be null if the service is not enabled.

### Modifications

Check `isEnablePackagesManagement` before each internal request.
  • Loading branch information
freeznet authored Nov 19, 2021
1 parent 1646be2 commit 747bf21
Show file tree
Hide file tree
Showing 5 changed files with 121 additions and 10 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -244,7 +244,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
private AdditionalServlets brokerAdditionalServlets;

// packages management service
private PackagesManagement packagesManagement;
private Optional<PackagesManagement> packagesManagement = Optional.empty();
private PrometheusMetricsServlet metricsServlet;
private List<PrometheusRawMetricsProvider> pendingMetricsProviders;

Expand Down Expand Up @@ -1544,16 +1544,22 @@ private void startWorkerService(AuthenticationService authenticationService,
}
}

public PackagesManagement getPackagesManagement() throws UnsupportedOperationException {
return packagesManagement.orElseThrow(() -> new UnsupportedOperationException("Package Management Service "
+ "is not enabled in the broker."));
}

private void startPackagesManagementService() throws IOException {
// TODO: using provider to initialize the packages management service.
this.packagesManagement = new PackagesManagementImpl();
PackagesManagement packagesManagementService = new PackagesManagementImpl();
this.packagesManagement = Optional.of(packagesManagementService);
PackagesStorageProvider storageProvider = PackagesStorageProvider
.newProvider(config.getPackagesManagementStorageProvider());
DefaultPackagesStorageConfiguration storageConfiguration = new DefaultPackagesStorageConfiguration();
storageConfiguration.setProperty(config.getProperties());
PackagesStorage storage = storageProvider.getStorage(storageConfiguration);
storage.initialize();
packagesManagement.initialize(storage);
packagesManagementService.initialize(storage);
}

public Optional<Integer> getListenPortHTTP() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ private Void handleError(Throwable throwable, AsyncResponse asyncResponse) {
asyncResponse.resume(new RestException(Response.Status.NOT_FOUND, throwable.getMessage()));
} else if (throwable instanceof WebApplicationException) {
asyncResponse.resume(throwable);
} else if (throwable instanceof UnsupportedOperationException) {
asyncResponse.resume(new RestException(Response.Status.SERVICE_UNAVAILABLE, throwable.getMessage()));
} else {
log.error("Encountered unexpected error", throwable);
asyncResponse.resume(new RestException(Response.Status.INTERNAL_SERVER_ERROR, throwable.getMessage()));
Expand Down Expand Up @@ -116,6 +118,8 @@ protected StreamingOutput internalDownload(String type, String tenant, String na
} else {
throw new RestException(Response.Status.INTERNAL_SERVER_ERROR, e.getCause().getMessage());
}
} catch (UnsupportedOperationException e) {
throw new RestException(Response.Status.SERVICE_UNAVAILABLE, e.getMessage());
}
};
} catch (IllegalArgumentException illegalArgumentException) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@ public class Packages extends PackagesBase {
@ApiResponse(code = 200, message = "Return the metadata of the specified package."),
@ApiResponse(code = 404, message = "The specified package is not existent."),
@ApiResponse(code = 412, message = "The package name is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
public void getMeta(
Expand All @@ -82,7 +83,8 @@ public void getMeta(
@ApiResponse(code = 200, message = "Update the metadata of the specified package successfully."),
@ApiResponse(code = 404, message = "The specified package is not existent."),
@ApiResponse(code = 412, message = "The package name is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
@Consumes(MediaType.APPLICATION_JSON)
Expand Down Expand Up @@ -113,7 +115,8 @@ public void updateMeta(
value = {
@ApiResponse(code = 200, message = "Upload the specified package successfully."),
@ApiResponse(code = 412, message = "The package name is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
@Consumes(MediaType.MULTIPART_FORM_DATA)
Expand Down Expand Up @@ -148,7 +151,8 @@ public void upload(
@ApiResponse(code = 200, message = "Download the specified package successfully."),
@ApiResponse(code = 404, message = "The specified package is not existent."),
@ApiResponse(code = 412, message = "The package name is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
public StreamingOutput download(
Expand All @@ -168,7 +172,8 @@ public StreamingOutput download(
@ApiResponse(code = 200, message = "Delete the specified package successfully."),
@ApiResponse(code = 404, message = "The specified package is not existent."),
@ApiResponse(code = 412, message = "The package name is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
@ApiOperation(value = "Delete a package with the package name.")
Expand All @@ -195,7 +200,8 @@ public void delete(
@ApiResponse(code = 200, message = "Return the package versions of the specified package."),
@ApiResponse(code = 404, message = "The specified package is not existent."),
@ApiResponse(code = 412, message = "The package name is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
public void listPackageVersion(
Expand All @@ -219,7 +225,8 @@ public void listPackageVersion(
@ApiResponse(code = 200, message =
"Return all the specified type package names in the specified namespace."),
@ApiResponse(code = 412, message = "The package type is illegal."),
@ApiResponse(code = 500, message = "Internal server error.")
@ApiResponse(code = 500, message = "Internal server error."),
@ApiResponse(code = 503, message = "Package Management Service is not enabled in the broker.")
}
)
public void listPackages(
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
/**
* 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 static org.testng.Assert.assertEquals;
import static org.testng.Assert.fail;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.packages.management.core.common.PackageMetadata;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

@Test(groups = "broker")
public class PackagesApiNotEnabledTest extends MockedPulsarServiceBaseTest {

@BeforeMethod
@Override
protected void setup() throws Exception {
// not enable Package Management Service
conf.setEnablePackagesManagement(false);
super.internalSetup();
}

@AfterMethod(alwaysRun = true)
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
}

@Test(timeOut = 60000)
public void testPackagesOperationsWithoutPackagesServiceEnabled() {
// download package api should return 503 Service Unavailable exception
String unknownPackageName = "function://public/default/unknown@v1";
try {
admin.packages().download(unknownPackageName, "/test/unknown");
fail("should throw 503 error");
} catch (PulsarAdminException e) {
assertEquals(503, e.getStatusCode());
}

// get metadata api should return 503 Service Unavailable exception
try {
admin.packages().getMetadata(unknownPackageName);
fail("should throw 503 error");
} catch (PulsarAdminException e) {
assertEquals(503, e.getStatusCode());
}

// update metadata api should return 503 Service Unavailable exception
try {
admin.packages().updateMetadata(unknownPackageName,
PackageMetadata.builder().description("unknown").build());
fail("should throw 503 error");
} catch (PulsarAdminException e) {
assertEquals(503, e.getStatusCode());
}

// list all the packages api should return 503 Service Unavailable exception
try {
admin.packages().listPackages("function", "unknown/unknown");
fail("should throw 503 error");
} catch (PulsarAdminException e) {
assertEquals(503, e.getStatusCode());
}

// list all the versions api should return 503 Service Unavailable exception
try {
admin.packages().listPackageVersions(unknownPackageName);
fail("should throw 503 error");
} catch (PulsarAdminException e) {
assertEquals(503, e.getStatusCode());
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -106,13 +106,15 @@ public void testPackagesOperationsFailed() {
String unknownPackageName = "function://public/default/unknown@v1";
try {
admin.packages().download(unknownPackageName, "/test/unknown");
fail("should throw 404 error");
} catch (PulsarAdminException e) {
assertEquals(404, e.getStatusCode());
}

// get the metadata of a non-existent package should return not found exception
try {
admin.packages().getMetadata(unknownPackageName);
fail("should throw 404 error");
} catch (PulsarAdminException e) {
assertEquals(404, e.getStatusCode());
}
Expand All @@ -121,6 +123,7 @@ public void testPackagesOperationsFailed() {
try {
admin.packages().updateMetadata(unknownPackageName,
PackageMetadata.builder().description("unknown").build());
fail("should throw 404 error");
} catch (PulsarAdminException e) {
assertEquals(404, e.getStatusCode());
}
Expand Down

0 comments on commit 747bf21

Please sign in to comment.