Add integration tests for package management service

---

Master Issue: #8676

*Motivation*

Add integration tests for testing the whole package management service.

*Modifications*

- Add integration tests
- Fix some issue when startup the package management service

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -1350,7 +1350,7 @@ private void startPackagesManagementService() throws IOException {
             .newProvider(config.getPackagesManagementStorageProvider());
         DefaultPackagesStorageConfiguration storageConfiguration = new DefaultPackagesStorageConfiguration();
         storageConfiguration.setProperty(config.getProperties());
-        PackagesStorage storage = storageProvider.getStorage(new DefaultPackagesStorageConfiguration());
+        PackagesStorage storage = storageProvider.getStorage(storageConfiguration);
         storage.initialize();
         packagesManagement.initialize(storage);
     }

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdPackages.java

@@ -36,7 +36,7 @@ class CmdPackages extends CmdBase {
 
     private final Packages packages;
 
-    CmdPackages(PulsarAdmin admin) {
+    public CmdPackages(PulsarAdmin admin) {
         super("packages", admin);
         this.packages = admin.packages();
 
@@ -133,7 +133,7 @@ private class ListPackageVersionsCmd extends CliCommand {
 
         @Override
         void run() throws Exception {
-            print(packages.listPackageVersions(packageName).toString());
+            print(packages.listPackageVersions(packageName));
         }
     }
 

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/PulsarAdminTool.java

@@ -142,6 +142,8 @@ public class PulsarAdminTool {
         // TODO eventually remove this
         commandMap.put("source", CmdSources.class);
         commandMap.put("sink", CmdSinks.class);
+
+        commandMap.put("packages", CmdPackages.class);
     }
 
     private void setupCommands(Function<PulsarAdminBuilder, ? extends PulsarAdmin> adminFactory) {

pulsar-package-management/bookkeeper-storage/src/main/java/org/apache/pulsar/packages/management/storage/bookkeeper/BookKeeperPackagesStorageConfiguration.java

@@ -35,15 +35,15 @@ public class BookKeeperPackagesStorageConfiguration implements PackagesStorageCo
     }
 
     int getNumReplicas() {
-        return Integer.parseInt(getProperty("numReplicas"));
+        return Integer.parseInt(getProperty("packagesReplicas"));
     }
 
     String getZkServers() {
-        return getProperty("zkServers");
+        return getProperty("zookeeperServers");
     }
 
     String getLedgersRootPath() {
-        return getProperty("ledgerRootPath");
+        return getProperty("packagesManagementLedgerRootPath");
     }
 
     String getBookkeeperClientAuthenticationPlugin() {

tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/PackagesCliTest.java

@@ -0,0 +1,133 @@
+/**
+ * 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.tests.integration.cli;
+
+import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
+import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
+import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
+import org.testcontainers.shaded.org.apache.commons.lang.RandomStringUtils;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Ignore;
+import org.testng.annotations.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class PackagesCliTest {
+
+    private final static String clusterNamePrefix = "packages-service";
+    private PulsarCluster pulsarCluster;
+
+    @BeforeClass
+    public void setup() throws Exception {
+        PulsarClusterSpec spec = PulsarClusterSpec.builder()
+            .clusterName(String.format("%s-%s", clusterNamePrefix, RandomStringUtils.randomAlphabetic(6)))
+            .brokerEnvs(getPackagesManagementServiceEnvs())
+            .build();
+        pulsarCluster = PulsarCluster.forSpec(spec);
+        pulsarCluster.start();
+    }
+
+    @AfterClass
+    public void teardown() {
+        if (pulsarCluster != null) {
+            pulsarCluster.stop();
+            pulsarCluster = null;
+        }
+    }
+
+    private Map<String, String> getPackagesManagementServiceEnvs() {
+        Map<String, String> envs = new HashMap<>();
+        envs.put("enablePackagesManagement", "true");
+        return envs;
+    }
+
+    @Test
+    public void testPackagesOperationsWithoutUploadingPackages() throws Exception {
+        ContainerExecResult result = runPackagesCommand("list", "--type", "function", "public/default");
+        assertEquals(result.getExitCode(), 0);
+        assertTrue(result.getStdout().isEmpty());
+
+        result = runPackagesCommand("list-versions", "function://public/default/test");
+        assertEquals(result.getExitCode(), 0);
+        assertTrue(result.getStdout().isEmpty());
+
+        result = runPackagesCommand("download", "function://public/default/test@v1");
+        assertNotEquals(result.getExitCode(), 0);
+        assertFalse(result.getStderr().isEmpty());
+        assertTrue(result.getStderr().contains("does not exist"));
+    }
+
+    // TODO: the upload command has some problem when uploading packages, enable this tests when issue
+    // https://github.com/apache/pulsar/issues/8874 is fixed.
+    @Ignore
+    @Test
+    public void testPackagesOperationsWithUploadingPackages() throws Exception {
+        String testPackageName = "function://public/default/test@v1";
+        ContainerExecResult result = runPackagesCommand("upload", "--description", "a test package",
+            "--path", PulsarCluster.ADMIN_SCRIPT, testPackageName);
+        assertEquals(result.getExitCode(), 0);
+
+        result = runPackagesCommand("get-metadata", testPackageName);
+        assertEquals(result.getExitCode(), 0);
+        assertFalse(result.getStdout().isEmpty());
+        assertTrue(result.getStdout().contains("a test package"));
+
+        result = runPackagesCommand("list", "--type", "function", "public/default");
+        assertEquals(result.getExitCode(), 0);
+        assertFalse(result.getStdout().isEmpty());
+        assertTrue(result.getStdout().contains(testPackageName));
+
+        result = runPackagesCommand("list-versions", "--type", "function://public/default/test");
+        assertEquals(result.getExitCode(), 0);
+        assertFalse(result.getStdout().isEmpty());
+        assertTrue(result.getStdout().contains("v1"));
+
+        String contact = "test@apache.org";
+        result = runPackagesCommand("update-metadata", "--contact", contact, "-PpropertyA=A", testPackageName);
+        assertEquals(result.getExitCode(), 0);
+
+        result = runPackagesCommand("get-metadata", testPackageName);
+        assertEquals(result.getExitCode(), 0);
+        assertFalse(result.getStdout().isEmpty());
+        assertTrue(result.getStdout().contains("a test package"));
+        assertTrue(result.getStdout().contains(contact));
+        assertTrue(result.getStdout().contains("propertyA"));
+
+        result = runPackagesCommand("delete", testPackageName);
+        assertEquals(result.getExitCode(), 0);
+
+        result = runPackagesCommand("list", "--type", "functions", "public/default");
+        assertEquals(result.getExitCode(), 0);
+        assertTrue(result.getStdout().isEmpty());
+    }
+
+    private ContainerExecResult runPackagesCommand(String... commands) throws Exception {
+        String[] cmds = new String[commands.length + 1];
+        cmds[0] = "packages";
+        System.arraycopy(commands, 0, cmds, 1, commands.length);
+        return pulsarCluster.runAdminCommandOnAnyBroker(cmds);
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java

@@ -171,6 +171,9 @@ private PulsarCluster(PulsarClusterSpec spec) {
                         brokerContainer.withEnv("bookkeeperExplicitLacIntervalInMills", "10");
                         brokerContainer.withEnv("bookkeeperUseV2WireProtocol", "false");
                     }
+                    if (spec.brokerEnvs != null) {
+                        brokerContainer.withEnv(spec.brokerEnvs);
+                    }
                     return brokerContainer;
                 }
             ));

tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java

@@ -137,4 +137,9 @@ public class PulsarClusterSpec {
      * Specify envs for proxy.
      */
     Map<String, String> proxyEnvs;
+
+    /**
+     * Specify envs for broker.
+     */
+    Map<String, String> brokerEnvs;
 }