Support infer type class name

.ci/tests/integration/cases/elasticsearch-sink-inferring-type/elasticsearch.yaml

@@ -0,0 +1,18 @@
+apiVersion: elasticsearch.k8s.elastic.co/v1
+kind: Elasticsearch
+metadata:
+  name: quickstart
+spec:
+  version: 7.9.2
+  http:
+    tls:
+      selfSignedCertificate:
+        disabled: true
+  nodeSets:
+    - name: default
+      count: 1
+      config:
+        node.master: true
+        node.data: true
+        node.ingest: true
+        node.store.allow_mmap: false

.ci/tests/integration/cases/elasticsearch-sink-inferring-type/manifests.yaml

@@ -0,0 +1,44 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Sink
+metadata:
+  name: sink-sample-inferring-type
+spec:
+  className: org.apache.pulsar.io.elasticsearch.ElasticSearchSink
+  replicas: 1
+  maxReplicas: 1
+  input:
+    topics:
+      - persistent://public/default/input-sink-topic
+  sinkConfig:
+    elasticSearchUrl: "http://quickstart-es-http.default.svc.cluster.local:9200"
+    username: "elastic"
+    password: "QqB1OtT6m79vfP7H9H0q2a82"
+  pulsar:
+    pulsarConfig: "test-sink"
+    tlsConfig:
+      enabled: false
+      allowInsecure: false
+      hostnameVerification: true
+      certSecretName: sn-platform-tls-broker
+      certSecretKey: ""
+  resources:
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+    requests:
+      cpu: "0.1"
+      memory: 1G
+  image: streamnative/pulsar-io-elastic-search:2.9.3.16
+  java:
+    jar: connectors/pulsar-io-elastic-search-2.9.3.16.nar
+    jarLocation: "" # use pulsar provided connectors
+  clusterName: test-pulsar
+  autoAck: true
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-sink
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650

.ci/tests/integration/cases/elasticsearch-sink-inferring-type/verify.sh

