add package management service tests

.ci/clusters/values_mesh_worker_service.yaml

@@ -113,6 +113,8 @@ broker:
     managedLedgerDefaultEnsembleSize: "1"
     managedLedgerDefaultWriteQuorum: "1"
     managedLedgerDefaultAckQuorum: "1"
+    enablePackagesManagement: "true"
+    PULSAR_PREFIX_enablePackagesManagement: "true"
 
 proxy:
   replicaCount: 1

.ci/helm.sh

@@ -342,3 +342,98 @@ function ci::verify_mesh_worker_service_pulsar_admin() {
   ${KUBECTL} get pods -n ${NAMESPACE}
 
 }
+
+function ci::upload_java_package() {
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin packages upload function://public/default/java-function@1.0 --path /pulsar/examples/api-examples.jar --description java-function@1.0)
+  if [[ $RET != *"successfully"* ]]; then
+    echo "${RET}"
+    return 1
+  fi
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin packages download function://public/default/java-function@1.0 --path /pulsar/api-examples.jar)
+  if [[ $RET != *"successfully"* ]]; then
+    echo "${RET}"
+    return 1
+  fi
+}
+
+function ci::verify_java_package() {
+  ${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions create --jar function://public/default/java-function@1.0 --name package-java-fn --className org.apache.pulsar.functions.api.examples.ExclamationFunction --inputs persistent://public/default/package-java-fn-input --cpu 0.1
+  sleep 15
+  ${KUBECTL} logs -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0
+  sleep 15
+  ${KUBECTL} get pods -A
+  sleep 5
+  WC=$(${KUBECTL} get pods -n ${NAMESPACE} --field-selector=status.phase=Running | grep "package-java-fn" | wc -l)
+  while [[ ${WC} -lt 1 ]]; do
+    echo ${WC};
+    sleep 20
+    ${KUBECTL} get pods -n ${NAMESPACE}
+    ${KUBECTL} describe pod package-java-fn-function-0
+    WC=$(${KUBECTL} get pods -n ${NAMESPACE} --field-selector=status.phase=Running | grep "package-java-fn" | wc -l)
+  done
+  echo "java function test done"
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions delete --name package-java-fn)
+  echo "${RET}"
+}
+
+function ci::upload_python_package() {
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin packages upload function://public/default/python-function@1.0 --path /pulsar/examples/python-examples/exclamation_function.py --description python-function@1.0)
+  if [[ $RET != *"successfully"* ]]; then
+    echo "${RET}"
+    return 1
+  fi
+}
+
+function ci::verify_python_package() {
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions create --py function://public/default/python-function@1.0 --name package-python-fn --classname exclamation_function.ExclamationFunction --inputs persistent://public/default/package-python-fn-input --cpu 0.1)
+  if [[ $RET != *"successfully"* ]]; then
+    echo "${RET}"
+    return 1
+  fi
+  sleep 15
+  ${KUBECTL} get pods -A
+  sleep 5
+  WC=$(${KUBECTL} get pods -n ${NAMESPACE} --field-selector=status.phase=Running | grep "package-python-fn" | wc -l)
+  while [[ ${WC} -lt 1 ]]; do
+    echo ${WC};
+    sleep 20
+    ${KUBECTL} get pods -n ${NAMESPACE}
+    ${KUBECTL} describe pod package-python-fn-function-0
+    WC=$(${KUBECTL} get pods -n ${NAMESPACE} --field-selector=status.phase=Running | grep "package-python-fn" | wc -l)
+  done
+  echo "python function test done"
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions delete --name package-python-fn)
+  echo "${RET}"
+}
+
+function ci::upload_go_package() {
+  ${KUBECTL} cp "${FUNCTION_MESH_HOME}/.ci/examples/go-examples" "${NAMESPACE}/${CLUSTER}-pulsar-broker-0:/pulsar/examples"
+  sleep 1
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin packages upload function://public/default/go-function@1.0 --path /pulsar/examples/go-examples --description go-function@1.0)
+  if [[ $RET != *"successfully"* ]]; then
+    echo "${RET}"
+    return 1
+  fi
+}
+
+function ci::verify_go_package() {
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions create --go function://public/default/go-function@1.0 --name package-go-fn --inputs persistent://public/default/package-go-fn-input --cpu 0.1)
+  if [[ $RET != *"successfully"* ]]; then
+    echo "${RET}"
+    return 1
+  fi
+  sleep 15
+  ${KUBECTL} get pods -A
+  sleep 5
+  WC=$(${KUBECTL} get pods -n ${NAMESPACE} --field-selector=status.phase=Running | grep "package-go-fn" | wc -l)
+  while [[ ${WC} -lt 1 ]]; do
+    echo ${WC};
+    sleep 20
+    ${KUBECTL} get pods -n ${NAMESPACE}
+    ${KUBECTL} describe pod package-go-fn-function-0
+    WC=$(${KUBECTL} get pods -n ${NAMESPACE} --field-selector=status.phase=Running | grep "package-go-fn" | wc -l)
+  done
+  echo "go function test done"
+  RET=$(${KUBECTL} exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin functions delete --name package-go-fn)
+  echo "${RET}"
+}