@@ -0,0 +1,100 @@
+#!/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
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"}
+PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"}
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/elasticsearch-sink-inferring-type/manifests.yaml
+es_file="${BASE_DIR}"/.ci/tests/integration/cases/elasticsearch-sink-inferring-type/elasticsearch.yaml
+
+function install_elasticsearch_cluster() {
+    # install eck operator
+    kubectl create -f https://download.elastic.co/downloads/eck/2.3.0/crds.yaml
+    kubectl apply -f https://download.elastic.co/downloads/eck/2.3.0/operator.yaml
+    num=0
+    while [[ ${num} -lt 1 ]]; do
+        sleep 5
+        kubectl get pods -n elastic-system
+        num=$(kubectl get pods -n elastic-system -l control-plane=elastic-operator | wc -l)
+    done
+    kubectl wait -n elastic-system -l control-plane=elastic-operator --for=condition=Ready pod --timeout=5m && true
+
+    # install es cluster
+    kubectl apply -f "${es_file}"
+    num=0
+    while [[ ${num} -lt 1 ]]; do
+        sleep 5
+        kubectl get pods
+        num=$(kubectl get pods -l elasticsearch.k8s.elastic.co/cluster-name=quickstart | wc -l)
+    done
+    kubectl wait -l elasticsearch.k8s.elastic.co/cluster-name=quickstart --for=condition=Ready pod --timeout=5m && true
+}
+
+function uninstall_elasticsearch_cluster() {
+    kubectl delete elasticsearches.elasticsearch.k8s.elastic.co quickstart
+    while true; do
+        kubectl get elasticsearches.elasticsearch.k8s.elastic.co quickstart > /dev/null 2>&1
+        if [ $? -eq 1 ]; then
+            break
+        fi
+        sleep 5
+    done
+
+    kubectl delete -f https://download.elastic.co/downloads/eck/2.3.0/operator.yaml
+    kubectl delete -f https://download.elastic.co/downloads/eck/2.3.0/crds.yaml
+}
+
+# setup es cluster
+setup_es_result=$(install_elasticsearch_cluster 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$setup_es_result"
+  uninstall_elasticsearch_cluster > /dev/null 2>&1
+fi
+
+password=$(kubectl get secret quickstart-es-elastic-user -o go-template='{{.data.elastic | base64decode}}')
+sed -i.bak "s/password: \(.*\)/password: ${password}/g" "${manifests_file}"
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh sink-sample-inferring-type 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_sink_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_sink 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_sink_result"
+fi
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+uninstall_elasticsearch_cluster > /dev/null 2>&1 || true

.ci/tests/integration/cases/java-function-inferring-type/manifests.yaml

@@ -0,0 +1,75 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: function-sample-inferring-type
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-java-sample:2.9.3.16
+  className: org.apache.pulsar.functions.api.examples.ExclamationFunction
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  maxReplicas: 5
+  logTopic: persistent://public/default/logging-function-logs
+  input:
+    topics:
+    - persistent://public/default/input-java-topic
+  output:
+    topic: persistent://public/default/output-java-topic
+  resources:
+    requests:
+      cpu: 50m
+      memory: 1G
+    limits:
+      memory: 1.1G
+  # each secret will be loaded ad an env variable from the `path` secret with the `key` in that secret in the name of `name`
+  secretsMap:
+    "name":
+        path: "test-secret"
+        key: "username"
+    "pwd":
+        path: "test-secret"
+        key: "password"
+  pulsar:
+    pulsarConfig: "test-pulsar"
+    tlsConfig:
+      enabled: false
+      allowInsecure: false
+      hostnameVerification: true
+      certSecretName: sn-platform-tls-broker
+      certSecretKey: ""
+    #authConfig: "test-auth"
+  java:
+    jar: /pulsar/examples/api-examples.jar
+  # to be delete & use admission hook
+  clusterName: test
+  autoAck: true
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-pulsar
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650
+#---
+#apiVersion: v1
+#kind: ConfigMap
+#metadata:
+#  name: test-auth
+#data:
+#  clientAuthenticationPlugin: "abc"
+#  clientAuthenticationParameters: "xyz"
+#  tlsTrustCertsFilePath: "uvw"
+#  useTls: "true"
+#  tlsAllowInsecureConnection: "false"
+#  tlsHostnameVerificationEnable: "true"
+---
+apiVersion: v1
+data:
+  username: YWRtaW4=
+  password: MWYyZDFlMmU2N2Rm
+kind: Secret
+metadata:
+  name: test-secret
+type: Opaque

.ci/tests/integration/cases/java-function-inferring-type/verify.sh

@@ -0,0 +1,52 @@
+#!/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
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"}
+PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"}
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/java-function-inferring-type/manifests.yaml
+
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh function-sample-inferring-type 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_java_function 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_java_result"
+fi
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true

.ci/tests/integration/cases/mongodb-source-inferring-type/manifests.yaml

@@ -0,0 +1,53 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Source
+metadata:
+  name: source-sample-inferring-type
+spec:
+  className: org.apache.pulsar.io.debezium.mongodb.DebeziumMongoDbSource
+  replicas: 1
+  maxReplicas: 1
+  output:
+    producerConf:
+      maxPendingMessages: 1000
+      maxPendingMessagesAcrossPartitions: 50000
+      useThreadLocalProducers: true
+    topic: persistent://public/default/output-source-topic
+  forwardSourceMessageProperty: true
+  resources:
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+    requests:
+      cpu: "0.1"
+      memory: 1G
+  sourceConfig:
+    mongodb.hosts: rs0/mongo-dbz-0.mongo.default.svc.cluster.local:27017,rs0/mongo-dbz-1.mongo.default.svc.cluster.local:27017,rs0/mongo-dbz-2.mongo.default.svc.cluster.local:27017
+    mongodb.name: dbserver1
+    mongodb.user: debezium
+    mongodb.password: dbz
+    mongodb.task.id: "1"
+    database.whitelist: inventory
+    pulsar.service.url: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650
+  pulsar:
+    pulsarConfig: "test-source"
+    tlsConfig:
+      enabled: false
+      allowInsecure: false
+      hostnameVerification: true
+      certSecretName: sn-platform-tls-broker
+      certSecretKey: ""
+  image: streamnative/pulsar-io-debezium-mongodb:2.9.3.16
+  java:
+    jar: connectors/pulsar-io-debezium-mongodb-2.9.3.16.nar
+    jarLocation: "" # use pulsar provided connectors
+    # use package name:
+    # jarLocation: function://public/default/nul-test-java-source@v1
+  clusterName: test-pulsar
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-source
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650