.ci/verify_package_management_service.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+set -e
+
+BINDIR=`dirname "$0"`
+PULSAR_HOME=`cd ${BINDIR}/..;pwd`
+TLS=${TLS:-"false"}
+SYMMETRIC=${SYMMETRIC:-"false"}
+FUNCTION=${FUNCTION:-"false"}
+
+source ${PULSAR_HOME}/.ci/helm.sh
+
+ci::upload_java_package
+ci::verify_java_package
+
+#ci::upload_python_package
+#ci::verify_python_package
+
+#ci::upload_go_package
+#ci::verify_go_package

.github/workflows/test-integration-mesh-worker-service.yml

@@ -81,3 +81,7 @@ jobs:
       - name: verify mesh-worker-service
         run: |
           .ci/verify_mesh_worker_service.sh
+
+      - name: verify package management service
+        run: |
+          .ci/verify_package_management_service.sh

mesh-worker-service/src/main/java/io/functionmesh/compute/util/FunctionsUtil.java

@@ -35,6 +35,7 @@
 import io.functionmesh.compute.models.CustomRuntimeOptions;
 import io.kubernetes.client.custom.Quantity;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.util.Strings;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -267,7 +268,7 @@ public static V1alpha1Function createV1alpha1FunctionFromFunctionConfig(String k
         }
         Class<?>[] typeArgs = null;
         if (componentPackageFile != null) {
-            typeArgs = extractTypeArgs(functionConfig, componentPackageFile);
+            typeArgs = extractTypeArgs(functionConfig, componentPackageFile, worker.getWorkerConfig().isForwardSourceMessageProperty());
         }
         if (StringUtils.isNotEmpty(functionConfig.getJar())) {
             V1alpha1FunctionSpecJava v1alpha1FunctionSpecJava = new V1alpha1FunctionSpecJava();
@@ -564,14 +565,22 @@ private static File downloadPackageFile(MeshWorkerService worker, String package
             // use the Nar extraction directory as a temporary directory for downloaded files
             tempDirectory = Paths.get(worker.getWorkerConfig().getNarExtractionDirectory());
         }
-        File file = Files.createTempFile(tempDirectory, "function", ".tmp").toFile();
-        worker.getBrokerAdmin().packages().download(packageName, file.toString());
-        return file;
+        if (Files.notExists(tempDirectory)) {
+            Files.createDirectories(tempDirectory);
+        }
+        String fileName = String.format("function-%s.tmp", RandomStringUtils.random(5, true, true).toLowerCase());
+        Path filePath = Paths.get(tempDirectory.toString(), fileName);
+        Files.deleteIfExists(filePath);
+        worker.getBrokerAdmin().packages().download(packageName, filePath.toString());
+        return filePath.toFile();
     }
 
     private static Class<?>[] extractTypeArgs(final FunctionConfig functionConfig,
-                                             final File componentPackageFile) {
+                                             final File componentPackageFile,
+                                              final boolean isForwardSourceMessageProperty) {
         Class<?>[] typeArgs = null;
+        FunctionConfigUtils.inferMissingArguments(
+                functionConfig, isForwardSourceMessageProperty);
         if (componentPackageFile == null) {
             return null;
         }